Decorators¶
The method decorators detailed in this section describe request properties that are relevant to all invocations of a consumer method.
headers¶
-
class
uplink.
headers
(arg=None, **kwargs)[source]¶ A decorator that adds static headers for API calls.
@headers({"User-Agent": "Uplink-Sample-App"}) @get("/user") def get_user(self): """Get the current user"""
When used as a class decorator,
headers
applies to all consumer methods bound to the class:@headers({"Accept": "application/vnd.github.v3.full+json"}) class GitHub(Consumer): ...
headers
takes the same arguments asdict
.Parameters: - arg – A dict containing header values.
- **kwargs – More header values.
params¶
-
class
uplink.
params
(arg=None, **kwargs)[source]¶ A decorator that adds static query parameters for API calls.
@params({"sort": "created"}) @get("/user") def get_user(self): """Get the current user"""
When used as a class decorator,
params
applies to all consumer methods bound to the class:@params({"client_id": "my-app-client-id"}) class GitHub(Consumer): ...
params
takes the same arguments asdict
.Parameters: - arg – A dict containing query parameters.
- **kwargs – More query parameters.
json¶
-
class
uplink.
json
[source]¶ Use as a decorator to make JSON requests.
You can annotate a method argument with
uplink.Body
, which indicates that the argument’s value should become the request’s body.uplink.Body
has to be either a dict or a subclass of py:class:abc.Mapping.Example
@json @patch(/user") def update_user(self, **info: Body): """Update the current user."""
You can alternatively use the
uplink.Field
annotation to specify JSON fields separately, across multiple arguments:Example
@json @patch(/user") def update_user(self, name: Field, email: Field("e-mail")): """Update the current user."""
Further, to set a nested field, you can specify the path of the target field with a tuple of strings as the first argument of
uplink.Field
.Example
Consider a consumer method that sends a PATCH request with a JSON body of the following format:
{ user: { name: "<User's Name>" }, }
The tuple
("user", "name")
specifies the path to the highlighted inner field:@json @patch(/user") def update_user( self, new_name: Field(("user", "name")) ): """Update the current user."""
form_url_encoded¶
-
class
uplink.
form_url_encoded
[source]¶ URL-encodes the request body.
Used on POST/PUT/PATCH request. It url-encodes the body of the message and sets the appropriate
Content-Type
header. Further, each field argument should be annotated withuplink.Field
.Example
@form_url_encoded @post("/users/edit") def update_user(self, first_name: Field, last_name: Field): """Update the current user."""
multipart¶
-
class
uplink.
multipart
[source]¶ Sends multipart form data.
Multipart requests are commonly used to upload files to a server. Further, annotate each part argument with
Part
.Example
@multipart @put(/user/photo") def update_user(self, photo: Part, description: Part): """Upload a user profile photo."""
timeout¶
-
class
uplink.
timeout
(seconds)[source]¶ Time to wait for a server response before giving up.
When used on other decorators it specifies how long (in secs) a decorator should wait before giving up.
Example
@timeout(60) @get("/user/posts") def get_posts(self): """Fetch all posts for the current users."""
When used as a class decorator,
timeout
applies to all consumer methods bound to the class.Parameters: seconds (int) – An integer used to indicate how long should the request wait.
args¶
-
class
uplink.
args
(*annotations, **more_annotations)[source]¶ Annotate method arguments for Python 2.7 compatibility.
Arrange annotations in the same order as their corresponding function arguments.
Example
@args(Path, Query) @get("/users/{username}) def get_user(self, username, visibility): """Get a specific user."""
Use keyword args to target specific method parameters.
Example
@args(visibility=Query) @get("/users/{username}) def get_user(self, username, visibility): """Get a specific user."""
Parameters: - *annotations – Any number of annotations.
- **more_annotations – More annotations, targeting specific method arguments.
response_handler¶
-
class
uplink.
response_handler
(handler, requires_consumer=False)[source]¶ A decorator for creating custom response handlers.
To register a function as a custom response handler, decorate the function with this class. The decorated function should accept a single positional argument, an HTTP response object:
Example
@response_handler def raise_for_status(response): response.raise_for_status() return response
Then, to apply custom response handling to a request method, simply decorate the method with the registered response handler:
Example
@raise_for_status @get("/user/posts") def get_posts(self): """Fetch all posts for the current users."""
To apply custom response handling on all request methods of a
uplink.Consumer
subclass, simply decorate the class with the registered response handler:Example
@raise_for_status class GitHub(Consumer): ...
Lastly, the decorator supports the optional argument
requires_consumer
. When this option is set toTrue
, the registered callback should accept a reference to theConsumer
instance as its leading argument:Example
@response_handler(requires_consumer=True) def raise_for_status(consumer, response): ...
New in version 0.4.0.
error_handler¶
-
class
uplink.
error_handler
(exception_handler, requires_consumer=False)[source]¶ A decorator for creating custom error handlers.
To register a function as a custom error handler, decorate the function with this class. The decorated function should accept three positional arguments: (1) the type of the exception, (2) the exception instance raised, and (3) a traceback instance.
Example
@error_handler def raise_api_error(exc_type, exc_val, exc_tb): # wrap client error with custom API error ...
Then, to apply custom error handling to a request method, simply decorate the method with the registered error handler:
Example
@raise_api_error @get("/user/posts") def get_posts(self): """Fetch all posts for the current users."""
To apply custom error handling on all request methods of a
uplink.Consumer
subclass, simply decorate the class with the registered error handler:Example
@raise_api_error class GitHub(Consumer): ...
Lastly, the decorator supports the optional argument
requires_consumer
. When this option is set toTrue
, the registered callback should accept a reference to theConsumer
instance as its leading argument:Example
@error_handler(requires_consumer=True) def raise_api_error(consumer, exc_type, exc_val, exc_tb): ...
New in version 0.4.0.
Note
Error handlers can not completely suppress exceptions. The original exception is thrown if the error handler doesn’t throw anything.
inject¶
returns.*¶
Converting an HTTP response body into a custom Python object is
straightforward with Uplink; the uplink.returns
modules
exposes optional decorators for defining the expected return type and
data serialization format for any consumer method.
-
class
uplink.returns.
json
(type=None, key=(), model=None, member=())[source]¶ Specifies that the decorated consumer method should return a JSON object.
# This method will return a JSON object (e.g., a dict or list) @returns.json @get("/users/{username}") def get_user(self, username): """Get a specific user."""
Returning a Specific JSON Field:
The
key
argument accepts a string or tuple that specifies the path of an internal field in the JSON document.For instance, consider an API that returns JSON responses that, at the root of the document, contains both the server-retrieved data and a list of relevant API errors:
{ "data": { "user": "prkumar", "id": "140232" }, "errors": [] }
If returning the list of errors is unnecessary, we can use the
key
argument to strictly return the nested fielddata.id
:@returns.json(key=("data", "id")) @get("/users/{username}") def get_user_id(self, username): """Get a specific user's ID."""
We can also configure Uplink to convert the field before it’s returned by also specifying the``type`` argument:
@returns.json(key=("data", "id"), type=int) @get("/users/{username}") def get_user_id(self, username): """Get a specific user's ID."""
New in version v0.5.0.
-
uplink.returns.
from_json
¶ Specifies that the decorated consumer method should produce instances of a
type
class using a registered deserialization strategy (seeuplink.loads.from_json()
)This decorator accepts the same arguments as
uplink.returns.json
.Often, a JSON response body represents a schema in your application. If an existing Python object encapsulates this schema, use the
type
argument to specify it as the return type:@returns.from_json(type=User) @get("/users/{username}") def get_user(self, username): """Get a specific user."""
For Python 3 users, you can alternatively provide a return value annotation. Hence, the previous code is equivalent to the following in Python 3:
@returns.from_json @get("/users/{username}") def get_user(self, username) -> User: """Get a specific user."""
Both usages typically require also registering a converter that knows how to deserialize the JSON into the specified
type
(seeuplink.loads.from_json()
). This step is unnecessary if thetype
is defined using a library for which Uplink has built-in support, such asmarshmallow
.New in version v0.6.0.
alias of
json
-
class
uplink.returns.
schema
(type)[source]¶ Specifies that the function returns a specific type of response.
In Python 3, to provide a consumer method’s return type, you can set it as the method’s return annotation:
@get("/users/{username}") def get_user(self, username) -> UserSchema: """Get a specific user."""
For Python 2.7 compatibility, you can use this decorator instead:
@returns.schema(UserSchema) @get("/users/{username}") def get_user(self, username): """Get a specific user."""
To have Uplink convert response bodies into the desired type, you will need to define an appropriate converter (e.g., using
uplink.loads
).New in version v0.5.1.
retry¶
-
class
uplink.retry.
retry
(when=None, max_attempts=None, on_exception=None, stop=None, backoff=None)[source]¶ A decorator that adds retry support to a consumer method or to an entire consumer.
Unless you specify the
when
oron_exception
argument, all failed requests that raise an exception are retried.Unless you specify the
max_attempts
orstop
argument, this decorator continues retrying until the server returns a response.Unless you specify the
backoff
argument, this decorator uses capped exponential backoff and jitter, which should benefit performance with remote services under high contention.Note
Response and error handlers (see here) are invoked after the retry condition breaks or all retry attempts are exhausted, whatever comes first. These handlers will receive the first response/exception that triggers the retry’s
stop
condition or doesn’t match itswhen
filter.In other words, responses or exceptions that match the retry condition (e.g., retry when status code is 5xx) are not subject to response or error handlers as long as the request doesn’t break the retry’s stop condition (e.g., stop retrying after 5 attempts).
Parameters: - when (optional) – A predicate that determines when a retry should be attempted.
- max_attempts (int, optional) – The number of retries to attempt. If not specified, requests are retried continuously until a response is rendered.
- on_exception (
Exception
, optional) – The exception type that should prompt a retry attempt. - stop (
callable
, optional) – A function that creates predicates that decide when to stop retrying a request. - backoff (
RetryBackoff
,callable
, optional) – A backoff strategy or a function that creates an iterator over the ordered sequence of timeouts between retries. If not specified, exponential backoff is used.
-
class
uplink.retry.
RetryBackoff
[source]¶ Base class for a strategy that calculates the timeout between retry attempts.
You can compose two
RetryBackoff
instances by using the|
operator:CustomBackoffA() | CustomBackoffB()
The resulting backoff strategy will first compute the timeout using the left-hand instance. If that timeout is
None
, the strategy will try to compute a fallback using the right-hand instance. If both instances returnNone
, the resulting strategy will also returnNone
.-
get_timeout_after_exception
(request, exc_type, exc_val, exc_tb)[source]¶ Returns the number of seconds to wait before retrying the request, or
None
to indicate that the given exception should be raised.
-
retry.when¶
The default behavior of the retry
decorator is to retry on
any raised exception. To override the retry criteria, use the retry
decorator’s when
argument to specify a retry condition exposed through the
uplink.retry.when
module:
from uplink.retry.when import raises
class GitHub(uplink.Consumer):
# Retry when a client connection timeout occurs
@uplink.retry(when=raises(retry.CONNECTION_TIMEOUT))
@uplink.get("/users/{user}")
def get_user(self, user):
"""Get user by username."""
Use the |
operator to logically combine retry conditions:
from uplink.retry.when import raises, status
class GitHub(uplink.Consumer):
# Retry when an exception is raised or the status code is 503
@uplink.retry(when=raises(Exception) | status(503))
@uplink.get("/users/{user}")
def get_user(self, user):
"""Get user by username."""
-
class
uplink.retry.when.
raises
(expected_exception)[source]¶ Retry when a specific exception type is raised.
retry.backoff¶
Retrying failed requests typically involves backoff: the client can wait some time before the next retry attempt to avoid high contention on the remote service.
To this end, the retry
decorator uses capped
exponential backoff with jitter
by default, To override this, use the decorator’s backoff
argument
to specify one of the alternative approaches exposed through the
uplink.retry.backoff
module:
from uplink import retry, Consumer, get
from uplink.retry.backoff import fixed
class GitHub(Consumer):
# Employ a fixed one second delay between retries.
@retry(backoff=fixed(1))
@get("user/{username}")
def get_user(self, username):
"""Get user by username."""
from uplink.retry.backoff import exponential
class GitHub(uplink.Consumer):
@uplink.retry(backoff=exponential(multiplier=0.5))
@uplink.get("/users/{user}")
def get_user(self, user):
"""Get user by username."""
You can implement a custom backoff strategy by extending the class
uplink.retry.RetryBackoff
:
from uplink.retry import RetryBackoff
class MyCustomBackoff(RetryBackoff):
...
class GitHub(uplink.Consumer):
@uplink.retry(backoff=MyCustomBackoff())
@uplink.get("/users/{user}")
def get_user(self, user):
pass
-
class
uplink.retry.backoff.
jittered
(base=2, multiplier=1, minimum=0, maximum=4.611686018427388e+18)[source]¶ Waits using capped exponential backoff and full jitter.
The implementation is discussed in this AWS Architecture Blog post, which recommends this approach for any remote clients, as it minimizes the total completion time of competing clients in a distributed system experiencing high contention.
retry.stop¶
By default, the retry
decorator will repeatedly retry the
original request until a response is rendered. To override this behavior,
use the retry
decorator’s stop
argument to specify one
of the strategies exposed in the uplink.retry.stop
module:
from uplink.retry.stop import after_attempt
class GitHub(uplink.Consumer):
@uplink.retry(stop=after_attempt(3))
@uplink.get("/users/{user}")
def get_user(self, user):
Use the |
operator to logically combine strategies:
from uplink.retry.stop import after_attempt, after_delay
class GitHub(uplink.Consumer):
# Stop after 3 attempts or after the backoff exceeds 10 seconds.
@uplink.retry(stop=after_attempt(3) | after_delay(10))
@uplink.get("/users/{user}")
def get_user(self, user):
pass
-
class
uplink.retry.stop.
after_attempt
(attempt)[source]¶ Stops retrying after the specified number of
attempts
.
-
class
uplink.retry.stop.
after_delay
(delay)[source]¶ Stops retrying after the backoff exceeds the specified
delay
in seconds.
-
uplink.retry.stop.
NEVER
¶ Continuously retry until the server returns a successful response
ratelimit¶
-
class
uplink.
ratelimit
(calls=15, period=900, raise_on_limit=False, group_by=<function _get_host_and_port>, clock=<built-in function monotonic>)[source]¶ A decorator that constrains a consumer method or an entire consumer to making a specified maximum number of requests within a defined time period (e.g., 15 calls every 15 minutes).
Note
The rate limit is enforced separately for each host-port combination. Logically, requests are grouped by host and port, and the number of requests within a time period are counted and capped separately for each group.
By default, when the limit is reached, the client will wait until the current period is over before executing any subsequent requests. If you’d prefer the client to raise an exception when the limit is exceeded, set the
raise_on_limit
argument.Parameters: - calls (int) – The maximum number of allowed calls that the consumer can make within the time period.
- period (float) – The duration of each time period in seconds.
- raise_on_limit (
Exception
or bool, optional) – Either an exception to raise when the client exceeds the rate limit or abool
. IfTrue
, aRateLimitExceeded
exception is raised.