Skip to content

Building Async GraphQL APIs with Strawberry

Strawberry builds a GraphQL schema from type-annotated Python classes and executes it on asyncio, which makes it a natural fit for services that already run on FastAPI or another ASGI framework. The executor runs async resolvers concurrently — and runs plain def resolvers directly on the event loop, which is where the main performance trap lies. Measured with Strawberry 0.329 on FastAPI and uvicorn, Python 3.14, 100 requests at 20 concurrent: a field whose sync resolver called time.sleep(0.05) — standing in for a synchronous client call — held the server to 19 requests per second with a p50 of 981 ms; the same wait as await asyncio.sleep(0.05) gave 277 requests per second and 57 ms; moving the blocking call to asyncio.to_thread gave 327 and 54 ms. Three aliased async fields in one query took 66 ms at p50, not 150: Strawberry resolved them concurrently. Three blocking ones took 3.0 s. A trivial { hello } query ran at about 1,040 requests per second. This guide builds a Strawberry API that keeps the loop free.

Prerequisites

1. Define the schema and mount it on FastAPI

Types are dataclass-like classes; fields with resolvers are methods. GraphQLRouter serves the schema over HTTP and WebSockets:

import strawberry
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter


@strawberry.type
class Product:
    id: int
    name: str
    price_cents: int


@strawberry.type
class Query:
    @strawberry.field
    async def product(self, info: strawberry.Info, id: int) -> Product | None:
        row = await info.context["db"].fetchrow(
            "SELECT id, name, price_cents FROM product WHERE id = $1", id)
        return Product(**row) if row else None


schema = strawberry.Schema(query=Query)

async def get_context(request: Request) -> dict:
    return {"db": request.app.state.pool, "request": request}

app = FastAPI(lifespan=lifespan)                      # lifespan opens app.state.pool
app.include_router(GraphQLRouter(schema, context_getter=get_context), prefix="/graphql")

The context getter runs per request; it is where the database pool, the authenticated user and per-request DataLoaders belong. Pools and clients are created once in the FastAPI lifespan, as in managing startup and shutdown with ASGI lifespan. Measured: a trivial query served about 1,040 requests per second on one uvicorn worker, so the GraphQL layer itself costs roughly a millisecond of CPU per small request — parse, validate, execute and serialize.

Verify: POST /graphql with { __typename } answers in a few milliseconds on an idle server.

100 requests, 20 concurrent, one field that waits 50 ms 3 horizontal bars comparing def resolver, time.sleep(0.05) with the others. 100 requests, 20 concurrent, one field that waits 50 ms def resolver, time.sleep(0.05) 19 req/s (p50 981 ms) async resolver, await asyncio.sleep 277 req/s (p50 57 ms) async resolver, await to_thread(blocking) 327 req/s (p50 54 ms) Strawberry 0.329 on FastAPI, one uvicorn worker, Python 3.14. A plain def resolver runs on the event loop; if it blocks, every request waits.

2. Keep blocking calls out of def resolvers

Strawberry calls a def resolver directly, on the event loop thread. That is the right choice for resolvers that compute from data already in memory — attribute access, formatting, arithmetic — and the wrong one for anything that waits:

@strawberry.type
class Query:
    @strawberry.field
    def blocking(self) -> str:
        time.sleep(0.05)               # a requests.get(), a sync DB driver, a file read...
        return "ok"

    @strawberry.field
    async def offloaded(self) -> str:
        return await asyncio.to_thread(legacy_client.get_status)   # blocking call in a thread

Measured: the blocking resolver turned 20 concurrent requests into a queue, serving 19 requests per second — each request waited for every request ahead of it. The to_thread version served 327. If a synchronous SDK has no async alternative, to_thread is the escape hatch, with the thread-pool limits described in running blocking SDK calls with asyncio.to_thread; a native async client is better when one exists.

Verify: grep resolver modules for requests., time.sleep, synchronous database drivers and open( inside def resolvers.

3. Let sibling fields run concurrently

GraphQL resolves sibling fields independently, and Strawberry awaits sibling async resolvers concurrently. A query that asks for several slow fields therefore costs the slowest, not the sum:

{
  a: inventory(sku: "A1") { available }
  b: inventory(sku: "B2") { available }
  c: inventory(sku: "C3") { available }
}

Measured with three aliases of a 50 ms async field: p50 66 ms per request. The same query against the blocking def field took 3.0 s at p50 — each 50 ms block serialized, multiplied by the 20 concurrent requests competing for the one loop. Concurrency within a query is a feature clients can exploit deliberately — fetching a dashboard's panels in one round trip — and also a load multiplier: one request can start many upstream calls at once, which is a reason to bound total upstream concurrency with a semaphore, as in limiting concurrent requests with asyncio.Semaphore, and to cap aliases as in limiting GraphQL query depth and aliases.

Verify: a query with N independent slow fields takes about one field's latency, not N times it.

Three aliased fields that each wait 50 ms A grid of 2 rows by 4 columns. Three aliased fields that each wait 50 ms resolver type p50 p99 why async def, awaits 66 ms 84 ms three awaits overlap def, time.sleep 3,045 ms 3,051 ms each block holds the loop 20 concurrent requests; one uvicorn worker.

4. Put per-request state in the context

Resolvers get request-scoped state through info.context: the database pool, the current user, DataLoaders. Build it in the context getter so every request starts clean:

from strawberry.dataloader import DataLoader


async def get_context(request: Request) -> dict:
    user = await authenticate(request.headers.get("authorization"))
    pool = request.app.state.pool
    return {
        "request": request,
        "user": user,
        "db": pool,
        "loaders": {
            "posts_by_author": DataLoader(load_fn=partial(load_posts, pool)),
            "users": DataLoader(load_fn=partial(load_users, pool)),
        },
    }

Authentication in the context getter runs once per request rather than once per field, and gives every resolver the same view of who is asking. DataLoaders created here batch within the request and are discarded with it — the per-request scoping that scoping DataLoaders per request shows to matter for correctness, not only memory. For values that must also be visible to code far from the resolver (logging, tracing), a ContextVar set in the context getter complements info.context, as in propagating request IDs with contextvars.

Verify: no resolver reads module-level mutable state that differs per user.

5. Handle errors per field, not per request

A GraphQL response can carry data and errors together: a failing field becomes null with an entry in errors, and its siblings still resolve. Use that deliberately:

@strawberry.type
class Dashboard:
    @strawberry.field
    async def recommendations(self, info: strawberry.Info) -> list[Product] | None:
        try:
            async with asyncio.timeout(0.3):
                return await info.context["recs"].for_user(info.context["user"].id)
        except (TimeoutError, RecsUnavailable):
            log.warning("recommendations degraded")
            return None                                  # the rest of the dashboard still renders

Making optional, non-critical fields nullable and bounding them with a timeout turns a slow dependency into a degraded field instead of a slow or failed page. Fields that must not silently disappear — prices, balances — should raise, so the client sees an error rather than a misleading null. The degradation pattern itself is covered in adding timeouts and fallbacks to degraded dependencies.

Verify: with one dependency down, the affected field is null with an error entry and every other field resolves.

How should this resolver be written? A decision on What does the field do with 4 outcomes. How should this resolver be written? What does the field do? computes from loaded data def resolver runs inline, cheap calls an async client async def siblings overlap must call a blocking library async def + to_thread 327 vs 19 req/s loads by parent ID per-request DataLoader one query per level A def resolver that waits blocks every request on the worker.

Verification

A Strawberry API is async-safe when:

  • Every resolver that waits is async def, and blocking libraries run through to_thread.
  • Sibling slow fields overlap, so N fields cost about one field's latency.
  • Per-request state lives in the context, built by the context getter.
  • Optional fields degrade to null within a timeout, and critical fields raise.

Diagnostic Hook: measure event-loop lag on each GraphQL worker alongside request latency. Lag that rises with traffic means a def resolver is blocking; latency that rises with flat lag means a dependency is slow and the field-level timeouts are where to look.

Pitfalls & edge cases

  • Blocking def resolvers. Measured: 19 req/s instead of 277–327.
  • Unbounded fan-out within a query. Aliases multiply upstream calls; cap them.
  • Authentication per field. Do it once in the context getter.
  • Nullable everything. Critical fields that silently become null hide real failures.

Frequently Asked Questions

Are Strawberry resolvers async?

Async def resolvers are awaited, and sibling async fields run concurrently; plain def resolvers run directly on the event loop thread. A def resolver that blocked for 50 ms held a FastAPI server to 19 req/s in testing.

How do I call a blocking library from a Strawberry resolver?

Make the resolver async def and call await asyncio.to_thread(blocking_fn, ...). In testing that raised throughput from 19 to 327 req/s for a 50 ms blocking call.

How do I add Strawberry GraphQL to FastAPI?

Create strawberry.Schema(query=Query) and include GraphQLRouter(schema, context_getter=...) with a prefix such as /graphql; open pools in the FastAPI lifespan and pass them through the context.

Do GraphQL fields resolve in parallel?

Sibling async fields do: three aliased 50 ms fields took 66 ms at p50 rather than 150 ms. Nested fields wait for their parent.