Datasources#

Cannula provides a handful of Datasources that assist with loading data while avoiding resolving the same data multiple times.

HTTP Data Source#

Note

This requires the http extras to be installed:

pip install cannula[http]

class cannula.datasource.http.HTTPDataSource(request, client=None)#

HTTP Data Source

This is modeled after the apollo http datasource. It uses httpx to preform async requests to any remote service you wish to query.

Properties:

base_url: Optional base_url to apply to all requests
timeout: Default timeout in seconds for requests (5 seconds)
resource_name: Optional name to use for __typename in responses.

async delete(path)#

Preform a DELETE request

Parameters:: path (str) – path of the request
Return type:: Any

did_receive_error(error, request)#: Handle errors from the remote resource

async did_receive_response(response, request)#

Hook to alter the response from the server.

example:

async def did_receive_response(
    self, response: httpx.Response, request: Request
) -> typing.Any:
    response.raise_for_status()
    return Widget(**response.json())

Return type:: Any

async get(path)#

Preform a GET request

Parameters:: path (str) – path of the request
Return type:: Any

async patch(path, body)#

Preform a PATCH request

Parameters:

path (str) – path of the request
body (Any) – body of the request

Return type:

Any

async post(path, body)#

Preform a POST request

Parameters:

path (str) – path of the request
body (Any) – body of the request

Return type:

Any

async put(path, body)#

Preform a PUT request

Parameters:

path (str) – path of the request
body (Any) – body of the request

Return type:

Any

will_send_request(request)#

Hook for subclasses to modify the request before it is sent.

For example setting Authorization headers:

def will_send_request(self, request):
    request.headers = self.request.headers
    return request

Return type:: Request

class cannula.datasource.http.Request(url, method, body, headers)#

body: Any#: Alias for field number 2

headers: Dict#: Alias for field number 3

method: str#: Alias for field number 1

url: str#: Alias for field number 0