API reference#

The entire public API is available at root level:

from snug import Query, Request, send_async, PATCH, paginated, ...

Query#

Types and functionality relating to queries

class snug.query.Query[source]#

Abstract base class for query-like objects. Any object whose __iter__() returns a Request/Response generator implements it.

Note

Generator iterators themselves also implement this interface (i.e. __iter__() returns the generator itself).

Note

Query is a Generic. This means you may write Query[<returntype>] as a descriptive type annotation.

For example: Query[bool] indicates a query which returns a boolean.

Examples

Creating a query from a generator function:

>>> def repo(name: str, owner: str) -> snug.Query[dict]:
...    """a repo lookup by owner and name"""
...    req = snug.GET(f'https://api.github.com/repos/{owner}/{name}')
...    response = yield req
...    return json.loads(response.content)
...
>>> query = repo('Hello-World', owner='octocat')

Creating a class-based query:

>>> # actually subclassing `Query` is not required
>>> class repo(snug.Query[dict]):
...    """a repository lookup by owner and name"""
...    def __init__(self, name: str, owner: str):
...        self.name, self.owner = name, owner
...
...    def __iter__(self):
...        owner, name = self.owner, self.name
...        request = snug.GET(
...            f'https://api.github.com/repos/{owner}/{name}')
...        response = yield request
...        return json.loads(response.content)
...
>>> query = repo('Hello-World', owner='octocat')

Creating a fully customized query (for rare cases):

>>> # actually subclassing `Query` is not required
>>> class asset_download(snug.Query):
...     """streaming download of a repository asset.
...     Can only be executed with particular clients"""
...     def __init__(self, repo_name, repo_owner, id):
...         self.request = snug.GET(
...             f'https://api.github.com/repos/{repo_owner}'
...             f'/{repo_name}/releases/assets/{id}',
...             headers={'Accept': 'application/octet-stream'})
...
...     def __execute__(self, client, authenticate):
...         """execute, returning a streaming requests response"""
...         assert isinstance(client, requests.Session)
...         req = authenticate(self.request)
...         return client.request(req.method, req.url,
...                               data=req.content,
...                               params=req.params,
...                               headers=req.headers,
...                               stream=True)
...
...     async def __execute_async__(self, client, authenticate):
...         ...
...
>>> query = asset_download('hub', rep_owner='github', id=4187895)
>>> response = snug.execute(download, client=requests.Session())
>>> for chunk in response.iter_content():
...    ...

__iter__()[source]#

A generator iterator which resolves the query

Return type:: Generator[Request, Response, T]

__execute__(client, auth)[source]#

Default execution logic for a query, which uses the query’s __iter__(). May be overriden for full control of query execution, at the cost of reusability.

New in version 1.1.

Note

You shouldn’t need to override this method, except in rare cases where implementing Query.__iter__() does not suffice.

Parameters:

client – the client instance passed to execute()
auth (Callable[[Request], Request]) – a callable to authenticate a Request

Returns:

the query result

Return type:

async __execute_async__(client, auth)[source]#

Default asynchronous execution logic for a query, which uses the query’s __iter__(). May be overriden for full control of query execution, at the cost of reusability.

New in version 1.1.

Note

You shouldn’t need to override this method, except in rare cases where implementing Query.__iter__() does not suffice.

Parameters:

client – the client instance passed to execute_async()
auth (Callable[[Request], Request]) – a callable to authenticate a Request

Returns:

the query result

Return type:

snug.query.execute(query, auth=None, client=<urllib.request.OpenerDirector object>)[source]#

Execute a query, returning its result

Parameters:

query (Query[T]) – The query to resolve
auth (Tuple[str, str] or Callable[[Request], Request] or None) –
This may be:
- A (username, password)-tuple for basic authentication
- A callable to authenticate requests.
- None (no authentication)
client – The HTTP client to use. Its type must have been registered with send(). If not given, the built-in urllib module is used.

Returns:

the query result

Return type:

snug.query.execute_async(query, auth=None, client=None)[source]#

Execute a query asynchronously, returning its result

Parameters:

query (Query[T]) – The query to resolve
auth (Tuple[str, str] or Callable[[Request], Request] or None) –
This may be:
- A (username, password)-tuple for basic authentication
- A callable to authenticate requests.
- None (no authentication)
client – The HTTP client to use. Its type must have been registered with send_async(). If not given, the current event loop from asyncio is used.

Returns:

the query result

Return type:

Note

The default client is very rudimentary. Consider using a aiohttp.ClientSession instance as client.

snug.query.executor(**kwargs)[source]#

Create a version of execute() with bound arguments.

Parameters:: **kwargs – arguments to pass to execute()
Returns:: an execute()-like function
Return type:: Callable[[Query[T]], T]

snug.query.async_executor(**kwargs)[source]#

Create a version of execute_async() with bound arguments.

Parameters:: **kwargs – arguments to pass to execute_async()
Returns:: an execute_async()-like function
Return type:: Callable[[Query[T]], Awaitable[T]]

class snug.query.related(cls)[source]#

Decorate classes to make them callable as methods. This can be used to implement related queries through nested classes.

Example

>>> class Foo:
...     @related
...     class Bar:
...         def __init__(self, foo, qux):
...             self.the_foo, self.qux = foo, qux
...
>>> f = Foo()
>>> b = p.Bar(qux=5)
>>> isinstance(b, Foo.Bar)
True
>>> b.the_foo is f
True

HTTP#

Basic HTTP abstractions and functionality

class snug.http.Request(method, url, content=None, params={}, headers={})[source]#

A simple HTTP request.

Parameters:

method (str) – The http method
url (str) – The requested url
content (bytes or None) – The request content
params (Mapping) – The query parameters.
headers (Mapping) – Request headers.

with_headers(headers)[source]#

Create a new request with added headers

Parameters:: headers (Mapping) – the headers to add

with_prefix(prefix)[source]#

Create a new request with added url prefix

Parameters:: prefix (str) – the URL prefix

with_params(params)[source]#

Create a new request with added query parameters

Parameters:: params (Mapping) – the query parameters to add

class snug.http.Response(status_code, content=None, headers={})[source]#

A simple HTTP response.

Parameters:

status_code (int) – The HTTP status code
content (bytes or None) – The response content
headers (Mapping) – The headers of the response.

snug.http.basic_auth(credentials)[source]#

Create an HTTP basic authentication callable

Parameters:: credentials (Tuple[str, str]) – The (username, password)-tuple
Returns:: A callable which adds basic authentication to a Request.
Return type:: Callable[[Request], Request]

snug.http.GET(url, content=None, params={}, headers={})#: Shortcut for a GET request

snug.http.POST(url, content=None, params={}, headers={})#: Shortcut for a POST request

snug.http.PUT(url, content=None, params={}, headers={})#: Shortcut for a PUT request

snug.http.PATCH(url, content=None, params={}, headers={})#: Shortcut for a PATCH request

snug.http.DELETE(url, content=None, params={}, headers={})#: shortcut for a DELETE request

snug.http.HEAD(url, content=None, params={}, headers={})#: Shortcut for a HEAD request

snug.http.OPTIONS(url, content=None, params={}, headers={})#: Shortcut for a OPTIONS request

snug.http.header_adder = functools.partial(<class 'operator.methodcaller'>, 'with_headers')#

Make a callable which adds headers to a request

Example

>>> func = snug.header_adder({'content-type': 'application/json'})
>>> func(snug.GET('https://test.dev')).headers
{'content-type': 'application/json'}

snug.http.prefix_adder = functools.partial(<class 'operator.methodcaller'>, 'with_prefix')#

Make a callable which adds a prefix to a request url

Example

>>> func = snug.prefix_adder('https://api.test.com/v1/')
>>> func(snug.GET('foo/bar/')).url
https://api.test.com/v1/foo/bar/

Pagination#

Tools for creating paginated queries

New in version 1.2.

class snug.pagination.paginated(query)[source]#

A paginated version of a query. Executing it returns an iterator or async iterator.

If the wrapped query is reusable, the paginated query is also reusable.

Parameters:: query (Query[Pagelike[T]]) – The query to paginate. This query must return a Pagelike object.

Example

def foo_page(...) -> Query[Pagelike[Foo]]  # example query
    ...
    return Page(...)

query = paginated(foo_page(...))

for foo in execute(query):
    ...

async for foo in execute_async(query):
    ...

__execute__(client, auth)[source]#

Execute the paginated query.

Returns:: An iterator yielding page content.
Return type:: Iterator[T]

__execute_async__(client, auth)[source]#

Execute the paginated query asynchronously.

Note

This method does not need to be awaited.

Returns:: An asynchronous iterator yielding page content.
Return type:: AsyncIterator[T]

class snug.pagination.Page(content, next_query=None)[source]#

A simple Pagelike object

Parameters:

content (T) – The page content.
next_query (Query[Pagelike[T]]] or None) – The query to retrieve the next page.

class snug.pagination.Pagelike[source]#

Abstract base class for page-like objects. Any object implementing the attributes content and next_query implements this interface. A query returning such an object may be paginated.

Note

Pagelike is a Generic. This means you may write Pagelike[<type-of-content>] as a descriptive type annotation.

For example: Pagelike[List[str]] indicates a page-like object whose content is a list of strings.

abstract property content#

The contents of the page.

Returns:: The page content.
Return type:: T

abstract property next_query#

The query to retrieve the next page, or None if there is no next page.

Returns:: The next query.
Return type:: Query[Pagelike[T]]] or None

Clients#

Funtions for dealing with for HTTP clients in a unified manner

snug.clients.send(client, request)[source]#

snug.clients.send(opener: OpenerDirector, req, **kwargs)

snug.clients.send(session: Session, req)

Given a client, send a Request, returning a Response.

A singledispatch() function.

Parameters:

client (any registered client type) –
The client with which to send the request.

Client types registered by default:
- urllib.request.OpenerDirector (e.g. from build_opener())
- requests.Session (if requests is installed)
request (Request) – The request to send

Returns:

the resulting response

Return type:

Response

Example of registering a new HTTP client:

>>> @send.register(MyClientClass)
... def _send(client, request: Request) -> Response:
...     r = client.send(request)
...     return Response(r.status, r.read(), headers=r.get_headers())

snug.clients.send_async(client, request)[source]#

snug.clients.send_async(_: None, req, *, timeout=10, max_redirects=10)

Given a client, send a Request, returning an awaitable Response.

A singledispatch() function.

Parameters:

client (any registered client type) –
The client with which to send the request.

Client types supported by default:
- asyncio.AbstractEventLoop (e.g. from get_event_loop())
- aiohttp.ClientSession (if aiohttp is installed)
request (Request) – The request to send

Returns:

the resulting response

Return type:

Response

Example of registering a new HTTP client:

>>> @send_async.register(MyClientClass)
... async def _send(client, request: Request) -> Response:
...     r = await client.send(request)
...     return Response(r.status, r.read(), headers=r.get_headers())