Skip to content

Function Annotations

For programming in general, function parameters drive a function's dynamic behavior; a function's output depends normally on its inputs. With uplink, function arguments parametrize an HTTP request, and you indicate the dynamic parts of the request by appropriately annotating those arguments with the classes detailed in this section.

uplink.Path 

Path(name=None, type=None)

Substitution of a path variable in a URI template.

URI template parameters are enclosed in braces (e.g., {name}). To map an argument to a declared URI parameter, use the Path annotation:

class TodoService(object):
    @get("/todos/{id}")
    def get_todo(self, todo_id: Path("id")): pass

Then, invoking get_todo with a consumer instance:

todo_service.get_todo(100)

creates an HTTP request with a URL ending in /todos/100.

Note

Any unannotated function argument that shares a name with a URL path parameter is implicitly annotated with this class at runtime.

For example, we could simplify the method from the previous example by matching the path variable and method argument names:

@get("/todos/{id}")
def get_todo(self, id): pass
Source code in uplink/arguments.py 
210
211
212
def __init__(self, name=None, type=None):
    self._arg_name = name
    super().__init__(type)

uplink.Query 

Query(name=None, encoded=False, type=None, encode_none=None)

Set a dynamic query parameter.

This annotation turns argument values into URL query parameters. You can include it as function argument annotation, in the format: <query argument>: uplink.Query.

If the API endpoint you are trying to query uses q as a query parameter, you can add q: uplink.Query to the consumer method to set the q search term at runtime.

Examples:

@get("/search/commits")
def search(self, search_term: Query("q")):
    """Search all commits with the given search term."""

To specify whether or not the query parameter is already URL encoded, use the optional encoded argument:

@get("/search/commits")
def search(self, search_term: Query("q", encoded=True)):
    """Search all commits with the given search term."""

To specify if and how None values should be encoded, use the optional encode_none argument:

@get("/search/commits")
def search(self, search_term: Query("q"),
           search_order: Query("o", encode_none="null")):
    """
    Search all commits with the given search term using the
    optional search order.
    """
PARAMETER DESCRIPTION
name

The name of the query parameter.

DEFAULT: None

encoded

Specifies whether the parameter name and value are already URL encoded.

DEFAULT: False

type

The type of the query parameter.

DEFAULT: None

encode_none

Specifies an optional string with which None values should be encoded. If not specified, parameters with a value of None will not be sent.

DEFAULT: None

Source code in uplink/arguments.py 
378
379
380
381
def __init__(self, name=None, encoded=False, type=None, encode_none=None):
    super().__init__(name, type)
    self._encoded = encoded
    self._encode_none = encode_none

uplink.QueryMap 

QueryMap(encoded=False, type=None)

A mapping of query arguments.

If the API you are using accepts multiple query arguments, you can include them all in your function method by using the format: <query argument>: uplink.QueryMap

Example 
@get("/search/users")
def search(self, **params: QueryMap):
    """Search all users."""
PARAMETER DESCRIPTION
encoded

Specifies whether the parameter name and value are already URL encoded.

DEFAULT: False

type

The type of the query parameters.

DEFAULT: None

Source code in uplink/arguments.py 
432
433
434
def __init__(self, encoded=False, type=None):
    super().__init__(type)
    self._encoded = encoded

uplink.Header 

Header(name=None, type=None)

Pass a header as a method argument at runtime.

While uplink.headers attaches request headers values that are static across all requests sent from the decorated consumer method, this annotation turns a method argument into a dynamic request header.

@get("/user")
def me(self, session_id: Header("Authorization")):
    """Get the authenticated user."""

To define an optional header, use the default value of None:

@get("/repositories")
def fetch_repos(self, auth: Header("Authorization") = None):
    """List all public repositories."""

When the argument is not used, the header will not be added to the request.

Source code in uplink/arguments.py 
210
211
212
def __init__(self, name=None, type=None):
    self._arg_name = name
    super().__init__(type)

uplink.HeaderMap 

HeaderMap(type=None)

Pass a mapping of header fields at runtime.

Source code in uplink/arguments.py 
184
185
def __init__(self, type=None):
    self._type = type

uplink.Field 

Field(name=None, type=None)

Defines a form field to the request body.

Use together with the decorator uplink.form_url_encoded and annotate each argument accepting a form field with uplink.Field.

@form_url_encoded
@post("/users/edit")
def update_user(self, first_name: Field, last_name: Field):
    """Update the current user."""
Source code in uplink/arguments.py 
210
211
212
def __init__(self, name=None, type=None):
    self._arg_name = name
    super().__init__(type)

uplink.FieldMap 

FieldMap(type=None)

Defines a mapping of form fields to the request body.

Use together with the decorator uplink.form_url_encoded and annotate each argument accepting a form field with uplink.FieldMap.

@form_url_encoded
@post("/user/edit")
def create_post(self, **user_info: FieldMap):
    """Update the current user."""
Source code in uplink/arguments.py 
184
185
def __init__(self, type=None):
    self._type = type

uplink.Part 

Part(name=None, type=None)

Marks an argument as a form part.

Use together with the decorator uplink.multipart and annotate each form part with uplink.Part.

@multipart
@put("/user/photo")
def update_user(self, photo: Part, description: Part):
    """Upload a user profile photo."""
Source code in uplink/arguments.py 
210
211
212
def __init__(self, name=None, type=None):
    self._arg_name = name
    super().__init__(type)

uplink.PartMap 

PartMap(type=None)

A mapping of form field parts.

Use together with the decorator uplink.multipart and annotate each part of form parts with uplink.PartMap

@multipart
@put("/user/photo")
def update_user(self, photo: Part, description: Part):
    """Upload a user profile photo."""
Source code in uplink/arguments.py 
184
185
def __init__(self, type=None):
    self._type = type

uplink.Body 

Body(type=None)

Set the request body at runtime.

Use together with the decorator uplink.json. The method argument value will become the request's body when annotated with uplink.Body.

@json
@patch("/user")
def update_user(self, **info: Body):
    """Update the current user."""
Source code in uplink/arguments.py 
184
185
def __init__(self, type=None):
    self._type = type

uplink.Url

Sets a dynamic URL.

Provides the URL at runtime as a method argument. Drop the decorator parameter path from uplink.get and annotate the corresponding argument with uplink.Url

@get
def get(self, endpoint: Url):
    """Execute a GET requests against the given endpoint"""

uplink.Timeout

Passes a timeout as a method argument at runtime.

While uplink.timeout attaches static timeout to all requests sent from a consumer method, this class turns a method argument into a dynamic timeout value.

Example 
@get("/user/posts")
def get_posts(self, timeout: Timeout() = 60):
    """Fetch all posts for the current users giving up after given
    number of seconds."""

uplink.Context 

Context(name=None, type=None)

Defines a name-value pair that is accessible to middleware at runtime.

Request middleware can leverage this annotation to give users control over the middleware's behavior.

Example

Consider a custom decorator @cache (this would be a subclass of uplink.decorators.MethodAnnotation):

@cache(hours=3)
@get("users/user_id")
def get_user(self, user_id)
    """Retrieves a single user."""

As its name suggests, the @cache decorator enables caching server responses so that, once a request is cached, subsequent identical requests can be served by the cache provider rather than adding load to the upstream service.

Importantly, the writers of the @cache decorators can allow users to pass their own cache provider implementation through an argument annotated with Context:

@cache(hours=3)
@get("users/user_id")
def get_user(self, user_id, cache_provider: Context)
    """Retrieves a single user."""

To add a name-value pair to the context of any request made from a Consumer instance, you can use the Consumer.session.context property. Alternatively, you can annotate a constructor argument of a Consumer subclass with Context, as explained [here][annotating constructor arguments].

Source code in uplink/arguments.py 
210
211
212
def __init__(self, name=None, type=None):
    self._arg_name = name
    super().__init__(type)