Schema and Resolvers#

Cannula levelages schema first design to generate schema. This allows your schema to have maximum portability and also is very easy to read and document.

Using the API#

class cannula.api.API(schema, context=None, middleware=[], **kwargs)#

Your entry point into the fun filled world of graphql. Just dive right in:

import cannula

api = cannula.API(schema='''
    extend type Query {
        hello(who: String): String
    }
''')

@api.resolver('Query')
def hello(who):
    return f'Hello {who}!'
Parameters:
  • schema (Union[str, DocumentNode, Path]) – GraphQL Schema for this resolver. This can either be a str or pathlib.Path object.

  • context (Optional[Any]) – Context class to hold shared state, added to GraphQLResolveInfo object.

  • middleware (List[Any]) – List of middleware to enable.

async call(document, request=None, variables=None)#

Preform a query against the schema.

This is meant to be called in an asyncio.loop, if you are using a web framework that is synchronous use the call_sync method.

Return type:

ExecutionResult

include_resolver(resolver)#

Include a set of resolvers

This is used to break up a larger application into different modules. For example you can group all the resolvers for a specific feature set:

from cannula import Resolver

from database import Book, filter_books, get_book

book_resolver = Resolver()

@book_resolver.query
async def books(parent, info, **args) -> list[Book]:
    return await filter_books(**args)

@book_resolver.query
async def book(parent, info, book_id: str) -> Book:
    return await get_book(book_id)

Then include the resolver in the main cannula API:

import cannula
import pathlib

from features.books import book_resolver

api = cannula.API(schema=pathlib.Path('.'))
api.include_resolver(book_resolver)
mutation(field_name=None)#

Mutation Resolver

Short cut to add a resolver for a mutation, by default it will use the name of the function as the field_name to be resolved:

api = cannula.API(schema="type Mutation { make_it(name: String): String }")

@api.mutation
async def make_it(parent, info, name: str):
    return "hello world"

@api.mutation(field_name="make_it")
async def some_other_something(parent, info, name: str):
    return "override the function name"
Parameters:

field_name (Optional[str]) – Field name to resolve, by default the function name will be used.

Return type:

Any

query(field_name=None)#

Query Resolver

Short cut to add a resolver for a query, by default it will use the name of the function as the field_name to be resolved:

api = cannula.API(schema="type Query { something: String }")

@api.query
async def something(parent, info):
    return "hello world"

@api.query(field_name="something")
async def some_other_something(parent, info):
    return "override the function name"
Parameters:

field_name (Optional[str]) – Field name to resolve, by default the function name will be used.

Return type:

Any

resolver(type_name, field_name=None)#

Field Resolver

Add a field resolver for a given type, by default it will use the name of the function as the field_name to be resolved:

api = cannula.API(schema="type Book { name: String }")

@api.resolver("Book")
async def name(parent, info):
    return "hello world"

@api.resolver("Book", field_name="something")
async def some_other_something(parent, info):
    return "override the function name
Parameters:
  • type_name (str) – Parent object type name that is being resolved.

  • field_name (Optional[str]) – Field name to resolve, by default the function name will be used.

Return type:

Any

class cannula.api.ParseResults(document_ast, errors)#
document_ast: DocumentNode#

Alias for field number 0

errors: List[GraphQLError]#

Alias for field number 1

class cannula.api.Resolver#

This class is a helper to organize your project as it grows. It allows you to put your resolver modules and schema in different packages. For example:

app/
    api.py     # `api = cannula.API(args)`
resolvers/
    books.py   # `books = cannula.Resolver()`
    movies.py  # `movies = cannula.Resolver()`

You then register resolvers and dataloaders in the same way:

resolvers/books.py:

import cannula

from database.books import get_books

books = cannula.Resolver()

@books.query('books')
async def get_books(source, info, args):
    return await get_books()

resolvers/moives.py:

import cannula

from database.movies import get_movies, fetch_movies_for_book

movies = cannula.Resolver()

@movies.query('movies')
async def get_movies(source, info, args):
    return await get_movies()

@movies.revolver('Books', 'movies')
async def list_movies_for_book(book, info) -> list[Movie]:
    return await fetch_movies_for_book(book.id)

app/api.py:

import cannula

from resolvers.books import books
from resolvers.movies import movies

api = cannula.API(schema=SCHEMA)
api.include_resolver(books)
api.include_resolver(movies)
mutation(field_name=None)#

Mutation Resolver

Short cut to add a resolver for a mutation, by default it will use the name of the function as the field_name to be resolved:

resolver = cannula.Resolver()

@resolver.mutation
async def make_it(parent, info, name: str) -> str:
    return "hello world"

@resolver.mutation(field_name="make_it")
async def some_other_something(parent, info, name: str) -> str:
    return "override the function name"
Parameters:

field_name (Optional[str]) – Field name to resolve, by default the function name will be used.

Return type:

Any

query(field_name=None)#

Query Resolver

Short cut to add a resolver for a query, by default it will use the name of the function as the field_name to be resolved:

resolver = cannula.Resolver()

@resolver.query
async def something(parent, info) -> str:
    return "hello world"

@resolver.query(field_name="something")
async def some_other_something(parent, info) -> str:
    return "override the function name"
Parameters:

field_name (Optional[str]) – Field name to resolve, by default the function name will be used.

Return type:

Any

resolver(type_name, field_name=None)#

Field Resolver

Add a field resolver for a given type, by default it will use the name of the function as the field_name to be resolved:

resolver = cannula.Resolver()

@resolver.resolver("Book")
async def name(parent, info) -> str:
    return "hello world"

@resolver.resolver("Book", field_name="something")
async def some_other_something(parent, info):
    return "override the function name
Parameters:
  • type_name (str) – Parent object type name that is being resolved.

  • field_name (Optional[str]) – Field name to resolve, by default the function name will be used.

Return type:

Any