Cannula Documentation#

Using GraphQL you can simplify your web application stack and reduce dependencies to achieve the same customer experience without regret. By using just a few core libraries you can increase productivity and make your application easier to maintain.

Our Philosophy:

Make your site easy to maintain.
Document your code.
Don’t lock yourself into a framework.
Be happy!

Listen to me talk about GraphQL:

Installation#

Note

Basic Requirements:

Python 3.8+
graphql-core

Using pip:

$ pip3 install cannula

Quick Start#

Here is a small hello world example:

import asyncio
import typing
import sys

import cannula

SCHEMA = """
    type Query {
        hello(who: String!): String
    }
"""

# Basic API setup with the schema we defined
api = cannula.API(schema=SCHEMA)


# The query resolver takes a `source` and `info` objects
# and any arguments defined by the schema. Here we
# only accept a single argument `who`.
@api.query()
async def hello(
    source: typing.Any,
    info: cannula.ResolveInfo,
    who: str,
) -> str:
    # Here the field_name is 'hello' so we'll
    # return 'hello {who}!'
    return f"{info.field_name} {who}!"


# Pre-parse your query to speed up your requests.
SAMPLE_QUERY = cannula.gql(
    """
    query HelloWorld ($who: String!) {
        hello(who: $who)
    }
"""
)


async def run_hello(who: str = "world"):
    results = await api.call(SAMPLE_QUERY, variables={"who": who})
    print(results.data)
    return results.data


if __name__ == "__main__":
    who = "world"
    if len(sys.argv) > 1:
        who = sys.argv[1]

    loop = asyncio.get_event_loop()
    loop.run_until_complete(run_hello(who))

Dataloaders#

Dataloaders help avoid the N+1 error. They also help with caching and reducing boilerplate. Here is an example using the HTTP Data Source which can be used to query a remote http resource:

import logging
import typing

import cannula
import fastapi
import httpx
from cannula.context import Context, ResolveInfo
from cannula.datasource import http


logging.basicConfig(level=logging.DEBUG)

# For demo purposes we'll just use a local asgi application
remote_app = fastapi.FastAPI()


@remote_app.get("/widgets")
async def get_widgets():
    return [{"name": "hammer", "type": "tool"}]


# Create a httpx client that responds with the 'remote_app'
client = httpx.AsyncClient(app=remote_app)


SCHEMA = cannula.gql(
    """
    type Widget {
        name: String
        type: String
    }

    type Query {
        widgets: [Widget]
    }

"""
)


# Our actual datasource object
class WidgetDatasource(http.HTTPDataSource):
    # set our base url to work with the demo fastapi app
    base_url = "http://localhost"

    async def get_widgets(self):
        return await self.get("/widgets")


# Create a custom context and add the datasource
class CustomContext(Context):
    widget_datasource: WidgetDatasource

    def handle_request(self, request: typing.Any) -> typing.Any:
        # Initialize the datasource using the request and
        # set the client to use the demo client app
        self.widget_datasource = WidgetDatasource(request, client=client)
        return request


api = cannula.API(schema=SCHEMA, context=CustomContext)


@api.query("widgets")
async def list_widgets(parent, info: ResolveInfo[CustomContext]):
    return await info.context.widget_datasource.get_widgets()


async def main():
    # This does the same query twice to show the results are cached
    query = cannula.gql(
        """
        query Widgets {
            widgets: widgets {
                name
                type
            }
            another: widgets {
                name
                type
            }
        }
    """
    )

    results = await api.call(query)
    print(results.data, results.errors)
    return results.data


if __name__ == "__main__":
    import asyncio

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

The output looks like this:

DEBUG:cannula.schema:Adding default empty Mutation type
DEBUG:asyncio:Using selector: EpollSelector
DEBUG:cannula.datasource.http:cache set for GET 'http://localhost/widgets'
DEBUG:cannula.datasource.http:cache found for GET 'http://localhost/widgets'
INFO:httpx:HTTP Request: GET http://localhost/widgets "HTTP/1.1 200 OK"
{
    'widgets': [{'name': 'hammer', 'type': 'tool'}],
    'another': [{'name': 'hammer', 'type': 'tool'}]
}

Notice the second request is cached since the datasource already resovled it. This cache is only stored for single GraphQL request. If you want to persist that for longer you’ll need to implement that yourself.

Testing Your Code#

Since GraphQL is typed it is trivial to mock the responses to any Query or Mutation. Cannula provides a MockMiddleware which can mock all types or only select few to provide flexibility when writing your tests.

Cannula Documentation#

Installation#

Quick Start#

Dataloaders#

Testing Your Code#

Read More About It#