0Pricing
FastAPI Backend Development Bootcamp · レッスン

リアルタイムGraphQLサブスクリプション

バックプレッシャーを考慮したWebSocketベースのGraphQLサブスクリプションで、クライアントにライブ更新をストリーミングします。

「リアルタイムGraphQLサブスクリプション」はCoddyKit上の無料FastAPI Backend Development Bootcampレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応の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サブスクリプション」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、FastAPI Backend Development Bootcampコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 FastAPI Backend Development Bootcampコースには全4レッスンが含まれています。

「リアルタイムGraphQLサブスクリプション」で何を学びますか？

バックプレッシャーを考慮したWebSocketベースのGraphQLサブスクリプションで、クライアントにライブ更新をストリーミングします。 ブラウザで直接実行するハンズオンコードでFastAPI Backend Development Bootcampを演習し、24時間対応の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. DataLoaderによるN+1クエリの解消
  3. リアルタイムGraphQLサブスクリプション
  4. クエリコスト分析と深さ制限
← FastAPI Backend Development Bootcampに戻る