Part3: Code Generation#

One of the best benefits to using a schema first approach is that it makes it incredibly easy to parse and generate code from this well defined schema. Cannula offers a simple way to generate data classes, models, and resolver definitions to assist with type hints and making sure your code is in sync with the schema.

First we need to add some details to the pyproject.toml for your application:

[tool.cannula.codegen]
scalars = ["cannula.scalars.util.UUID"]
dest = "gql"
schema = "schema.graphql"

You’ll notice we are including some custom Scalars in this config, we can use these in our schema file to alter the input/output of certain fields. Cannula provides some basic ones that are useful but you can add your own via this same config. Here is our schema that is using the UUID scalar. The UUID Scalar will convert a string into a UUID for output and convert a string to a UUID on input.

Next in order to create SQLAlchemy models from the schema file we’ll need to add metadata directives to the types in our schema. There are two main directives at the moment:

db_sql
field_meta

These directives will indicate to cannula how to generate type classes and relations. For db_sql this indicates we wish to create a SQLAlchemy model, which by default will use the pluralized lowercase of the type name as the db table. For field_meta there are many more options as fields can represent data or related items, there are also arguments and resolver functions.

Here is our simplied schema to demostrate some of the basic options:

scalar UUID

type User @db_sql(table_name: "users_part3") {
  id: UUID! @field_meta(primary_key: true)
  name: String!
  email: String @field_meta(index: true, unique: true)
}

type Query {
  people: [User]
}

Now we just need to run the codegen command in this folder to generate the base types:

$ cannula codegen

This will create the gql folder that we set in the pyproject.toml and add the following files.

types.py#

from __future__ import annotations
from abc import ABC
from cannula import ResolveInfo
from dataclasses import dataclass
from typing import Optional, Protocol, Sequence, TYPE_CHECKING
from typing_extensions import TypedDict
from uuid import UUID

if TYPE_CHECKING:
    from .context import Context


@dataclass(kw_only=True)
class User(ABC):
    __typename = "User"
    id: UUID
    name: str
    email: Optional[str] = None


class peopleQuery(Protocol):

    async def __call__(
        self, info: ResolveInfo["Context"]
    ) -> Optional[Sequence[User]]: ...


class RootType(TypedDict, total=False):
    people: Optional[peopleQuery]

For each type in our schema cannula generates a dataclass with all the simple fields as class vars. The related fields that reference another type or have arguments are rendered as async resolver functions. These functions call the corresponding datasource on the context object to retrieve the results.

Then it creates Protocols for the operation resolvers with the correct signature. For example the peopleQuery will return a list of User this will ensure that our RootValue will have the correct signature and return values and our editor will highlight these errors.

sql.py#

from __future__ import annotations
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from typing import Optional
from uuid import UUID


class Base(DeclarativeBase):
    pass


class DBUser(Base):
    __tablename__ = "users_part3"
    id: Mapped[UUID] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(nullable=False)
    email: Mapped[Optional[str]] = mapped_column(index=True, unique=True, nullable=True)

The sql file contains all the database table definitions that we have defined. For a full reference please refer to the Code Generation

context.py#

from __future__ import annotations
from cannula.context import Context as BaseContext
from cannula.datasource.orm import DatabaseRepository
from sqlalchemy import true
from sqlalchemy.ext.asyncio import async_sessionmaker
from typing import Optional, Sequence
from .sql import DBUser
from .types import User


class UserDatasource(
    DatabaseRepository[DBUser, User], graph_model=User, db_model=DBUser
):

    async def query_people(self) -> Optional[Sequence[User]]:
        return await self.get_models(true())


class Context(BaseContext):
    users: UserDatasource

    def __init__(self, session_maker: async_sessionmaker):
        self.users = UserDatasource(session_maker)

The context is added to all the resolvers and is a way to share datasources between all the functions that are resolving data for a given query. Here have the class UserDatasource that maps the types to the User type and the DBUser table. The Context object exposes these and adds any initialization the datasources need.

Wire everything up#

With most of the code generated for us all we have to do is connect these pieces to our graph and FastAPI.

First we’ll add a bit of code to create the new tables and add some test data:

# in dashboard.core.seed_data
async def add_part3_users(session) -> list[uuid.UUID]:
    from dashboard.part3.gql.context import UserDatasource

    users = UserDatasource(session)
    user_id = uuid.uuid4()
    await users.add(id=user_id, name="test", email="sam@ex.com")
    another_id = uuid.uuid4()
    await users.add(id=another_id, name="another", email="sammie@ex.com")
    return [user_id, another_id]

We will call this in our tests to seed the database. Then we have to provide a couple resolvers that were not autogenerated for us and create a CannulaAPI instance:

import pathlib
from typing import Sequence, Optional

import cannula
from cannula.scalars.util import UUID

from ..core.config import config
from .gql.types import User, RootType
from .gql.context import Context


async def resolve_people(
    info: cannula.ResolveInfo[Context],
) -> Optional[Sequence[User]]:
    return await info.context.users.query_people()


# The RootType object generated in `gql/types.py` will warn us if we use
# a resolver with an incorrect signature
root_value: RootType = {
    "people": resolve_people,
}

cannula_app = cannula.CannulaAPI[RootType](
    schema=pathlib.Path(config.root / "part3"),
    scalars=[
        UUID,
    ],
    root_value=root_value,
)

Finally we need to connect our application to an endpoint so we can access this. We will use cannula contrib dependency for FastAPI that will handle converting the request body into a graphql request (query, variables, operationName). This dependency returns a callable which we can use to inject our custom context.

from typing import Annotated

from cannula.contrib.asgi import GraphQLDepends, ExecutionResponse, GraphQLExec
from fastapi import APIRouter, Depends

from dashboard.core.config import config
from .gql.context import Context
from .graph import cannula_app

part3 = APIRouter(prefix="/part3")


@part3.post("/graph")
async def part3_root(
    graph_call: Annotated[
        GraphQLExec,
        Depends(GraphQLDepends(cannula_app)),
    ],
) -> ExecutionResponse:
    # create a context instance for our resolvers to use
    context = Context(config.session)
    return await graph_call(context=context)

We can use the apollo sandbox to test this out, first run the following:

$ make initdb
$ make addusers
$ make run

This will start up the application locally with a few test users next go to the Apollo sandbox:

https://studio.apollographql.com/sandbox/explorer/

Change the connection url to http://localhost:8000/part3/graph

Once it loads you should see your schema and you can try the following query:

query ExampleQuery {
    people {
        email
        id
    }
}

This should return something like this:

{
    "data": {
        "people": [
            {
                "email": "user@email.com",
                "id": "683f89e1-b9e2-4af8-bb7e-7b2bccfe54a3",
            }
        ]
    },
    "errors": null,
    "extensions": null
}

Great this is looking better, now we need to explore Part4: Related Field Resolution

Part3: Code Generation#

types.py#

sql.py#

context.py#

Wire everything up#

This Page