Decorators
The method decorators detailed in this section describe request properties that are relevant to all invocations of a consumer method.
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 takes the same arguments as dict.
| PARAMETER | DESCRIPTION |
|---|---|
arg
|
A dict containing header values.
DEFAULT:
|
**kwargs
|
More header values.
DEFAULT:
|
Source code in uplink/decorators.py
A decorator that adds static query parameters for API calls.
When used as a class decorator, params applies to
all consumer methods bound to the class:
params takes the same arguments as dict.
| PARAMETER | DESCRIPTION |
|---|---|
arg
|
A dict containing query parameters.
DEFAULT:
|
**kwargs
|
More query parameters.
DEFAULT:
|
Source code in uplink/decorators.py
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 abc.Mapping.
You can alternatively use the uplink.Field annotation to
specify JSON fields separately, across multiple arguments:
Example
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:
The tuple ("user", "name") specifies the path to the
highlighted inner field:
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 with
uplink.Field.
Example
Sends multipart form data.
Multipart requests are commonly used to upload files to a server.
Further, annotate each part argument with Part.
Examples:
@multipart
@put("/user/photo")
def update_user(self, photo: Part, description: Part):
"""Upload a user profile photo."""
Time to wait for a server response before giving up.
When used on other decorators it specifies how long (in seconds) a decorator should wait before giving up.
Example
When used as a class decorator, timeout applies to all
consumer methods bound to the class.
| PARAMETER | DESCRIPTION |
|---|---|
seconds
|
An integer used to indicate how long should the request wait.
|
Annotate method arguments using positional or keyword arguments.
Arrange annotations in the same order as their corresponding function arguments.
Examples:
Basic usage with positional annotations:
@args(Path, Query)
@get("/users/{username}")
def get_user(self, username, visibility):
"""Get a specific user."""
Using keyword args to target specific method parameters:
@args(visibility=Query)
@get("/users/{username}")
def get_user(self, username, visibility):
"""Get a specific user."""
| PARAMETER | DESCRIPTION |
|---|---|
*annotations
|
Any number of annotations.
DEFAULT:
|
**more_annotations
|
More annotations, targeting specific method arguments.
DEFAULT:
|
Source code in uplink/decorators.py
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:
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:
Lastly, the decorator supports the optional argument
requires_consumer. When this option is set to True,
the registered callback should accept a reference to the
Consumer instance as its leading argument:
Example:
Added in version 0.4.0
Source code in uplink/hooks.py
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:
Lastly, the decorator supports the optional argument
requires_consumer. When this option is set to True,
the registered callback should accept a reference to the
Consumer instance as its leading argument:
Example:
@error_handler(requires_consumer=True)
def raise_api_error(consumer, exc_type, exc_val, exc_tb):
...
Added 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.
Source code in uplink/hooks.py
A decorator that applies one or more hooks to a request method.
Added in version 0.4.0
Source code in uplink/hooks.py
HTTP Method Decorators
These decorators define the HTTP method to use for a request.
Creates a decorator that makes the decorated function an HTTP GET request handler.
| PARAMETER | DESCRIPTION |
|---|---|
uri
|
The URI template for the request. Can include path parameters in the format {param_name}. If not provided, can be specified when the decorator is used.
TYPE:
|
args
|
A sequence of argument annotations or mapping of argument names to annotations for the handler method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
|
A decorator that can be used to define a GET request handler method. |
Examples:
Basic usage with path parameter:
Using query parameters:
@get("users")
def list_users(self, page: Query, per_page: Query = 30):
"""List users with pagination."""
Source code in uplink/commands.py
Creates a decorator that makes the decorated function an HTTP POST request handler.
| PARAMETER | DESCRIPTION |
|---|---|
uri
|
The URI template for the request. Can include path parameters in the format {param_name}. If not provided, can be specified when the decorator is used.
TYPE:
|
args
|
A sequence of argument annotations or mapping of argument names to annotations for the handler method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
|
A decorator that can be used to define a POST request handler method. |
Examples:
Using JSON body:
Using form data:
@form_url_encoded
@post("auth/login")
def login(self, username: Field, password: Field):
"""Login with credentials."""
Source code in uplink/commands.py
Creates a decorator that makes the decorated function an HTTP PUT request handler.
| PARAMETER | DESCRIPTION |
|---|---|
uri
|
The URI template for the request. Can include path parameters in the format {param_name}. If not provided, can be specified when the decorator is used.
TYPE:
|
args
|
A sequence of argument annotations or mapping of argument names to annotations for the handler method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
|
A decorator that can be used to define a PUT request handler method. |
Examples:
Using request body:
@put("users/{username}")
def update_user(self, username, **user_info: Body):
"""Update a specific user."""
Using form fields:
@form_url_encoded
@put("users/{username}")
def update_user(self, username, name: Field, email: Field):
"""Update a user's information."""
Source code in uplink/commands.py
Creates a decorator that makes the decorated function an HTTP PATCH request handler.
| PARAMETER | DESCRIPTION |
|---|---|
uri
|
The URI template for the request. Can include path parameters in the format {param_name}. If not provided, can be specified when the decorator is used.
TYPE:
|
args
|
A sequence of argument annotations or mapping of argument names to annotations for the handler method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
|
A decorator that can be used to define a PATCH request handler method. |
Examples:
Partial update with JSON:
@json
@patch("users/{username}")
def update_user_partially(self, username, **updates: Body):
"""Partially update a user."""
Update specific fields:
@form_url_encoded
@patch("users/{username}")
def update_status(self, username, status: Field):
"""Update a user's status."""
Source code in uplink/commands.py
Creates a decorator that makes the decorated function an HTTP DELETE request handler.
| PARAMETER | DESCRIPTION |
|---|---|
uri
|
The URI template for the request. Can include path parameters in the format {param_name}. If not provided, can be specified when the decorator is used.
TYPE:
|
args
|
A sequence of argument annotations or mapping of argument names to annotations for the handler method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
|
A decorator that can be used to define a DELETE request handler method. |
Examples:
Basic usage:
With query parameters:
@delete("users/{username}/posts")
def delete_posts(self, username, before_date: Query = None):
"""Delete user's posts, optionally before a specific date."""
Source code in uplink/commands.py
Creates a decorator that makes the decorated function an HTTP HEAD request handler.
| PARAMETER | DESCRIPTION |
|---|---|
uri
|
The URI template for the request. Can include path parameters in the format {param_name}. If not provided, can be specified when the decorator is used.
TYPE:
|
args
|
A sequence of argument annotations or mapping of argument names to annotations for the handler method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
|
A decorator that can be used to define a HEAD request handler method. |
Examples:
Basic usage:
Source code in uplink/commands.py
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.
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:
If returning the list of errors is unnecessary, we can use the
key argument to strictly return the nested field
data.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."""
Added in version 0.5.0
Source code in uplink/returns.py
module-attribute
from_json = json
Specifies that the decorated consumer method should produce
instances of a type class using a registered
deserialization strategy (see uplink.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
:py:attr: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
(see uplink.loads.from_json). This step is unnecessary if
the type is defined using a library for which Uplink has
built-in support, such as marshmallow.
Added in version 0.6.0
Specifies that the function returns a specific type of response.
The standard way to provide a consumer method's return type is to set it as the method's return annotation:
Alternatively, 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).
Added in version 0.5.1
retry.*
Base class for a strategy that calculates the timeout between retry attempts.
You can compose two RetryBackoff instances by using the |
operator:
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 return None, the resulting strategy will also
return None.
retry.when
The default behavior of the uplink.retry decorator is to retry on any
raised exception. To override the retry criteria, use the
uplink.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."""
Retry after receiving a 5xx (server error) response.
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 uplink.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
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.
Source code in uplink/retry/backoff.py
Waits using capped exponential backoff, meaning that the delay
is multiplied by a constant base after each attempt, up to
an optional maximum value.
Source code in uplink/retry/backoff.py
retry.stop
By default, the uplink.retry decorator will repeatedly retry the
original request until a response is rendered. To override this
behavior, use the uplink.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
module-attribute
Continuously retry until the server returns a successful response.
ratelimit
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.
| PARAMETER | DESCRIPTION |
|---|---|
calls
|
The maximum number of allowed calls that the consumer can make within the time period.
DEFAULT:
|
period
|
The duration of each time period in seconds.
DEFAULT:
|
raise_on_limit
|
Either an exception to raise when the client
exceeds the rate limit or a boolean. If
DEFAULT:
|