Python GraphQL with Strawberry: Schema-First API Design

Strawberry is a modern Python GraphQL library that uses type annotations to define your schema — no separate SDL files, no code generation. Your Python dataclasses become GraphQL types automatically. This guide covers schema definition, queries, mutations, subscriptions, N+1 problem prevention with DataLoader, and integrating Strawberry with FastAPI for a production-ready GraphQL API.

Setup and First Schema

pip install strawberry-graphql[fastapi] uvicorn sqlalchemy aiosqlite

import strawberry
from typing import Optional
from datetime import datetime

# Strawberry types are defined as Python dataclasses with @strawberry.type
@strawberry.type
class Author:
    id: strawberry.ID
    name: str
    email: str

@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    content: str
    published: bool
    created_at: datetime
    author: Author

# Input types for mutations
@strawberry.input
class CreatePostInput:
    title: str
    content: str
    author_id: strawberry.ID
    published: bool = False

# Define the root Query type
@strawberry.type
class Query:
    @strawberry.field
    async def post(self, id: strawberry.ID) -> Optional[Post]:
        return await get_post_by_id(id)

    @strawberry.field
    async def posts(self, published_only: bool = False) -> list[Post]:
        return await get_posts(published_only)

schema = strawberry.Schema(query=Query)

Queries and Resolvers

Resolvers are async methods on your Query type. Strawberry passes the info object (containing context, request, etc.) as the second argument when you need it. Return types can be Python objects, dataclasses, or Strawberry types — the library handles serialization automatically.

from strawberry.types import Info
from typing import Annotated

@strawberry.type
class Query:
    @strawberry.field
    async def post(self, id: strawberry.ID, info: Info) -> Optional[Post]:
        db = info.context["db"]
        row = await db.execute("SELECT * FROM posts WHERE id = ?", [id])
        data = await row.fetchone()
        if not data:
            return None
        return Post(
            id=str(data["id"]),
            title=data["title"],
            content=data["content"],
            published=bool(data["published"]),
            created_at=data["created_at"],
            author=await load_author(data["author_id"], info),
        )

    @strawberry.field
    async def posts(
        self,
        info: Info,
        published_only: bool = False,
        limit: int = 20,
        offset: int = 0,
    ) -> list[Post]:
        db = info.context["db"]
        query = "SELECT * FROM posts"
        params = []
        if published_only:
            query += " WHERE published = 1"
        query += f" LIMIT ? OFFSET ?"
        params.extend([limit, offset])
        rows = await (await db.execute(query, params)).fetchall()
        return [row_to_post(r) for r in rows]

    @strawberry.field
    async def search_posts(self, info: Info, query: str) -> list[Post]:
        db = info.context["db"]
        rows = await (await db.execute(
            "SELECT * FROM posts WHERE title LIKE ? OR content LIKE ?",
            [f"%{query}%", f"%{query}%"]
        )).fetchall()
        return [row_to_post(r) for r in rows]

Mutations

Mutations are defined on a Mutation type and passed to the schema. Use @strawberry.input classes for mutation arguments to keep the schema clean.

from strawberry.scalars import JSON

@strawberry.type
class MutationResult:
    success: bool
    message: str

@strawberry.type
class Mutation:
    @strawberry.mutation
    async def create_post(self, info: Info, input: CreatePostInput) -> Post:
        db = info.context["db"]
        user = info.context.get("user")
        if not user:
            raise strawberry.exceptions.GraphQLError("Unauthorized")

        result = await db.execute(
            "INSERT INTO posts (title, content, author_id, published) VALUES (?, ?, ?, ?)",
            [input.title, input.content, input.author_id, input.published],
        )
        await db.commit()
        return await get_post_by_id(str(result.lastrowid), db)

    @strawberry.mutation
    async def publish_post(self, info: Info, id: strawberry.ID) -> Post:
        db = info.context["db"]
        await db.execute("UPDATE posts SET published = 1 WHERE id = ?", [id])
        await db.commit()
        return await get_post_by_id(id, db)

    @strawberry.mutation
    async def delete_post(self, info: Info, id: strawberry.ID) -> MutationResult:
        db = info.context["db"]
        await db.execute("DELETE FROM posts WHERE id = ?", [id])
        await db.commit()
        return MutationResult(success=True, message=f"Post {id} deleted")

schema = strawberry.Schema(query=Query, mutation=Mutation)

Subscriptions

Strawberry supports real-time subscriptions via WebSockets. Each subscription is an async generator that yields values over time.

import asyncio
from typing import AsyncGenerator

# In-memory pub/sub for demo — use Redis in production
_subscribers: list[asyncio.Queue] = []

async def publish_post_event(event: dict):
    for queue in _subscribers:
        await queue.put(event)

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def post_published(self) -> AsyncGenerator[Post, None]:
        queue: asyncio.Queue = asyncio.Queue()
        _subscribers.append(queue)
        try:
            while True:
                event = await queue.get()
                if event["type"] == "post_published":
                    yield event["post"]
        finally:
            _subscribers.remove(queue)

schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    subscription=Subscription,
)

DataLoader: Solving N+1

The N+1 problem occurs when fetching a list of N posts each triggers a separate query to load the author. DataLoader batches all author ID lookups from a single request into one query, then distributes the results.

from strawberry.dataloader import DataLoader
from typing import Sequence

async def load_authors_batch(author_ids: Sequence[strawberry.ID]) -> list[Optional[Author]]:
    """Batch-load authors by IDs — called once per request with all IDs."""
    db = get_db()  # get current DB connection
    placeholders = ",".join("?" * len(author_ids))
    rows = await (await db.execute(
        f"SELECT * FROM authors WHERE id IN ({placeholders})",
        list(author_ids),
    )).fetchall()

    id_to_author = {str(r["id"]): row_to_author(r) for r in rows}
    # Must return results in the same order as the input IDs
    return [id_to_author.get(str(aid)) for aid in author_ids]

# Create DataLoader per request (not global — each request gets fresh state)
def get_context(request):
    return {
        "request": request,
        "db": get_db_session(),
        "author_loader": DataLoader(load_fn=load_authors_batch),
        "user": get_current_user(request),
    }

# Use in Post type resolver
@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    author_id: strawberry.Private[strawberry.ID]  # private — not in schema

    @strawberry.field
    async def author(self, info: Info) -> Author:
        return await info.context["author_loader"].load(self.author_id)

FastAPI Integration

from fastapi import FastAPI, Request
from strawberry.fastapi import GraphQLRouter

app = FastAPI()

async def get_context(request: Request):
    return {
        "request": request,
        "db": await get_db_session(),
        "user": await get_current_user_optional(request),
        "author_loader": DataLoader(load_fn=load_authors_batch),
    }

graphql_app = GraphQLRouter(schema, context_getter=get_context)
app.include_router(graphql_app, prefix="/graphql")

# GraphQL Playground available at http://localhost:8000/graphql

Authentication and Permissions

Strawberry has a built-in @strawberry.permission_classes decorator for field-level authorization.

from strawberry.permission import BasePermission
from strawberry.types import Info

class IsAuthenticated(BasePermission):
    message = "Authentication required"

    def has_permission(self, source, info: Info, **kwargs) -> bool:
        return info.context.get("user") is not None

class IsAdmin(BasePermission):
    message = "Admin access required"

    def has_permission(self, source, info: Info, **kwargs) -> bool:
        user = info.context.get("user")
        return user is not None and "admin" in user.roles

@strawberry.type
class Query:
    @strawberry.field(permission_classes=[IsAuthenticated])
    async def my_posts(self, info: Info) -> list[Post]:
        user = info.context["user"]
        return await get_posts_by_author(user.id, info.context["db"])

    @strawberry.field(permission_classes=[IsAdmin])
    async def all_users(self, info: Info) -> list[Author]:
        return await get_all_users(info.context["db"])

Frequently Asked Questions

GraphQL vs REST: when to choose GraphQL?

Choose GraphQL when clients need flexible queries (mobile apps with different data needs than web), when you want to avoid over-fetching/under-fetching, or when aggregating data from multiple backends. Choose REST for simple CRUD, caching requirements (GraphQL POST requests bypass HTTP caches), or when clients are simple and always need the same shape of data.

How do I handle file uploads with Strawberry?

Enable multipart uploads: strawberry.Schema(..., extensions=[UploadExtension]) and use the Upload scalar type in your mutation input. The file is available as input.file as an UploadFile object compatible with Starlette.

Can I use Strawberry with Django?

Yes. Strawberry ships a Django view: strawberry.django.views.GraphQLView. There is also a separate strawberry-django package that generates types directly from Django models.