Recipes#

This page features some interesting recipes for common use-cases.

GraphQL#

A rudimentary example using the Github GraphQL API v4, similar to the repo query in the tutorial:

from gentools import relay

@relay
def graphql(request: str):
    """decorator for GraphQL requests"""
    response = yield snug.POST('https://api.gitub.com/graphql',
                               content=json.dumps({'query': request}))
    return json.loads(response.content)['data']


@graphql
def repo(name, owner, *, fields=('id', )) -> snug.Query[dict]:
    """lookup a repo by owner and name, returning only certain fields"""
    response = yield f'''
      query {
        repository(owner: "{owner}", name: "{name}") {
           {"\n".join(fields)}
        }
      }
    '''
    return response['repository']

q = repo('Hello-World', owner='octocat', fields=('description', 'id'))
snug.execute(q)
# {'description': ..., 'id': ...}

Conditional requests#

Many APIs support conditional requests with If-None-Match or If-Modified-Since headers. The example below shows a reusable implementation using relay:

from gentools import relay

class NotModified(Exception):
    pass

@relay
def if_modified_since(req):
    """decorator for queries supporting 'If-Modified-Since' headers"""
    resp = yield req
    if 'If-Modified-Since' in req.headers and resp.status_code == 304:
        raise NotModified
    return resp

@if_modified_since
def repo(name, owner, modified_since=None):
    """lookup a repo, or raise NotModified"""
    response = yield snug.GET(
        f'https://api.github.com/repos/{owner}/{name}',
        headers=({'If-Modified-Since': modified_since}
                 if modified_since else {}))
    return json.loads(response.content)

q = repo('Hello-World', 'octocat', modified_since='2018-02-08T00:30:01Z')
snug.execute(q)
# NotModified()

Testing#

Because queries are generators, we can easily write unittests that don’t touch the network.

Here is an annotated example of testing the example gitub repo query:

from gentools import sendreturn

def test_repo():
    # iter() ensures this works for function- and class-based queries
    query = iter(repo('Hello-World', owner='octocat'))

    # check the request is OK
    request = next(query)
    assert request.url.endswith('repos/octocat/Hello-World')

    # construct our test response
    response = snug.Response(200, b'...<test response content>...')

    # getting the return value of a generator requires
    # catching StopIteration.
    # the following shortcut with `sendreturn` is equivalent to:
    #
    # try:
    #     query.send(response)
    # except StopIteration as e:
    #     result = e.value
    # else:
    #     raise RuntimeError('generator did not return')
    result = sendreturn(query, response)

    # check the result is OK
    assert result['description'] == 'My first repository on github!'

The slack and NS API tests show real-world cases for this.

Django-like querysets#

Class-based queries can be used to create a queryset-like API. We can use github’s issues endpoint to illustrate:

import snug

class issues(snug.Query):
    """select assigned issues within an organization"""

    def __init__(self, org, state='open', labels='', sort='created',
                 direction='desc', since=None):
        self.org = org
        self.params = {
            'state': state,
            'labels': labels,
            'sort': sort,
            'direction': direction,
        }
        if since:
            self.params['since'] = since

    def filter(self, state=None, labels=None):
        updated = self.params.copy()
        if state is not None: updated['state'] = state
        if labels is not None: updated['labels'] = labels
        return issues(self.org, **updated)

    def ascending(self):
        return issues(self.org, **{**self.params, 'direction': 'asc'})

    def sort_by(self, sort):
        return issues(self.org, **{**self.params, 'sort': sort})

    def __iter__(self):
        req = snug.GET(f'https://api.github.com/orgs/{self.org}/issues',
                       params=self.params)
        resp = yield req
        return json.loads(resp.content)

The resulting query class can be used as follows:

>>> my_query = (issues(org='github')
...            .filter(state='all')
...            .filter(labels='bug,ui')
...            .sort_by('updated')
...            .ascending())
...
>>> snug.execute(my_query, auth=('me', 'password'))
[{"number": ..., ...}, ...]

Method chaining#

With the following helper class, it is possible to access all query functionality by method chaining:

import snug

class Explorer:

    def __init__(self, obj, *, executor=snug.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 an attribute of the underlying object, wrapped"""
        return Explorer(getattr(self.__wrapped__, name),
                        executor=self._executor)

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

    def __call__(self, *args, **kwargs):
        """call the underlying object, wrapping the result"""
        return Explorer(self.__wrapped__(*args, **kwargs),
                        executor=self._executor)

    def paginated(self):
        """make the wrapped query paginated"""
        return Explorer(snug.paginated(self.__wrapped__))

This allows us to write expressions like this:

import github

bound_ghub = Explorer(github, executor=...)
issues = (bound_ghub.repo('Hello-World', owner='octocat')
          .issues(state='closed')
          .paginated()
          .execute())

# instead of:
issues = snug.execute(snug.paginated(
    my_github.repo('Hello-World', owner='octocat')
    .issues(state='closed')))