Bootcamp i backendutveckling med FastAPI · Lektion

GraphQL-prenumerationer i realtid

Strömma liveuppdateringar till klienter via WebSocket-baserade GraphQL-prenumerationer med hänsyn till backpressure.

Lektion 3 av 413 steg

GraphQL-prenumerationer i realtid är en gratis lektion i Bootcamp i backendutveckling med FastAPI på CoddyKit. Detta är lektion 3 av 4. Du kan läsa vilka 3 lektioner som helst i den här lärvägen kostnadsfritt i sin helhet – därefter låser CoddyKit PRO upp alla lektioner, plus praktisk övning med en inbyggd kodredigerare och en AI-lärare dygnet runt. Den ingår i lärvägen för Bootcamp i backendutveckling med FastAPI, och Era framsteg synkroniseras mellan webben och CoddyKit-appen. Kursen i Bootcamp i backendutveckling med FastAPI innehåller totalt 4 lektioner.

Varför prenumerationer finns

GraphQL har tre typer av rotoperationer: query (läsning), mutation (skrivning) och subscription (live-ström). En prenumeration är den enda som håller en långlivad anslutning öppen och skickar flera resultat till klienten över tid.

Query/Mutation: en begäran, ett svar, anslutningen stängs.
Subscription: en begäran, en ström av svar tills någon av sidorna stänger anslutningen.

I Strawberry är en subscription-resolver en async-generator: varje yield blir en payload som levereras till klienten. Under huven transporterar FastAPI detta via en WebSocket, eftersom vanlig HTTP inte kan skicka serverinitierade frames på ett smidigt sätt.

En async-generator är kärnidén

Innan ni kopplar in GraphQL bör ni förstå motorn: Pythons async-generatorer. Varje yield överlämnar ett värde till konsumenten och pausar sedan tills konsumenten begär nästa. Detta naturliga pull-baserade flöde ger prenumerationer backpressure automatiskt: producenten går bara vidare när konsumenten är redo.

Detta kodexempel kan köras var som helst — det har exakt den form som en Strawberry subscription-resolver har, bortsett från dekoratorn.

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())

Er första Strawberry-prenumeration

En Strawberry-prenumeration finns på en typ som dekorerats med @strawberry.type och använder @strawberry.subscription på en async def som ger värden med yield. Returannotationen använder AsyncGenerator[T, None] så att schemat känner till payload-typen.

Här får en klient som prenumererar på count ett heltal var 500:e millisekund tills målvärdet har nåtts.

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)

Montera WebSocket-routen i FastAPI

Prenumerationer behöver en WebSocket-transport. Strawberry levereras med GraphQLRouter, som exponerar både HTTP-endpointen (för queries/mutationer) och WebSocket-endpointen (för prenumerationer) på samma sökväg.

Ni måste deklarera vilka WebSocket-subprotokoll som stöds. Det moderna är graphql-transport-ws (graphql-ws-biblioteket), och det äldre är graphql-ws (subscriptions-transport-ws). Erbjud båda för kompatibilitet med klienter.

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")

Vad klienten skickar

En prenumeration är ett GraphQL-dokument precis som alla andra — skillnaden är operationsnyckelordet. Klienten öppnar en WebSocket till /graphql, förhandlar om subprotokollet och skickar sedan ett subscribe-meddelande som innehåller detta dokument:

Servern svarar med ett next-meddelande per yield.
Servern skickar complete när generatorn är klar.
Vilken sida som helst kan skicka complete för att stoppa i förtid.

Urvalsmängdens struktur måste matcha payload-typen som er resolver ger upphov till.

subscription OnCount {
  count(target: 5)
}

Strömma domänhändelser, inte bara räknare

Riktiga prenumerationer skickar domänobjekt. Definiera en Strawberry-typ för payloaden och ge instanser av den som resultat med yield. Nedan ger ett flöde med orderstatus upphov till ett strukturerat objekt varje gång något ändras.

async for här skulle i produktion läsa från en riktig händelsekälla (en kö, Redis Pub/Sub, en Postgres-kanal med LISTEN/NOTIFY). Resolverns enda uppgift är att översätta dessa händelser till 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 med en asyncio Broadcast Queue

Många klienter prenumererar vanligtvis på samma händelse. En enda publicerare måste skicka vidare händelsen till N prenumeranter utan att en långsam prenumerant blockerar de andra. Det klassiska mönstret i processen är att varje prenumerant får sin egen asyncio.Queue, och att publiceraren lägger händelsen i varje kö.

Detta fristående exempel modellerar denna fan-out: två prenumeranter tömmer var och en sin egen kö med begränsad storlek, oberoende av varandra.

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: köer med begränsad storlek och policy för borttagning

Backpressure uppstår när en producent arbetar snabbare än en konsument. Med en obegränsad kö får en långsam klient minnet att växa utan gräns tills servern kraschar. Ni har tre ärliga strategier:

Blockera: använd en begränsad Queue och await q.put() — publiceraren saktar ned till den långsammaste konsumenten (säkert, men kopplar ihop klienterna).
Ta bort: när QueueFull inträffar kasserar ni det äldsta eller nyaste objektet — begränsad minnesanvändning, men data går förlorade (bra för telemetri och tickers).
Koppla från: om en klient hamnar efter mer än en viss gräns stänger ni dess prenumeration.

Kodexemplet nedan demonstrerar en icke-blockerande policy där det äldsta objektet tas bort från en begränsad kö.

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())

Städa vid frånkoppling med try/finally

När en klient stänger WebSocket-anslutningen (eller generatorn avbryts) kastar Strawberry asyncio.CancelledError in i er resolver vid det pausade yield-uttrycket. Om ni har allokerat en kö, en Redis-prenumeration eller en DB-LISTEN måste ni frigöra den — annars läcker resurser och broadcastaren fortsätter att skicka till en död kö.

Omslut loopen med try/finally. finally körs vid normalt slutförande, fel och avbrott.

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

Autentisera en prenumeration

WebSocket-anslutningar innehåller inte headers per begäran på samma sätt som HTTP gör, så ni autentiserar under handskakningen för connection init. Med graphql-transport-ws skickar klienten ett connection_init-meddelande med en payload (till exempel en token). Strawberry exponerar detta via en anpassad hook för on_ws_connect eller via anslutningsparametrarna i context.

Avvisa tidigt: kasta ett undantag i on_ws_connect för att neka socketen innan någon prenumeration startar. Auktorisera varje prenumeration i resolvern med den validerade identiteten från 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}

Filtrering och heartbeats

Två produktionsdetaljer håller prenumerationer stabila:

Filtrering på serversidan: strömma inte händelser som en klient inte ska se. Filtrera i resolvern före yield med den autentiserade user-identiteten och prenumerationens argument. Förlita er aldrig på att klienten kasserar dem.
Heartbeats / keepalive: inaktiva WebSockets avslutas av proxyservrar och lastbalanserare. Protokollet graphql-transport-ws har inbyggda ping/pong; ni kan även ge periodiska keepalive-payloads med yield. Konfigurera er reverse proxy (nginx) med en timeout för inaktivitet som är längre än ert heartbeat-intervall.

Denna resolver filtrerar efter tenant och vidarebefordrar endast relevanta händelser.

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"]

Snabb kontroll: hantera en långsam konsument

Ni kör en prenumeration för marknadsdata. En klients nätverk är långsamt och klarar inte händelsetakten. Andra klienter på samma publicerare är snabba. Vilken är den säkraste standarddesignen för att skydda servern och samtidigt hålla de snabba klienterna uppdaterade i realtid?

Besvara frågan i den interaktiva lektionen

Sammanfattning

Ni har byggt GraphQL i realtid över FastAPI med Strawberry:

Prenumerationer är async-generatorer: varje yield är en payload som skickas, och det pull-baserade protokollet ger naturlig backpressure.
Transport: GraphQLRouter hanterar queries/mutationer över HTTP och prenumerationer över WebSocket, och annonserar graphql-transport-ws (modernt) och graphql-ws (äldre).
Fan-out: ge varje prenumerant en egen kö så att en långsam klient aldrig blockerar de andra.
Backpressure: begränsa köernas storlek och välj en policy — blockera, ta bort eller koppla från. Använd aldrig obegränsade buffertar.
Livscykel: använd try/finally för att frigöra köer och externa prenumerationer vid avbrott.
Säkerhet och hälsa: autentisera under connection_init, filtrera händelser på serversidan och kör heartbeats så att proxyservrar inte avslutar inaktiva sockets.

Gratis att börja

Lär dig Bootcamp i backendutveckling med FastAPI med en AI-lärare – gratis

Skriv och kör riktig kod i webbläsaren, få omedelbar hjälp av en AI-lärare dygnet runt och fortsätt där du slutade – på webben eller i appen.

Kurser: 21
Lektioner: 84

Börja lära dig gratis Visa hela kursen

Vanliga frågor

Är lektionen ”GraphQL-prenumerationer i realtid” gratis?

Ja – du kan läsa vilka 3 lektioner som helst i lärvägen Bootcamp i backendutveckling med FastAPI, inklusive ”GraphQL-prenumerationer i realtid”, kostnadsfritt i sin helhet här på webben. Därefter låser CoddyKit PRO upp alla lektioner, plus interaktiv övning med en inbyggd kodredigerare och en AI-lärare dygnet runt. Kursen i Bootcamp i backendutveckling med FastAPI innehåller totalt 4 lektioner.

Vad lär jag mig i ”GraphQL-prenumerationer i realtid”?

Strömma liveuppdateringar till klienter via WebSocket-baserade GraphQL-prenumerationer med hänsyn till backpressure. Ni övar på Bootcamp i backendutveckling med FastAPI med praktisk kod som körs direkt i webbläsaren, medan en AI-handledare som är tillgänglig dygnet runt svarar på Era frågor under lektionen.

Behöver jag någon erfarenhet för att börja lära mig Bootcamp i backendutveckling med FastAPI?

Du behöver inga förkunskaper. Utbildningen i Bootcamp i backendutveckling med FastAPI på CoddyKit är upplagd för allt från nybörjare till avancerade elever, så att du kan börja här eller från början och gå fram i din egen takt. Detta är lektion 3 av 4.

Hur lång tid tar lektionen ”GraphQL-prenumerationer i realtid”?

De flesta CoddyKit-lektioner tar cirka 5–10 minuter. Varje lektion är kort och interaktiv, så att du gör stadiga framsteg och kan fortsätta precis där du slutade – på webben eller i appen.

Kan jag skriva och köra kod i den här Bootcamp i backendutveckling med FastAPI-lektionen?

Ja. Varje Bootcamp i backendutveckling med FastAPI-lektion innehåller en inbyggd kodredigerare, så att du kan skriva och köra riktig kod direkt i webbläsaren och få omedelbar AI-feedback – utan lokal installation.

Alla lektioner i den här kursen

Definiera typer, queries och mutationer
Lösning av N+1-queries med DataLoaders
GraphQL-prenumerationer i realtid
Analys av query-kostnad och djupbegränsning

← Tillbaka till Bootcamp i backendutveckling med FastAPI