0Pricing
FastAPI Backend Development Bootcamp · 강의

실시간 GraphQL 구독

백프레셔를 고려한 WebSocket 기반 GraphQL 구독으로 클라이언트에 실시간 업데이트를 스트리밍합니다.

실시간 GraphQL 구독은(는) CoddyKit의 무료 FastAPI Backend Development Bootcamp 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 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/7 AI 튜터), CoddyKit PRO로 업그레이드하면 FastAPI Backend Development Bootcamp 강의 전체를 잠금 해제할 수 있습니다. FastAPI Backend Development Bootcamp 강의에는 총 4개의 강의가 포함되어 있습니다.

“실시간 GraphQL 구독”에서 뭘 배우나요?

백프레셔를 고려한 WebSocket 기반 GraphQL 구독으로 클라이언트에 실시간 업데이트를 스트리밍합니다. 브라우저에서 직접 실행하는 실습 코드로 FastAPI Backend Development Bootcamp을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

FastAPI Backend Development Bootcamp을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 FastAPI Backend Development Bootcamp은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.

“실시간 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(으)로 돌아가기