Uplink 📡

A Declarative HTTP Client for Python. Inspired by Retrofit.

Note

Uplink is currently in initial development. Until the official release (v1.0.0), the public API should be considered provisional, Although we don’t expect any considerable changes to the API at this point, please avoid using the code in production, for now.

However, while Uplink is under construction, we invite eager users to install early and provide open feedback, which can be as simple as opening a GitHub issue when you notice a missing feature, latent defect, documentation oversight, etc.

Moreover, for those interested in contributing, checkout the Contribution Guide on GitHub!

Uplink turns your HTTP API into a Python class.

from uplink import Consumer, get, headers, Path, Query

@headers({"Accept": "application/vnd.github.v3.full+json"})
class GitHub(Consumer):

   @get("users/{user}/repos")
   def list_repos(self, user: Path, sort_by: Query("sort")):
      """Get user's public repositories."""

Build an instance to interact with the webservice.

github = GitHub(base_url="https://api.github.com/")

Then, executing an HTTP request is as simply as invoking a method.

repos = github.list_repos("octocat", sort_by="created")

The returned object is a friendly requests.Response:

print(repos.json())
# Output: [{'id': 64778136, 'name': 'linguist', ...

For sending non-blocking requests, Uplink comes with support for aiohttp and twisted (example).

Use decorators and function annotations to describe the HTTP request:

• URL parameter replacement and query parameter support
• Convert responses into Python objects (e.g., using marshmallow)
• JSON, URL-encoded, and multipart request body and file upload

The User Manual

Follow this guide to get up and running with Uplink.

Installation

Using pip

With pip (or pipenv), you can install Uplink simply by typing:

$ pip install -U uplink

Download the Source Code

Uplink’s source code is in a public repository hosted on GitHub.

As an alternative to installing with pip, you could clone the repository,

$ git clone https://github.com/prkumar/uplink.git

then, install; e.g., with setup.py:

$ cd uplink
$ python setup.py install

Extras

These are optional integrations and features that extend the library’s core functionality and typically require an additional dependency.

When installing Uplink with pip, you can specify any of the following extras, to add their respective dependencies to your installation:

Extra	Description
aiohttp	Enables uplink.AiohttpClient, for sending non-blocking requests and receiving awaitable responses.
marshmallow	Enables uplink.MarshmallowConverter, for converting JSON responses directly into Python objects using marshmallow.Schema.
twisted	Enables uplink.TwistedClient, for sending non-blocking requests and receiving Deferred responses.

To download all available features, run

$ pip install -U uplink[aiohttp, marshmallow, twisted]

Introduction

Uplink delivers reusable and self-sufficient objects for accessing HTTP webservices, with minimal code and user pain. Simply define your consumers using decorators and function annotations, and we’ll handle the rest for you… pun intended, obviously 😎

Static Request Handling

Method decorators describe request properties that are relevant to all invocations of a consumer method.

For instance, consider the following GitHub API consumer:

class GitHub(uplink.Consumer):
    @uplink.timeout(60)
    @uplink.get("/repositories")
    def get_repos(self):
        """Dump every public repository."""

Annotated with timeout, the method get_repos() will build HTTP requests that wait an allotted number of seconds – 60, in this case – for the server to respond before giving up.

As method annotations are simply decorators, you can stack one on top of another for chaining:

class GitHub(uplink.Consumer):
    @uplink.headers({"Accept": "application/vnd.github.v3.full+json"})
    @uplink.timeout(60)
    @uplink.get("/repositories")
    def get_repos(self):
        """Dump every public repository."""

Dynamic Request Handling

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.

To illustrate, for the method get_user() in the following snippet, we have flagged the argument username as a URI placeholder replacement using the Path annotation:

class GitHub(uplink.Consumer):
    @uplink.get("users/{username}")
    def get_user(self, username: uplink.Path("username")): pass

Invoking this method on a consumer instance, like so:

github.get_user(username="prkumar")

Builds an HTTP request that has a URL ending with users/prkumar.

Note

As you probably took away from the above example: when parsing the method’s signature for argument annotations, uplink skips the instance reference argument, which is the leading method parameter and usually named self.

Quickstart

Decorators and function annotations indicate how a request will be handled.

Request Method

Uplink offers decorators that turn any method into a request definition. These decorators provide the request method and relative URL of the intended request: get, post, put, patch and delete.

The relative URL of the resource is specified in the decorator.

@get("users/list")

You can also specify query parameters in the URL.

@get("users/list?sort=desc")

Moreover, request methods must be bound to a Consumer subclass.

class MyApi(Consumer):
    @get("users/list")
    def list_users(self):
        """List all users."""

URL Manipulation

A request URL can be updated dynamically using URI template parameters. A simple URI parameter is an alphanumeric string surrounded by { and }.

To match the parameter with a method argument, either match the argument’s name with the alphanumeric string, like so

@get("group/{id}/users")
def group_list(self, id): pass

or use the Path annotation.

@get("group/{id}/users")
def group_list(self, group_id: Path("id")): pass

Query parameters can also be added.

@get("group/{id}/users")
def group_list(self, group_id: Path("id"), sort: Query): pass

For complex query parameter combinations, a mapping can be used:

@get("group/{id}/users")
def group_list(self, group_id: Path("id"), options: QueryMap): pass

Request Body

An argument’s value can be specified for use as an HTTP request body with the Body annotation:

@post("users/new")
def create_user(self, user: Body): pass

This annotation works well with the keyword arguments parameter (denoted by the ** prefix):

@post("users/new")
def create_user(self, **user_info: Body): pass

Form Encoded, Multipart, and JSON

Methods can also be declared to send form-encoded, multipart, and JSON data.

Form-encoded data is sent when form_url_encoded decorates the method. Each key-value pair is annotated with a Field annotation:

@form_url_encoded
@post("user/edit")
def update_user(self, first_name: Field, last_name: Field): pass

Multipart requests are used when multipart decorates the method. Parts are declared using the Part annotation:

@multipart
@put("user/photo")
def update_user(self, photo: Part, description: Part): pass

JSON data is sent when json decorates the method. The Body annotation declares the JSON payload:

@uplink.json
@uplink.patch("/user")
def update_user(self, **user_info: uplink.Body):
    """Update an authenticated user."""

Header Manipulation

You can set static headers for a method using the headers decorator.

@headers({
    "Accept": "application/vnd.github.v3.full+json",
    "User-Agent": "Uplink-Sample-App"
})
@get("users/{username}")
def get_user(self, username): pass

headers can be used as a class decorator for headers that need to be added to every request:

@headers({
    "Accept": "application/vnd.github.v3.full+json",
    "User-Agent": "Uplink-Sample-App"
})
class GitHub(Consumer):
    ...

A request header can be updated dynamically using the Header function parameter annotation:

@get("user")
def get_user(self, authorization: Header):
    """Get an authenticated user."""

Synchronous vs. Asynchronous

By default, Uplink uses the Requests library to make requests. However, the client parameter of the Consumer constructor offers a way to swap out Requests with another HTTP client:

github = GitHub(BASE_URL, client=...)

Notably, Requests blocks while waiting for a response from a server. For non-blocking requests, Uplink comes with optional support for asyncio and twisted. Checkout this example on GitHub for more.

Deserializing the Response

The converter parameter of the Consumer constructor accepts an adapter class that handles deserialization of HTTP response objects.

github = GitHub(BASE_URL, converter=...)

For instance, the MarshmallowConverter adapter turns JSON HTTP responses into Python objects using the marshmallow.Schema object. Checkout this example on GitHub for more.

Tips & Tricks

Here are a few ways to simplify consumer definitions.

Decorating All Request Methods in a Class

To apply an decorator across all methods in a class, you can simply decorate the class rather than each method individually:

 @uplink.timeout(60)
 class GitHub(uplink.Consumer):
     @uplink.get("/repositories")
     def get_repos(self):
         """Dump every public repository."""

     @uplink.get("/organizations")
     def get_organizations(self):
         """List all organizations."""

Hence, the consumer defined above is equivalent to the following, slightly more verbose alternative:

class GitHub(uplink.Consumer):
    @uplink.timeout(60)
    @uplink.get("/repositories")
    def get_repos(self):
        """Dump every public repository."""

    @uplink.timeout(60)
    @uplink.get("/organizations")
    def get_organizations(self):
        """List all organizations."""

Adopting the Argument’s Name

When you initialize a named annotation, such as a Path or Field, without a name (by omitting the name parameter), it adopts the name of its corresponding method argument.

For example, in the snippet below, we can omit naming the Path annotation since the corresponding argument’s name, username, matches the intended URI path parameter:

class GitHub(uplink.Consumer):
    @uplink.get("users/{username}")
    def get_user(self, username: uplink.Path): pass

Annotating Your Arguments For Python 2.7

There are several ways to annotate arguments. Most examples in this documentation use function annotations, but this approach is unavailable for Python 2.7 users. Instead, you can use argument annotations as decorators or utilize the method annotation args.

Argument Annotations as Decorators

For one, annotations can work as function decorators. With this approach, annotations are mapped to arguments from “bottom-up”.

For instance, in the below definition, the Url annotation corresponds to commits_url, and Path to sha.

 class GitHub(uplink.Consumer):
     @uplink.Path
     @uplink.Url
     @uplink.get
     def get_commit(self, commits_url, sha): pass

Using uplink.args

The second approach involves using the method annotation args, arranging annotations in the same order as their corresponding function arguments (again, ignore self):

 class GitHub(uplink.Consumer):
     @uplink.args(uplink.Url, uplink.Path)
     @uplink.get
     def get_commit(self, commits_url, sha): pass

Function Annotations (Python 3 only)

Finally, when using Python 3, you can use these classes as function annotations (PEP 3107):

 class GitHub(uplink.Consumer):
     @uplink.get
     def get_commit(self, commit_url: uplink.Url, sha: uplink.Path):
         pass

The Public API

This guide details the classes and methods in Uplink’s public API.

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, **kwargs)

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 as dict.

Parameters:

• *arg – A dict containing header values.
• **kwargs – More header values.

json

class uplink.json

Use as a decorator to make JSON requests.

You should 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:collections.Mapping.

Example

@json
@patch(/user")
def update_user(self, **info: Body):
    """Update the current user."""

form_url_encoded

class uplink.form_url_encoded

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

@form_url_encoded
@post("/users/edit")
def update_user(self, first_name: Field, last_name: Field):
    """Update the current user."""

multipart

class uplink.multipart

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)

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)

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.

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.

Path

class uplink.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

Query

class uplink.Query(name=None, encoded=False, type=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.

Example

@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."""

Parameters:encoded (bool, optional) – Specifies whether the parameter name and value are already URL encoded.

QueryMap

class uplink.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."""

Parameters:encoded (bool, optional) – Specifies whether the parameter name and value are already URL encoded.

Header

class uplink.Header(name=None, type=None)

Pass a header as a method argument at runtime.

While uplink.headers attaches static headers that define all requests sent from a consumer method, this class turns a method argument into a dynamic header value.

Example

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

HeaderMap

class uplink.HeaderMap(type=None)

Pass a mapping of header fields at runtime.

Field

class uplink.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.

Example::

@form_url_encoded
@post("/users/edit")
def update_user(self, first_name: Field, last_name: Field):
    """Update the current user."""

FieldMap

class uplink.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.

Example

@form_url_encoded
@post("/user/edit")
def create_post(self, **user_info: FieldMap):
    """Update the current user."""

Part

class uplink.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.

Example

@multipart
@put(/user/photo")
def update_user(self, photo: Part, description: Part):
    """Upload a user profile photo."""

PartMap

class uplink.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

Example

@multipart
@put(/user/photo")
def update_user(self, photo: Part, description: Part):
    """Upload a user profile photo."""

Body

class uplink.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.

Example

@json
@patch(/user")
def update_user(self, **info: Body):
    """Update the current user."""

Url

class 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

Example

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

HTTP Clients

The client parameter of the Consumer constructor offers a way to swap out Requests with another HTTP client, including those listed here:

github = GitHub(BASE_URL, client=...)

Requests

class uplink.RequestsClient(session=None)

A requests client that returns requests.Response responses.

Parameters:session (requests.Session, optional) – The session that should handle sending requests. If this argument is omitted or set to None, a new session will be created.

Aiohttp

class uplink.AiohttpClient(session=None)

An aiohttp client that creates awaitable responses.

Note

This client is an optional feature and requires the aiohttp package. For example, here’s how to install this extra using pip:

$ pip install uplink[aiohttp]

Parameters:session (aiohttp.ClientSession, optional) – The session that should handle sending requests. If this argument is omitted or set to None, a new session will be created.

Twisted

class uplink.TwistedClient(session=None)

Client that returns twisted.internet.defer.Deferred responses.

Note

This client is an optional feature and requires the twisted package. For example, here’s how to install this extra using pip:

$ pip install uplink[twisted]

Parameters:session (requests.Session, optional) – The session that should handle sending requests. If this argument is omitted or set to None, a new session will be created.

Converters

The converter parameter of the uplink.Consumer constructor accepts an adapter class that handles deserialization of HTTP response objects:

github = GitHub(BASE_URL, converter=...)

Marshmallow

class uplink.MarshmallowConverter

A converter that serializes and deserializes values using marshmallow schemas.

To deserialize JSON responses into Python objects with this converter, define a marshmallow.Schema subclass and set it as the return annotation of a consumer method:

@get("/users")
def get_users(self, username) -> UserSchema():
    '''Fetch a single user'''

Also, when instantiating a consumer, be sure to set this class as a converter for the instance:

github = GitHub(BASE_URL, converter=MarshmallowConverter())

Note

This converter is an optional feature and requires the marshmallow package. For example, here’s how to install this feature using pip:

$ pip install uplink[marshmallow]

Changelog

All notable changes to this project will be documented in this file.

The format is based on Keep a Changelog, and this project adheres to the Semantic Versioning scheme.

0.3.0 - 2018-1-09

Added

• HTTP HEAD request decorator by @brandonio21
• Support for returning deserialized response objects using marshmallow schemas.
• Constructor parameter for uplink.Query and uplink.QueryMap to support already encoded URL parameters
• Support for using requests.Session and aiohttp.ClientSession instances with the client parameter of the uplink.Consumer constructor.

Changed

• aiohttp and twisted are now optional dependencies/extras.

Fixed

• Fix issue with calling a request method with super.
• Fix issue where method decorators would incorrectly decorate inherited request methods.

0.2.2 - 2017-11-23

Fixed

• Fix error raised when an object that is not a class is passed into the client parameter of the Consumer constructor, by @kadrach.

0.2.0 - 2017-11-03

Added

• The class uplink.Consumer. Consumer classes should inherit this base class, and creating consumer instances happens through instantiation.
• Support for asyncio for Python 3.4 and above.
• Support for twisted for all supported Python versions.

Changed

• BREAKING: Invoking a consumer method now builds and executes the request, removing the extra step of calling the execute method.

Deprecated

• Building consumer instances with uplink.build. Instead, Consumer classes should inherit uplink.Consumer.

Fixed

• Header link for version 0.1.1 in changelog.

0.1.1 - 2017-10-21

Added

• Contribution guide, CONTRIBUTING.rst.
• “Contributing” Section in README.rst that links to contribution guide.
• AUTHORS.rst file for listing project contributors.
• Adopt Contributor Covenant Code of Conduct.

Changed

• Replaced tentative contributing instructions in preview notice on documentation homepage with link to contribution guide.

0.1.0 - 2017-10-19

Added

• Python ports for almost all method and argument annotations in Retrofit.
• Adherence to the variation of the semantic versioning scheme outlined in the official Python package distribution tutorial.
• MIT License
• Documentation with introduction, instructions for installing, and quick getting started guide covering the builder and all method and argument annotations.
• README that contains GitHub API v3 example, installation instructions with pip, and link to online documentation.