Resolvers#

Resolvers define how to fetch the data defined in your schema. There are many ways to access your data but resolvers are just a simple async function that return either an object or a dictionary that matches the shape of the Schema return type.

This could be the raw json response from a third party library or it could be a database model that represents your data. The great thing about resolvers are they are completely flexible as long as the type you are returning is correct. Which means you can combine loosly related items into a single parent.

Using the API#

class cannula.api.CannulaAPI(schema, context=None, middleware=[], root_value=None, scalars=[], logger=None, level=10, **kwargs)#

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

import cannula

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

@api.resolver('Query')
def hello(who):
    return f'Hello {who}!'

Parameters:

schema (str | DocumentNode | Path) – GraphQL Schema for this resolver. This can either be a str or pathlib.Path object.
context (Type[Context] | None (default: None)) – Context class to hold shared state, added to GraphQLResolveInfo object.
middleware (List[Any] (default: [])) – List of middleware to enable.
root_value (TypeVar(RootType, covariant=True) | None (default: None)) –
Mapping of operation names to resolver functions. This can be a TypedDict object that is generated by the codegen. Type hints are available if you initialize the api with this type like so:
```
class MyRootType(TypedDict, total=False):
    hello: str

graph_api = cannula.CannulaAPI[MyRootType](
    root_value={"hello": "hi"}
)
```
scalars (List[ScalarInterface] (default: [])) – List of custom scalars to attach to the graph schema. These all should be a subclass of the base cannula.scalars.ScalarType.
logger (Logger | None (default: None)) – Optional logger to use for messages that cannula logs. The default will be: cannula.api
level (int (default: 10)) – Optional logging level to log as (default: DEBUG)
kwargs – Any extra kwargs passed directly to graphql.execute function.

async call(document, *, variables=None, operation_name=None, context=None, request=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.

Parameters:

document (DocumentNode | str) – The query or mutation to execute.
variables (Dict[str, Any] | None (default: None)) – Dictionary of variable values.
operation_name (str | None (default: None)) – The named operation this can be used to cache queries.
context (Any | None (default: None)) – The context instance to use for this operation.
request (Any | None (default: None)) – The original request instance for the query, this is used when no context is passed. By default it will be set on the info object: info.context.request

Return type:

ExecutionResult

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.CannulaAPI(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 (str | None (default: None)) – Field name to resolve, by default the function name will be used.
Return type:: Any

parse_document(document)#

Parse and store the document in lru_cache.

Parameters:: document (str)
Return type:: ParseResults

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.CannulaAPI(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 (str | None (default: None)) – 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.CannulaAPI(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 (str | None (default: None)) – Field name to resolve, by default the function name will be used.

Return type:

Any

async subscribe(document, *, variables=None, operation_name=None, context=None, request=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.

Parameters:

document (DocumentNode | str) – The query or mutation to execute.
variables (Dict[str, Any] | None (default: None)) – Dictionary of variable values.
operation_name (str | None (default: None)) – The named operation this can be used to cache queries.
context (Any | None (default: None)) – The context instance to use for this operation.
request (Any | None (default: None)) – The original request instance for the query, this is used when no context is passed. By default it will be set on the info object: info.context.request

Return type:

AsyncIterable[ExecutionResult] | ExecutionResult

validate(document)#

Validate the document against the schema and store results in lru_cache.

Parameters:: document (DocumentNode)
Return type:: List[GraphQLError]

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

Parameters:

document_ast (DocumentNode)
errors (List[GraphQLError])

document_ast: DocumentNode#: Alias for field number 0

errors: List[GraphQLError]#: Alias for field number 1

Resolvers#

Using the API#

This Page