Part4: Related Field Resolution#

In many web applications not everything fits in the same data types or structures. We usually have a relation from one thing to another and that relation can be either hard or soft. For instance in a relational database you might have a foreign key from a parent to their child. That would be a hard relation and it is easy to reason about and code up. However what if your manager asked you to find all the cousins of a given person, well that is not so simple, the data is there but your gonna have to do a bunch of work to get the answer.

Cannula has built into the codegen logic the concept of Related fields. These are fields you want to expose on the model but you don’t have an easy attribute to lookup. While it doesn’t make the work of finding cousins easier, it does make it possible to do in an easy to follow structure. We already saw this in the generated Python, all the Query and Mutations are examples of this. Any field that has arguments will be rendered as a function on the base type. In the case of SQLAlchemy types if any field’s return type is a SQLAlchemy model Cannula will automatically add a resolver for it.

In some cases where there is no clear foreign key relation we must provide a where clause to preform the query. This uses a text query to allow a great deal of flexibility to write queries. Then Cannula will use bindparams to map all the graphql arguments plus any defined by the directive. For example the graphql query arguments might include a secondary rule but the primary lookup would be the id of the parent.

type User {
    id: ID!
    # imaging you have a lookup that finds the users best bestFriend
    # by some sort of ranking system
    bestFriends(ranking: Int): Friend
        @field_meta(
            # Query the 'friends' table sorted by rank
            where: "user_id = :id AND ranking >= :ranking ORDER BY ranking ASC",
            # "id" here is the id of the user aka `self.id` since this is a field
            # on the user object it does not make since to require it in the graphql
            # query but we have to tell the system which field to use to relate them.
            args: ["id"],
        )
}

Starting with the schema from part3, we have added a new type ‘Quota’ that is has a foreign_key to the ‘User’ type for a simple example of how the relations work:

scalar UUID

type User @db_sql(table_name: "users_part4") {
  id: UUID! @field_meta(primary_key: true)
  name: String!
  email: String! @field_meta(index: true, unique: true)
  quota: [Quota!] @field_meta(where: "user_id = :id", args: ["id"])
  overQuota(resource: String!): Quota
    @field_meta(where: "user_id = :id AND resource = :resource", args: ["id"])
}

type Quota @db_sql(table_name: "quotas_part4") {
  id: UUID! @field_meta(primary_key: true)
  user_id: UUID! @field_meta(foreign_key: "users_part4.id")
  user: User
  resource: String!
  limit: Int!
  count: Int!
}

type Query {
  people: [User]
  person(id: UUID!): User @field_meta(where: "id = :id")
}

Now when we run cannula codegen we get these types and sql:

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 Quota(ABC):
    __typename = "Quota"
    id: UUID
    user_id: UUID
    resource: str
    limit: int
    count: int

    async def user(self, info: ResolveInfo["Context"]) -> Optional[User]:
        return await info.context.users.get_model_by_pk(self.user_id)


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

    async def quota(self, info: ResolveInfo["Context"]) -> Optional[Sequence[Quota]]:
        return await info.context.quotas.user_quota(id=self.id)

    async def overQuota(
        self, info: ResolveInfo["Context"], resource: str
    ) -> Optional[Quota]:
        return await info.context.quotas.user_overQuota(id=self.id, resource=resource)


class peopleQuery(Protocol):

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


class personQuery(Protocol):

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


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

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


class Base(DeclarativeBase):
    pass


class DBQuota(Base):
    __tablename__ = "quotas_part4"
    id: Mapped[UUID] = mapped_column(primary_key=True)
    user_id: Mapped[UUID] = mapped_column(
        foreign_key=ForeignKey("users_part4.id"), nullable=False
    )
    resource: Mapped[str] = mapped_column(nullable=False)
    limit: Mapped[int] = mapped_column(nullable=False)
    count: Mapped[int] = mapped_column(nullable=False)


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

You can see that we have foreign_key relations in the db and our types now have resolvers that call the context which is where all the magic happens. let’s look at the context that was generated to see our custom queries in action:

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


class QuotaDatasource(
    DatabaseRepository[DBQuota, Quota], graph_model=Quota, db_model=DBQuota
):

    async def user_quota(self, id: UUID) -> Optional[Sequence[Quota]]:
        return await self.get_models(text("user_id = :id").bindparams(id=id))

    async def user_overQuota(self, id: UUID, resource: str) -> Optional[Quota]:
        return await self.get_model(
            text("user_id = :id AND resource = :resource").bindparams(
                id=id, resource=resource
            )
        )


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())

    async def query_person(self, id: UUID) -> Optional[User]:
        return await self.get_model(text("id = :id").bindparams(id=id))


class Context(BaseContext):
    quotas: QuotaDatasource
    users: UserDatasource

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

Now we just need to add in the new query we added to the RootType:

import pathlib
import uuid
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()


async def resolve_person(
    info: cannula.ResolveInfo[Context],
    id: uuid.UUID,
) -> Optional[User]:
    return await info.context.users.query_person(id=id)


root_value: RootType = {
    "people": resolve_people,
    "person": resolve_person,
}

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

You can now update the route to http://localhost:8000/part4/graph and try the following query:

query People {
    people {
        name
        email
        quota {
            resource
            limit
        }
        overQuota(resource: "water") {
            count
        }
    }
}

And get results like this:

{
    "data": {
        "people": [
        {
            "name": "Normal User",
            "email": "user@email.com",
            "quota": [
            {
                "resource": "fire",
                "limit": 10
            },
            {
                "resource": "water",
                "limit": 15
            }
            ],
            "overQuota": {
            "count": 4
            }
        },
        {
            "name": "Admin User",
            "email": "admin@example.com",
            "quota": [
            {
                "resource": "fire",
                "limit": 5
            }
            ],
            "overQuota": null
        }
        ]
    },
    "errors": null,
    "extensions": null
}

Part4: Related Field Resolution#

This Page