0Pricing
FastAPI Backend Development Bootcamp · 课时

实时 GraphQL 订阅

通过基于 WebSocket 的 GraphQL 订阅向客户端推送实时更新，并关注背压问题。

实时 GraphQL 订阅 是 CoddyKit 上的免费 FastAPI Backend Development Bootcamp 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 FastAPI Backend Development Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

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.

常见问题解答

「实时 GraphQL 订阅」课时是免费的吗？

是的 — 「实时 GraphQL 订阅」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 FastAPI Backend Development Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

「实时 GraphQL 订阅」这节课中我会学到什么？

通过基于 WebSocket 的 GraphQL 订阅向客户端推送实时更新，并关注背压问题。 你通过在浏览器中直接运行的动手代码来练习 FastAPI Backend Development Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 FastAPI Backend Development Bootcamp 需要有经验吗？

无需任何先前经验。CoddyKit 上的 FastAPI Backend Development Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。

「实时 GraphQL 订阅」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 FastAPI Backend Development Bootcamp 课中编写并运行代码吗？

能。每节 FastAPI Backend Development Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 定义类型、查询与变更
  2. 使用 DataLoaders 解决 N+1 查询
  3. 实时 GraphQL 订阅
  4. 查询成本分析与深度限制
← 返回 FastAPI Backend Development Bootcamp