FastAPI Backend Development Bootcamp · Lesson

Real-Time GraphQL Subscriptions

Stream live updates to clients over WebSocket-backed GraphQL subscriptions with backpressure awareness.

Real-Time GraphQL Subscriptions is a free FastAPI Backend Development Bootcamp lesson on CoddyKit — lesson 3 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the FastAPI Backend Development Bootcamp learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Subscriptions Exist

GraphQL gives you three root operation types: query (read), mutation (write), and subscription (live stream). A subscription is the only one that keeps a long-lived connection open and pushes multiple results to the client over time.

Query/Mutation: one request, one response, connection closes.
Subscription: one request, a stream of responses, until either side closes it.

In Strawberry, a subscription resolver is an async generator: every yield becomes one payload delivered to the client. Under the hood, FastAPI carries this over a WebSocket, because plain HTTP cannot push server-initiated frames cleanly.

An Async Generator Is the Core Idea

Before wiring GraphQL, understand the engine: Python async generators. Each yield hands one value to the consumer and then suspends until the consumer asks for the next one. This natural pull-based flow is what gives subscriptions backpressure for free: the producer only advances when the consumer is ready.

This snippet runs anywhere — it is the exact shape a Strawberry subscription resolver takes, minus the decorator.

import asyncio

async def counter(limit: int):
    for i in range(limit):
        await asyncio.sleep(0.1)  # simulate work / waiting for an event
        yield i

async def main():
    async for value in counter(5):
        print(f"received: {value}")

asyncio.run(main())

Your First Strawberry Subscription

A Strawberry subscription lives on a type decorated with @strawberry.type and uses @strawberry.subscription on an async def that yields. The return annotation uses AsyncGenerator[T, None] so the schema knows the payload type.

Here a client subscribing to count receives one integer every 500ms until the target is reached.

import asyncio
from typing import AsyncGenerator
import strawberry


@strawberry.type
class Query:
    @strawberry.field
    def ping(self) -> str:
        return "pong"


@strawberry.type
class Subscription:
    @strawberry.subscription
    async def count(self, target: int = 5) -> AsyncGenerator[int, None]:
        for i in range(target):
            yield i
            await asyncio.sleep(0.5)


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

Mounting the WebSocket Route in FastAPI

Subscriptions need a WebSocket transport. Strawberry ships GraphQLRouter, which exposes both the HTTP endpoint (for queries/mutations) and the WebSocket endpoint (for subscriptions) at the same path.

You must declare the supported WebSocket subprotocols. The modern one is graphql-transport-ws (the graphql-ws library); the legacy one is graphql-ws (subscriptions-transport-ws). Offer both for client compatibility.

from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
from strawberry.subscriptions import (
    GRAPHQL_TRANSPORT_WS_PROTOCOL,
    GRAPHQL_WS_PROTOCOL,
)

from schema import schema  # the schema from the previous scene

graphql_app = GraphQLRouter(
    schema,
    subscription_protocols=[
        GRAPHQL_TRANSPORT_WS_PROTOCOL,
        GRAPHQL_WS_PROTOCOL,
    ],
)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

What the Client Sends

A subscription is a GraphQL document like any other — the difference is the operation keyword. The client opens a WebSocket to /graphql, negotiates the subprotocol, then sends a subscribe message carrying this document:

Server replies with a next message per yield.
Server sends complete when the generator finishes.
Either side can send complete to stop early.

The shape of the selection set must match the payload type your resolver yields.

subscription OnCount {
  count(target: 5)
}

Streaming Domain Events, Not Just Counters

Real subscriptions push domain objects. Define a Strawberry type for the payload and yield instances of it. Below, an order-status feed yields a structured object each time something changes.

The async for here would, in production, be reading from a real event source (a queue, Redis Pub/Sub, a Postgres LISTEN/NOTIFY channel). The resolver's only job is to translate those events into GraphQL payloads.

from typing import AsyncGenerator
import strawberry


@strawberry.type
class OrderUpdate:
    order_id: strawberry.ID
    status: str
    updated_at: str


@strawberry.type
class Subscription:
    @strawberry.subscription
    async def order_status(
        self, order_id: strawberry.ID
    ) -> AsyncGenerator[OrderUpdate, None]:
        async for event in event_source(order_id):  # external event stream
            yield OrderUpdate(
                order_id=order_id,
                status=event["status"],
                updated_at=event["ts"],
            )

Fan-Out with an asyncio Broadcast Queue

Many clients usually subscribe to the same event. A single publisher must fan out to N subscribers without one slow subscriber stalling the others. The classic in-process pattern: each subscriber gets its own asyncio.Queue, and the publisher puts the event on every queue.

This standalone example models that fan-out: two subscribers each drain their own bounded queue independently.

import asyncio


class Broadcaster:
    def __init__(self):
        self._subscribers: list[asyncio.Queue] = []

    def subscribe(self) -> asyncio.Queue:
        q: asyncio.Queue = asyncio.Queue(maxsize=10)
        self._subscribers.append(q)
        return q

    async def publish(self, item):
        for q in self._subscribers:
            await q.put(item)


async def subscriber(name, q, n):
    for _ in range(n):
        item = await q.get()
        print(f"{name} got {item}")


async def main():
    b = Broadcaster()
    a, c = b.subscribe(), b.subscribe()
    consumers = asyncio.gather(subscriber("A", a, 3), subscriber("B", c, 3))
    for i in range(3):
        await b.publish(i)
    await consumers


asyncio.run(main())

Backpressure: Bounded Queues and Drop Policy

Backpressure is what happens when a producer outpaces a consumer. With an unbounded queue, a slow client makes memory grow without limit until the server falls over. You have three honest strategies:

Block: use a bounded Queue and await q.put() — the publisher slows to the slowest consumer (safe, but couples clients).
Drop: on QueueFull, discard the oldest or newest item — bounded memory, lossy (good for telemetry/tickers).
Disconnect: if a client lags past a threshold, close its subscription.

The snippet below demonstrates a non-blocking drop-oldest policy on a bounded queue.

import asyncio


def offer(q: asyncio.Queue, item) -> bool:
    """Try to enqueue; on overflow drop the oldest. Never blocks."""
    try:
        q.put_nowait(item)
        return True
    except asyncio.QueueFull:
        _ = q.get_nowait()      # drop oldest
        q.put_nowait(item)      # make room for newest
        return False


async def main():
    q: asyncio.Queue = asyncio.Queue(maxsize=2)
    results = [offer(q, i) for i in range(5)]
    print("accepted-without-drop:", results)
    drained = [q.get_nowait() for _ in range(q.qsize())]
    print("survivors:", drained)


asyncio.run(main())

Cleanup on Disconnect with try/finally

When a client closes the WebSocket (or the generator is cancelled), Strawberry throws asyncio.CancelledError into your resolver at the suspended yield. If you allocated a queue, a Redis subscription, or a DB LISTEN, you must release it — otherwise you leak resources and the broadcaster keeps pushing into a dead queue.

Wrap the loop in try/finally. The finally runs on normal completion, on error, and on cancellation.

from typing import AsyncGenerator
import strawberry


@strawberry.type
class Subscription:
    @strawberry.subscription
    async def order_status(
        self, order_id: strawberry.ID
    ) -> AsyncGenerator[str, None]:
        queue = broadcaster.subscribe(order_id)
        try:
            while True:
                status = await queue.get()
                yield status
        finally:
            broadcaster.unsubscribe(order_id, queue)  # always cleans up

Authenticating a Subscription

WebSocket connections do not carry per-request headers the way HTTP does, so you authenticate during the connection init handshake. With graphql-transport-ws, the client sends a connection_init message with a payload (e.g. a token). Strawberry exposes this via a custom on_ws_connect hook or through the connection params in context.

Reject early: raise inside on_ws_connect to refuse the socket before any subscription starts. Authorize per-subscription inside the resolver using the validated identity from context.

from strawberry.fastapi import GraphQLRouter
from strawberry.subscriptions.protocols.graphql_transport_ws.types import (
    ConnectionInitMessage,
)


class AuthGraphQLRouter(GraphQLRouter):
    async def on_ws_connect(self, context):
        params = context["connection_params"] or {}
        token = params.get("authToken")
        user = verify_token(token)        # raises if invalid
        if user is None:
            raise ConnectionRejectionError()  # closes the socket
        context["user"] = user
        return {"ack": True}

Filtering and Heartbeats

Two production touches keep subscriptions healthy:

Server-side filtering: do not stream events a client should not see. Filter inside the resolver before yield, using the authenticated user and the subscription arguments. Never rely on the client to discard.
Heartbeats / keepalive: idle WebSockets get killed by proxies and load balancers. The graphql-transport-ws protocol has built-in ping/pong; you can also yield periodic keepalive payloads. Configure your reverse proxy (nginx) idle timeout above your heartbeat interval.

This resolver filters by tenant and only forwards relevant events.

from typing import AsyncGenerator
import strawberry


@strawberry.type
class Subscription:
    @strawberry.subscription
    async def notifications(
        self, info: strawberry.Info
    ) -> AsyncGenerator[str, None]:
        user = info.context["user"]
        async for event in broadcaster.stream():
            if event["tenant_id"] != user.tenant_id:
                continue  # server-side filter; never trust the client
            yield event["message"]

Quick Check: Handling a Slow Consumer

You run a market-data subscription. One client's network is slow and cannot keep up with the event rate. Other clients on the same publisher are fast. What is the safest default design to protect the server while keeping fast clients real-time?

Recap

You built real-time GraphQL over FastAPI with Strawberry:

Subscriptions are async generators: each yield is one pushed payload; the pull-based protocol gives natural backpressure.
Transport: GraphQLRouter serves queries/mutations over HTTP and subscriptions over WebSocket, advertising graphql-transport-ws (modern) and graphql-ws (legacy).
Fan-out: give each subscriber its own queue so one slow client never stalls the others.
Backpressure: bound your queues and pick a policy — block, drop, or disconnect. Never use unbounded buffers.
Lifecycle: use try/finally to release queues and external subscriptions on cancellation.
Security and health: authenticate during connection_init, filter events server-side, and run heartbeats so proxies don't drop idle sockets.

Frequently asked questions

Is the “Real-Time GraphQL Subscriptions” lesson free?

Yes — the full text of “Real-Time GraphQL Subscriptions” is free to read here on the web, and the FastAPI Backend Development Bootcamp course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the FastAPI Backend Development Bootcamp course, upgrade to CoddyKit PRO.

What will I learn in “Real-Time GraphQL Subscriptions”?

Stream live updates to clients over WebSocket-backed GraphQL subscriptions with backpressure awareness. You practise FastAPI Backend Development Bootcamp with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start FastAPI Backend Development Bootcamp?

No prior experience is required. FastAPI Backend Development Bootcamp on CoddyKit is structured for beginners through advanced learners; this is — lesson 3 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Real-Time GraphQL Subscriptions” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this FastAPI Backend Development Bootcamp lesson?

Yes. Every FastAPI Backend Development Bootcamp lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

← Back to FastAPI Backend Development Bootcamp