ariebovenberg/snug

🧣 Write reusable web API interactions

Snug 🧣

Snug is a tiny toolkit for writing reusable interactions with web APIs. Key features:

• Write once, run with different HTTP clients (sync and async)
• Fits any API architecture (e.g. REST, RPC, GraphQL)
• Simple, lightweight and versatile

Why?

Writing reusable web API interactions is difficult. Consider a generic example:

import json

def repo(name, owner):
    """get a github repo by owner and name"""
    request = Request(f'https://api.github.com/repos/{owner}/{name}')
    response = my_http_client.send(request)
    return json.loads(response.content)

Nice and simple. But...

• What about async? Do we write another function for that?
• How do we write clean unittests for this?
• What if we want to use another HTTP client or session?
• How do we use this with different credentials?

Snug allows you to write API interactions independent of HTTP client, credentials, or whether they are run (a)synchronously.

In contrast to most API client toolkits, snug makes minimal assumptions and design decisions for you. Its simple, adaptable foundation ensures you can focus on what makes your API unique. Snug fits in nicely whether you're writing a full-featured API wrapper, or just making a few API calls.

Quickstart

1. API interactions ("queries") are request/response generators.

import snug

def repo(name, owner):
    """get a github repo by owner and name"""
    request = snug.GET(f'https://api.github.com/repos/{owner}/{name}')
    response = yield request
    return json.loads(response.content)

2. Queries can be executed:

>>> query = repo('Hello-World', owner='octocat')
>>> snug.execute(query)
{"description": "My first repository on Github!", ...}

Features

1. Effortlessly async. The same query can also be executed asynchronously:

   query = repo('Hello-World', owner='octocat')
   repo = await snug.execute_async(query)

2. Flexibility. Since queries are just generators, customizing them requires no special glue-code. For example: add validation logic, or use any serialization method:

   from my_types import User, UserSchema
   
   def user(name: str) -> snug.Query[User]:
       """lookup a user by their username"""
       if len(name) == 0:
           raise ValueError('username must have >0 characters')
       request = snug.GET(f'https://api.github.com/users/{name}')
       response = yield request
       return UserSchema().load(json.loads(response.content))

3. Pluggable clients. Queries are fully agnostic of the HTTP client. For example, to use requests instead of the standard library:

   import requests
   query = repo('Hello-World', owner='octocat')
   snug.execute(query, client=requests.Session())

   Read here how to register your own.

4. Testability. Queries can easily be run without touching the network. No need for complex mocks or monkeypatching.

   >>> query = repo('Hello-World', owner='octocat')
   >>> next(query).url.endswith('/repos/octocat/Hello-World')
   True
   >>> query.send(snug.Response(200, b'...'))
   StopIteration({"description": "My first repository on Github!", ...})

5. Swappable authentication. Queries aren't tied to a session or credentials. Use different credentials to execute the same query:

   def follow(name: str) -> snug.Query[bool]:
       """follow another user"""
       req = snug.PUT('https://api.github.com/user/following/{name}')
       return (yield req).status_code == 204
   
   snug.execute(follow('octocat'), auth=('me', 'password'))
   snug.execute(follow('octocat'), auth=('bob', 'hunter2'))

6. Related queries. Use class-based queries to create an expressive, chained API for related objects:

   class repo(snug.Query[dict]):
       """a repo lookup by owner and name"""
       def __init__(self, name, owner): ...
   
       def __iter__(self): ...  # query for the repo itself
   
       def issue(self, num: int) -> snug.Query[dict]:
           """retrieve an issue in this repository by its number"""
           r = snug.GET(f'/repos/{self.owner}/{self.name}/issues/{num}')
           return json.loads((yield r).content)
   
   my_issue = repo('Hello-World', owner='octocat').issue(348)
   snug.execute(my_issue)

7. Pagination. Define paginated queries for (asynchronous) iteration.

   def organizations(since: int=None):
       """retrieve a page of organizations since a particular id"""
       resp = yield snug.GET('https://api.github.com/organizations',
                             params={'since': since} if since else {})
       orgs = json.loads(resp.content)
       next_query = organizations(since=orgs[-1]['id'])
       return snug.Page(orgs, next_query=next_query)
   
   my_query = snug.paginated(organizations())
   
   for orgs in snug.execute(my_query):
       ...
   
   # or, with async
   async for orgs in snug.execute_async(my_query):
       ...

8. Function- or class-based? You decide. One option to keep everything DRY is to use class-based queries and inheritance:

   class BaseQuery(snug.Query):
       """base github query"""
   
       def prepare(self, request): ...  # add url prefix, headers, etc.
   
       def __iter__(self):
           """the base query routine"""
           request = self.prepare(self.request)
           return self.load(self.check_response((yield request)))
   
       def check_response(self, result): ...  # raise nice errors
   
   class repo(BaseQuery):
       """get a repo by owner and name"""
       def __init__(self, name, owner):
           self.request = snug.GET(f'/repos/{owner}/{name}')
   
       def load(self, response):
           return my_repo_loader(response.content)
   
   class follow(BaseQuery):
       """follow another user"""
       def __init__(self, name):
           self.request = snug.PUT(f'/user/following/{name}')
   
       def load(self, response):
           return response.status_code == 204

   Or, if you're comfortable with higher-order functions and decorators, make use of gentools to modify query yield, send, and return values:

   from gentools import (map_return, map_yield, map_send,
                         compose, oneyield)
   
   class Repository: ...
   
   def my_repo_loader(...): ...
   
   def my_error_checker(...): ...
   
   def my_request_preparer(...): ...  # add url prefix, headers, etc.
   
   basic_interaction = compose(map_send(my_error_checker),
                               map_yield(my_request_preparer))
   
   @map_return(my_repo_loader)
   @basic_interaction
   @oneyield
   def repo(owner: str, name: str) -> snug.Query[Repository]:
       """get a repo by owner and name"""
       return snug.GET(f'/repos/{owner}/{name}')
   
   @basic_interaction
   def follow(name: str) -> snug.Query[bool]:
       """follow another user"""
       response = yield snug.PUT(f'/user/following/{name}')
       return response.status_code == 204

For more info, check out the tutorial, advanced features, recipes, or examples.

Installation

There are no required dependencies. Installation is easy as:

pip install snug

Although snug includes basic sync and async HTTP clients, you may wish to install requests, httpx, and/or aiohttp.

pip install requests aiohttp httpx

Alternatives

If you're looking for a less minimalistic API client toolkit, check out uplink or tapioca.

snug's Issues

Dependabot couldn't authenticate with https://pypi.python.org/simple/

Dependabot couldn't authenticate with https://pypi.python.org/simple/.

You can provide authentication details in your Dependabot dashboard by clicking into the account menu (in the top right) and selecting 'Config variables'.

View the update logs.

support for other authentication types

currently only basic auth is supported.

executing queries with method chaining

It is possible to implement a method chaining API, with:

class Explorer:

    def __init__(self, obj, *, executor=execute):
        self.__wrapped__ = obj
        self._executor = executor

    def execute(self, **kwargs):
        """execute the wrapped object as a query

        Parameters
        ----------
        **kwargs
            arguments passed to the executor
        """
        return self._executor(self.__wrapped__, **kwargs)

    def __getattr__(self, name):
        return Explorer(getattr(self.__wrapped__, name),
                        executor=self._executor)

    def __repr__(self):
        return f'Explorer({self.__wrapped__!r})'

    def __call__(self, *args, **kwargs):
        return Explorer(self.__wrapped__(*args, **kwargs),
                        executor=self._executor)

Usable like

module = Explorer(my_module)
module.my_query(bla=4).related_query().execute()

only support aiohttp on python 3.5.3+

aiohttp >=3.0 only supports python 3.5.3+. This should be reflected in the docs and tests

docs improvements

some docs improvements:

• QuerySet-pattern recipe
• make readme more clear and to the point

Native support for streaming requests/responses

For some cases, streaming requests and responses could be useful. E.g. downloading a file.

urllib sender raises HTTPError instead of returning Response object

Encountered in ariebovenberg/quiz#10.

For consistency with other HTTP clients, the urllib client should not raise an HTTP error, but simply return the response object.

python 3.7 deprecation warning

/usr/local/var/pyenv/versions/3.7.0/envs/quiz/lib/python3.7/site-packages/snug/http.py:3: DeprecationWarning: Using or importing the ABCs from 'collections' instead of from 'collections.abc' is deprecated, and in 3.8 it will stop working
  from collections import Mapping

Add type annotations

When the library was originally written, python 2 still needed support. Now that it is off the table, we can use proper type annotations 🎉

deprecate `auth_method` parameter in favour of callable `auth`

To make execute and __execute__ more consistent, auth_method should be deprecated. To implement other forms of authentication, callables can then be passed to auth.

This approach is used in the requests library

python2 support

blockers for python2 support:

1. returning values from generators with return is not possible on python2. A hacky solution would have to be devised, such as raising and catching custom exceptions. Simply raising StopIteration is not possible because this breaks on python 3.5+ (PEP479).
2. generator composition is more difficult under python2 without yield from.
3. type annotations would have to be removed
4. async components would have to be moved into a separate module, to be imported only on python3.4+

edit: numbered items

improve builtin asyncio client

possible improvements:

• encode headers as latin-1
• implement automatic redirects (consistent with other clients)
• timeouts

pagination helpers

Currently pagination has to be implemented on a per-API basis. Perhaps it is possible to provide a common abstraction for dealing with paginated queries.

Dependabot couldn't fetch all your path-based dependencies

Dependabot couldn't fetch one or more of your project's path-based Python dependencies. The affected dependencies were requirements/setup.py.

To use path-based dependencies with Dependabot the paths must be relative and resolve to a directory in this project's source code.

You can mention @dependabot in the comments below to contact the Dependabot team.

"escape hatch" for HTTP client-specific features

It is possible that sometimes advanced client-specific features are needed. For example, streaming responses, or multipart data. (see also #16)

proposal: implement a hook to allow full customization of query execution. Although using this will negate some of the benefits of queries, it will ensure all functionality can be supported.

class advanced_query(snug.Query):
    
    def __execute__(self, client, authenticate: t.Callable[[Request], Request]):
        ...  # query logic here...
        return result

    def __execute_async__(self, client, authenticate: t.Callable[[Request], Request]):
        ...  # async query logic here...
        return result

edit: posted too soon

Drop python 3.4 support

Python 3.4 is EOL since 19-03. The code can be cleaned up using async in py3-only code.