GraphQL Server with Strawberry Python

GraphQL Server with Strawberry Python

Problem Statement

Modern web applications often need flexible, type-safe APIs where clients can request exactly the data they need — no more, no less. REST endpoints tend to over-fetch or under-fetch data, leading to chatty clients and bloated payloads. GraphQL solves this by exposing a single endpoint with a strongly-typed schema that clients query declaratively.

This tutorial builds a production-ready GraphQL server using Strawberry (a modern, dataclass-based GraphQL library for Python) mounted on FastAPI and served by Uvicorn. It demonstrates a realistic in-memory book catalog with queries, mutations, pagination, input validation, and structured error handling — the patterns you'll reuse for any domain.

How It Works

The architecture is a three-layer stack:

1. Uvicorn — ASGI server that handles HTTP transport and process management.
2. FastAPI — the web framework that mounts Strawberry's ASGI app at /graphql, also exposing a /health endpoint and serving the GraphiQL IDE for interactive exploration.
3. Strawberry — defines the GraphQL schema from Python dataclasses (@strawberry.type), resolves fields via functions, validates inputs (@strawberry.input), and executes queries against a thread-safe in-memory store.

Data flows: client sends a POST to /graphql → FastAPI routes to Strawberry → Strawberry parses, validates against the schema, and executes resolvers → resolvers read/write the BookStore singleton → response is serialized back as JSON. We use a repository pattern (BookStore) to isolate persistence so swapping in a real database (SQLAlchemy, Postgres) later requires changing only that class. Errors are raised as GraphQLError with extensions so clients get structured, machine-readable failures instead of raw 500s.

Full Script

#!/usr/bin/env python3
"""
GraphQL Server with Strawberry Python
-------------------------------------
A production-ready GraphQL API for a book catalog, built with Strawberry,
mounted on FastAPI, and served by Uvicorn.

Run with:
    python3 -m uvicorn graphql_app:app --host 0.0.0.0 --port 8000

Environment variables:
    GRAPHQL_PATH          Path the GraphQL endpoint is mounted at (default: /graphql)
    HOST                  Bind host (default: 0.0.0.0)
    PORT                  Bind port (default: 8000)
    LOG_LEVEL             Uvicorn log level (default: info)
    ENABLE_GRAPHIQL       Enable the GraphiQL IDE (default: true)
    MAX_PAGE_SIZE         Hard cap on pagination page size (default: 100)
"""

from __future__ import annotations

import logging
import os
import threading
import uuid
from dataclasses import dataclass, field
from typing import Optional

import strawberry
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from graphql import GraphQLError
from strawberry.fastapi import GraphQLRouter

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
GRAPHQL_PATH: str = os.environ.get("GRAPHQL_PATH", "/graphql")
HOST: str = os.environ.get("HOST", "0.0.0.0")
PORT: int = int(os.environ.get("PORT", "8000"))
LOG_LEVEL: str = os.environ.get("LOG_LEVEL", "info").lower()
ENABLE_GRAPHIQL: bool = os.environ.get("ENABLE_GRAPHIQL", "true").lower() == "true"
MAX_PAGE_SIZE: int = int(os.environ.get("MAX_PAGE_SIZE", "100"))

logging.basicConfig(
    level=getattr(logging, LOG_LEVEL.upper(), logging.INFO),
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("graphql_app")


# ---------------------------------------------------------------------------
# Domain model + repository (in-memory, thread-safe)
# ---------------------------------------------------------------------------
@dataclass
class BookRecord:
    """Internal persistence record. Kept separate from the GraphQL type."""
    id: str
    title: str
    author: str
    year: int
    genres: list[str] = field(default_factory=list)


class BookStore:
    """
    Thread-safe in-memory store. Swap this for a DB-backed repository
    without touching the GraphQL layer.
    """

    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._books: dict[str, BookRecord] = {}
        self._seed()

    def _seed(self) -> None:
        seed = [
            ("Clean Code", "Robert C. Martin", 2008, ["software", "craft"]),
            ("The Pragmatic Programmer", "Hunt & Thomas", 1999, ["software"]),
            ("Designing Data-Intensive Applications", "Martin Kleppmann", 2017, ["distributed", "data"]),
            ("Fluent Python", "Luciano Ramalho", 2022, ["python", "software"]),
        ]
        for title, author, year, genres in seed:
            self.create(title=title, author=author, year=year, genres=genres)

    def create(self, title: str, author: str, year: int, genres: list[str]) -> BookRecord:
        with self._lock:
            book_id = str(uuid.uuid4())
            record = BookRecord(id=book_id, title=title, author=author, year=year, genres=genres)
            self._books[book_id] = record
            logger.info("Created book %s (%s)", book_id, title)
            return record

    def get(self, book_id: str) -> Optional[BookRecord]:
        with self._lock:
            return self._books.get(book_id)

    def list(self, offset: int, limit: int, author_contains: Optional[str] = None) -> list[BookRecord]:
        with self._lock:
            items = list(self._books.values())
        if author_contains:
            needle = author_contains.lower()
            items = [b for b in items if needle in b.author.lower()]
        items.sort(key=lambda b: b.title.lower())
        return items[offset : offset + limit]

    def count(self, author_contains: Optional[str] = None) -> int:
        with self._lock:
            items = list(self._books.values())
        if author_contains:
            needle = author_contains.lower()
            items = [b for b in items if needle in b.author.lower()]
        return len(items)

    def delete(self, book_id: str) -> bool:
        with self._lock:
            return self._books.pop(book_id, None) is not None


STORE = BookStore()


# ---------------------------------------------------------------------------
# GraphQL schema
# ---------------------------------------------------------------------------
@strawberry.type
class Book:
    id: strawberry.ID
    title: str
    author: str
    year: int
    genres: list[str]

    @classmethod
    def from_record(cls, record: BookRecord) -> "Book":
        return cls(
            id=strawberry.ID(record.id),
            title=record.title,
            author=record.author,
            year=record.year,
            genres=record.genres,
        )


@strawberry.type
class BookPage:
    """Cursor-free offset pagination result."""
    items: list[Book]
    total: int
    offset: int
    limit: int
    has_next: bool


@strawberry.input
class CreateBookInput:
    title: str
    author: str
    year: int
    genres: list[str] = strawberry.field(default_factory=list)


@strawberry.type
class Query:
    @strawberry.field
    def book(self, id: strawberry.ID) -> Optional[Book]:
        """Fetch a single book by ID. Returns null if not found."""
        record = STORE.get(str(id))
        return Book.from_record(record) if record else None

    @strawberry.field
    def books(
        self,
        offset: int = 0,
        limit: int = 20,
        author_contains: Optional[str] = None,
    ) -> BookPage:
        """Paginated book listing with optional author substring filter."""
        if offset < 0:
            raise GraphQLError("offset must be >= 0", extensions={"code": "BAD_INPUT"})
        if limit <= 0:
            raise GraphQLError("limit must be > 0", extensions={"code": "BAD_INPUT"})
        if limit > MAX_PAGE_SIZE:
            raise GraphQLError(
                f"limit exceeds MAX_PAGE_SIZE ({MAX_PAGE_SIZE})",
                extensions={"code": "LIMIT_EXCEEDED", "max": MAX_PAGE_SIZE},
            )
        records = STORE.list(offset=offset, limit=limit, author_contains=author_contains)
        total = STORE.count(author_contains=author_contains)
        return BookPage(
            items=[Book.from_record(r) for r in records],
            total=total,
            offset=offset,
            limit=limit,
            has_next=(offset + len(records)) < total,
        )


@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_book(self, input: CreateBookInput) -> Book:
        """Create a new book. Validates required fields."""
        title = input.title.strip()
        author = input.author.strip()
        if not title or not author:
            raise GraphQLError(
                "title and author must be non-empty",
                extensions={"code": "VALIDATION_ERROR"},
            )
        if input.year < 1000 or input.year > 2100:
            raise GraphQLError(
                "year must be between 1000 and 2100",
                extensions={"code": "VALIDATION_ERROR"},
            )
        record = STORE.create(
            title=title, author=author, year=input.year, genres=list(input.genres)
        )
        return Book.from_record(record)

    @strawberry.mutation
    def delete_book(self, id: strawberry.ID) -> bool:
        """Delete a book by ID. Returns true if it existed."""
        deleted = STORE.delete(str(id))
        if not deleted:
            raise GraphQLError(
                f"book {id} not found",
                extensions={"code": "NOT_FOUND"},
            )
        return True


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


# ---------------------------------------------------------------------------
# FastAPI wiring
# ---------------------------------------------------------------------------
graphql_router = GraphQLRouter(
    schema,
    graphiql=ENABLE_GRAPHIQL,
    path=GRAPHQL_PATH,
)

app = FastAPI(title="Strawberry Book Catalog", version="1.0.0")
app.include_router(graphql_router, prefix=GRAPHQL_PATH)


@app.get("/health", tags=["ops"])
async def health() -> dict[str, str | int]:
    """Liveness probe for orchestrators (Kubernetes, ECS, etc.)."""
    return {"status": "ok", "books": STORE.count()}


@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """Catch-all so unhandled errors return JSON, not an HTML traceback."""
    logger.exception("Unhandled error on %s %s", request.method, request.url.path)
    return JSONResponse(
        status_code=500,
        content={"error": "internal_server_error", "detail": str(exc)},
    )


# ---------------------------------------------------------------------------
# Entry point (dev convenience; production uses uvicorn CLI)
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    import uvicorn

    logger.info(
        "Starting Strawberry GraphQL server on http://%s:%s%s (GraphiQL=%s)",
        HOST, PORT, GRAPHQL_PATH, ENABLE_GRAPHIQL,
    )
    uvicorn.run(
        "graphql_app:app",
        host=HOST,
        port=PORT,
        log_level=LOG_LEVEL,
        reload=False,
    )

Configuration

Usage Examples

1. Start the server:

$ python3 -m uvicorn graphql_app:app --host 0.0.0.0 --port 8000
INFO:     Started server process [12345]
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     Application startup complete.

2. Query books with curl:

$ curl -s http://localhost:8000/graphql \
    -H 'Content-Type: application/json' \
    -d '{"query":"{ books(limit: 2) { total hasNext items { id title author year } } }"}'

{"data":{"books":{"total":4,"hasNext":true,"items":[
  {"id":"a1b2...","title":"Clean Code","author":"Robert C. Martin","year":2008},
  {"id":"c3d4...","title":"Designing Data-Intensive Applications","author":"Martin Kleppmann","year":2017}
]}}}

3. Create a book via mutation:

$ curl -s http://localhost:8000/graphql \
    -H 'Content-Type: application/json' \
    -d '{"query":"mutation($i: CreateBookInput!){ createBook(input:$i){ id title year } }",
         "variables":{"i":{"title":"Refactoring","author":"Martin Fowler","year":2018,"genres":["software"]}}}'

{"data":{"createBook":{"id":"e5f6...","title":"Refactoring","year":2018}}}

4. Health check:

$ curl -s http://localhost:8000/health
{"status":"ok","books":5}

Error Handling

• Validation errors (empty title, out-of-range year, negative offset) raise GraphQLError with an extensions.code so clients can branch on VALIDATION_ERROR vs BAD_INPUT without parsing strings.
• Not found — book(id:) returns null (GraphQL-native "optional"), while deleteBook raises NOT_FOUND since deletion is an explicit intent.
• Pagination abuse — limit > MAX_PAGE_SIZE is rejected with LIMIT_EXCEEDED, protecting the backend from expensive queries.
• Malformed GraphQL — Strawberry returns a standard errors array with locations and path, never a raw 500.
• Unhandled exceptions — the FastAPI catch-all logs the full traceback server-side and returns a JSON 500 body, keeping internal details out of the client response.
• Concurrency — BookStore uses an RLock around every read/write, so parallel requests from Uvicorn workers won't corrupt the in-memory dict.

Extending Ideas

• Swap the in-memory store for a database. Replace BookStore with a SQLAlchemy or async SQLModel repository and add a DataLoader (strawberry.dataloader.DataLoader) to batch nested lookups and avoid the N+1 problem.
• Add authentication and per-field authorization. Use a FastAPI dependency to extract a JWT from the Authorization header, inject it into the GraphQL context via GraphQLRouter(..., context_getter=...), and check permissions inside resolvers.
• Introduce subscriptions. Strawberry supports WebSocket subscriptions with @strawberry.subscription. Wire it up with strawberry.fastapi and an async pub/sub (Redis, NATS) to push live book updates to clients.
• Add observability. Integrate strawberry.extensions.apollo_tracing or OpenTelemetry to emit per-field traces and metrics, then expose /metrics for Prometheus scraping.