Week 21: GraphQL with Strawberry

1. Code-First Schema Design with Strawberry

Where Spring for GraphQL is schema-first — writing SDL, then code to fulfill it — Strawberry is code-first: you define types as ordinary Python dataclasses decorated with @strawberry.type, and Strawberry derives the GraphQL schema directly from your type hints. This should feel familiar — it's the same relationship Pydantic models have to FastAPI's OpenAPI docs, just for a GraphQL schema instead.

import strawberry
from enum import Enum

@strawberry.enum
class TaskStatus(Enum):
    OPEN = "open"
    IN_PROGRESS = "in_progress"
    DONE = "done"

@strawberry.type
class User:
    id: strawberry.ID
    name: str

@strawberry.type
class Task:
    id: strawberry.ID
    title: str
    status: TaskStatus
    assignee: User | None
    tags: list[str]

This single set of type-hinted classes is the schema's actual source of truth — Strawberry inspects it at import time and builds the corresponding GraphQL SDL automatically, so there's no separate .graphql schema file to keep in sync with your Python types by hand.

2. Query & Field Resolvers

A resolver is the Python function that fulfills one field or query from the schema. Query resolvers live on a root Query type; field resolvers can be attached to any type as a method, including for a field that isn't a direct column on the underlying model at all.

import strawberry
from strawberry.types import Info

@strawberry.type
class Query:

    @strawberry.field
    async def task(self, info: Info, id: strawberry.ID) -> Task | None:
        db_task = await task_repository.get(info.context["db"], int(id))
        return to_graphql_task(db_task) if db_task else None

    @strawberry.field
    async def tasks(self, info: Info, status: TaskStatus | None = None) -> list[Task]:
        db_tasks = await task_repository.search(info.context["db"], status=status)
        return [to_graphql_task(t) for t in db_tasks]

Info is Strawberry's per-request context object — this is where you thread through the database session and the authenticated user, the same dependency a FastAPI route would normally get via Depends(), since GraphQL resolvers don't participate in FastAPI's dependency injection system directly.

A field resolver fills in a value that isn't a plain column — assignee on Task, resolved from a separate User lookup rather than being a column on the task table itself:

@strawberry.type
class Task:
    id: strawberry.ID
    title: str
    status: TaskStatus
    _assignee_id: strawberry.Private[int | None]   # not exposed in the schema

    @strawberry.field
    async def assignee(self, info: Info) -> User | None:
        if self._assignee_id is None:
            return None
        db_user = await user_repository.get(info.context["db"], self._assignee_id)
        return User(id=db_user.id, name=db_user.name)

strawberry.Private marks _assignee_id as internal plumbing the resolver needs but that never appears in the public schema at all — a client asking only for { task(id: "1") { title } } never triggers the assignee resolver, since only the fields actually present in the query get resolved. A REST endpoint returning a full response DTO has no equivalent mechanism; it either always includes every field or needs a separate, purpose-built endpoint for every different shape a client might want.

3. Solving N+1 with DataLoader

That same per-field resolution is also exactly where GraphQL reintroduces the N+1 problem Week 15 solved for SQLAlchemy — in a new, GraphQL-specific shape. Querying a list of 20 tasks, each with its assignee field requested, calls the assignee resolver from Section 2 once per task — 20 separate database queries for 20 tasks' assignees, exactly the pattern selectinload fixed for a plain SQLAlchemy query, except there's no direct equivalent here, because GraphQL resolves each field independently by design.

query {
  tasks {
    title
    assignee { name }   # fires the "assignee" resolver once PER task in the list
  }
}
// 1 query for the task list, then N more queries -- one per task -- for assignees

DataLoader (via the strawberry.dataloader.DataLoader helper) solves this by batching: instead of each field resolver immediately querying the database, it registers the ID it needs with a shared DataLoader and awaits a not-yet-resolved future. DataLoader collects every ID requested during the current request "tick," then fires a single batched query for all of them at once.

from strawberry.dataloader import DataLoader

async def load_users(keys: list[int]) -> list[User | None]:
    async with SessionLocal() as db:
        result = await db.execute(select(UserModel).where(UserModel.id.in_(keys)))
        users_by_id = {u.id: u for u in result.scalars().all()}
        return [
            User(id=users_by_id[k].id, name=users_by_id[k].name) if k in users_by_id else None
            for k in keys
        ]

def get_context() -> dict:
    return {"user_loader": DataLoader(load_fn=load_users)}

@strawberry.field
async def assignee(self, info: Info) -> User | None:
    if self._assignee_id is None:
        return None
    return await info.context["user_loader"].load(self._assignee_id)

With the DataLoader in place, the same 20-task query with every assignee requested fires exactly 2 queries total — one for the tasks, one batched query fetching every distinct requested user ID at once — regardless of whether the list has 20 tasks or 2,000. Critically, the DataLoader instance has to be created fresh per request (as in get_context() above, called once per incoming request), not shared as a global — a DataLoader that persists batches and results across requests would leak one user's data into another's response.

4. Hands-on Exercise

5. Knowledge Check

Three quick questions. Expand each to check your answer.