Serving an API with FastAPI¶

A Turbine app can run inside a FastAPI app. The web server and the stream processor share one process: your routes read what Turbine computes — open windows, state, the cluster's nodes — through an in-process client, with no second service to deploy and no HTTP call back into Turbine.

pip install 'turbine-fast[fastapi]'

Example: a risk profile per contributor¶

A stream of scored Wikipedia edits arrives on wiki.edits.scored — who edited, on which wiki, the model's risk score, and whether the edit was later reverted. (The score could come from Turbine too, upstream, with the inference operator.) Turbine keeps, per contributor, a running profile (edits, reverts, highest risk) and the mean risk over the last hour. FastAPI serves three business routes on top: a contributor's profile, a verdict combining both, and the pipeline's health.

The app¶

One file, one process:

import pyarrow as pa
import uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from turbine import KafkaBroker, RecordBatch, Turbine, TurbineState
from turbine import aggregates as agg
from turbine import windowing as win
from turbine.client import NotFoundError
from turbine.fastapi import Cluster, lifespan


class Edit(BaseModel):
    user: str
    wiki: str
    rev_id: int
    risk: float  # the model's score, 0..1
    reverted: bool


kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka, app_id="wiki-risk", state_dir="/tmp/wiki-risk", from_latest=True)


@app.subscribe(kafka.topic("wiki.edits.scored", schema=Edit), partition_key="user")
class ScoreUsers:
    """A profile per contributor, and their mean risk over the last hour."""

    def __init__(self) -> None:
        self.trend = win.PersistentSliding(
            self,
            name="risk_trend",
            size_ms=3_600_000,  # 1 h
            slide_ms=300_000,  # every 5 min
            key=["user"],
            value=agg.Mean("risk"),
            on_close=self.closed,
        )

    def process(self, batch: RecordBatch, state: TurbineState) -> None:
        self.trend.update(batch, state)
        for row in batch.to_pylist():
            user = state.scope(row["user"])
            profile = user.get("profile") or {"edits": 0, "reverted": 0, "max_risk": 0.0}
            profile["edits"] += 1
            profile["reverted"] += int(row["reverted"])
            profile["max_risk"] = max(profile["max_risk"], row["risk"])
            profile["last_rev"] = row["rev_id"]
            user.put("profile", profile)

    def closed(self, state: TurbineState, panes: pa.Table) -> None:
        return None  # the routes read the open windows; nothing to publish


api = FastAPI(title="Wiki risk", lifespan=lifespan(app))


@api.get("/users/{user}")
async def profile(user: str, cluster: Cluster) -> dict:
    """The profile, as the handler wrote it."""
    try:
        return await cluster.get("ScoreUsers", "profile", partition_key=user)
    except NotFoundError:
        raise HTTPException(404, f"no edit known for {user}")


@api.get("/users/{user}/verdict")
async def verdict(user: str, cluster: Cluster) -> dict:
    """Business logic lives in the route: combine the state and the window."""
    try:
        p = await cluster.get("ScoreUsers", "profile", partition_key=user)
    except NotFoundError:
        raise HTTPException(404, f"no edit known for {user}")
    try:
        trend = await cluster.window_series("risk_trend", group={"user": user})
        recent = trend.get("aggregate")
    except NotFoundError:
        recent = None  # no edit in the last hour
    revert_rate = p["reverted"] / p["edits"]
    watch = revert_rate > 0.3 or (recent or 0) > 0.7
    return {
        "user": user,
        "edits": p["edits"],
        "revert_rate": round(revert_rate, 2),
        "mean_risk_1h": recent,
        "verdict": "watch" if watch else "ok",
    }


@api.get("/pipeline")
async def pipeline(cluster: Cluster) -> dict:
    nodes = (await cluster.nodes())["nodes"]
    return {"nodes": [{"id": n["node_id"], "status": n["status"]} for n in nodes]}


if __name__ == "__main__":
    uvicorn.run(api, host="0.0.0.0", port=8000)

What each piece does:

partition_key="user" — records are routed, and state is sharded, by contributor. state.scope(row["user"]) writes under that contributor.
user.put("profile", ...) stores a plain dict; Turbine encodes it with the app's state_format (JSON by default) and cluster.get hands it back decoded.
risk_trend is a sliding window: the mean of risk per contributor over the last hour, refreshed every five minutes. cluster.window_series reads its open panes and their aggregate.
The business rule lives in the route (verdict): Turbine stores and aggregates, your code decides what "suspicious" means.

Run it¶

pip install 'turbine-fast[fastapi]'
export TURBINE_API_TOKEN="$(openssl rand -hex 32)"
python app.py

Two ports are open: 8000 is your API, the one to publish; 8400 is Turbine's own API and console, which the other nodes of a cluster use — keep it internal, protected by the token (see Ports and security).

Feed it¶

Any producer works. With Redpanda's rpk, for instance:

rpk topic produce wiki.edits.scored <<'EOF'
{"user": "Alice", "wiki": "enwiki", "rev_id": 1001, "risk": 0.12, "reverted": false}
{"user": "Alice", "wiki": "enwiki", "rev_id": 1002, "risk": 0.31, "reverted": false}
{"user": "Alice", "wiki": "frwiki", "rev_id": 1003, "risk": 0.22, "reverted": false}
{"user": "Mallory", "wiki": "enwiki", "rev_id": 2001, "risk": 0.88, "reverted": true}
{"user": "Mallory", "wiki": "enwiki", "rev_id": 2002, "risk": 0.93, "reverted": true}
{"user": "Mallory", "wiki": "enwiki", "rev_id": 2003, "risk": 0.41, "reverted": false}
EOF

The app starts with from_latest=True: it reads what arrives after it starts.

Query it¶

$ curl localhost:8000/users/Alice
{"edits":3,"last_rev":1003,"max_risk":0.31,"reverted":0}

$ curl localhost:8000/users/Mallory/verdict
{"user":"Mallory","edits":3,"revert_rate":0.67,"mean_risk_1h":0.7400000000000001,"verdict":"watch"}

$ curl localhost:8000/users/Nobody
{"detail":"no edit known for Nobody"}          # HTTP 404

A value is readable once the batch that wrote it is committed — a fraction of a second after the record arrives.

In a cluster¶

Run the same file on every node, with the usual cluster settings and the same TURBINE_API_TOKEN, and put the nodes' port 8000 behind your load balancer. A request can land on any node:

cluster.get("ScoreUsers", "profile", partition_key="Alice") reads locally if this node holds Alice, otherwise goes straight to the node that owns her partition, and only asks every node if that one does not have her.
cluster.window_series(...) does the same for the window.
Stopping a node (SIGTERM) stops the web server, which stops Turbine: batches in flight finish, state is checkpointed, and the node leaves the cluster before the process exits.

Writing routes¶

The subscription name passed to get is the handler's name — the class or function decorated with @app.subscribe — or the one given with @app.subscribe(name=...). Give it an explicit name= if you may rename the class: it also names the subscription's consumer group.
partition_key is the value you scoped the write with (state.scope(value)). A key written without a scope is held separately by each partition; with several partitions, read it with worker_id= or keep such values scoped.
Absent is an exception, not None: get raises NotFoundError for a key nobody wrote, and window_series for a group with no open window (for instance, no edit in the last hour). Turn it into the answer your API means — a 404, or a default, as verdict does.
An unavailable read keeps its last value: get, get_entry and window_series raise UnavailableError (503) when nothing was found but some nodes could not answer. Its unreachable and unauthorized lists name those nodes. Retry without clearing the last value or series shown to users. This includes startup and worker recovery: entering the lifespan and having an answering HTTP server do not mean every worker's state is readable yet.
Values are what the handler stored, at the moment of the last committed batch. Your routes only read: nothing an HTTP client does can change Turbine's state.

What the client reads¶

The client reads the whole cluster, exactly what the /cluster/* routes of the REST API serve, and returns the same JSON bodies as Python dicts. The one you will use most is get (see the example): it returns what your handler wrote with state.scope(partition_key).put(key, value), decoded, from whichever node holds it. The full list:

Method	Same answer as
`await cluster.get(subscription, key, partition_key=...)`	the `value` of `POST /cluster/state/key`
`await cluster.get_entry(subscription, key, partition_key=...)`	`POST /cluster/state/key` (with the worker, and whether it decoded)
`await cluster.window_series(name, group={...})`	`GET /cluster/windows/{name}/series`
`await cluster.windows(q=..., kind=..., limit=...)`	`GET /cluster/windows`
`await cluster.state(prefix=..., partition_key=..., ...)`	`POST /cluster/state`
`await cluster.nodes()`	`GET /cluster/nodes`
`await cluster.topology()`	`GET /topology`

In a cluster, the node that received the HTTP request asks the other nodes for their part, like the REST routes do — each node's answer covers only the partitions it owns, and the client merges them. Nodes that could not answer are listed under unreachable (or unauthorized, for a node that refuses the API token) instead of being silently missing.

Errors raise turbine.client.NotFoundError (404, absence confirmed by all queried nodes), UnavailableError (503, the read could not determine absence), BadRequestError, or their parent ClusterApiError. Each carries the HTTP status and a message. UnavailableError also carries unreachable and unauthorized lists; these can both be empty for local startup or if the app stopped while a request was pending. Local store failures raise ClusterApiError with status 500.

Every method is a coroutine: call it from a route (or anything else running on an asyncio loop). The work runs on Turbine's own threads, so a slow answer — waiting on another node — never blocks the web server's other requests. Handlers inside Turbine (process, window callbacks) do not use the client: they already have their own state.

Combining with your own lifespan¶

Your app often has startup work of its own — a background task, a database pool, a consumer feeding a live page. lifespan(app) is an ordinary FastAPI lifespan, so you nest it inside yours instead of rewriting it:

import asyncio
from contextlib import asynccontextmanager

from turbine.fastapi import lifespan as turbine_lifespan


@asynccontextmanager
async def my_lifespan(api: FastAPI):
    task = asyncio.create_task(my_background_job())
    try:
        async with turbine_lifespan(app)(api) as state:  # {"turbine": client}
            yield {**state, "job": task}
    finally:
        task.cancel()


api = FastAPI(lifespan=my_lifespan)

Keep the "turbine" key in the state you yield: it is where the Cluster dependency finds the client. On shutdown Turbine stops first, then your task is cancelled.

Live pages (server-sent events)¶

A route can stream updates to a browser with server-sent events, reading the cluster on each tick:

import asyncio

from fastapi import Request
from fastapi.responses import StreamingResponse
from turbine.client import UnavailableError


@api.get("/users/{user}/live")
async def live(user: str, request: Request, cluster: Cluster) -> StreamingResponse:
    async def events():
        while not await request.is_disconnected():
            try:
                profile = await cluster.get("ScoreUsers", "profile", partition_key=user)
                yield f"data: {json.dumps(profile)}\n\n"
            except NotFoundError:
                yield "data: null\n\n"
            except UnavailableError:
                pass  # Retry; the browser keeps the last value it received.
            await asyncio.sleep(1)

    return StreamingResponse(events(), media_type="text/event-stream")

No token needed in the browser. Turbine's token protects Turbine's own port, not your routes, and the client reads in process — so an EventSource, which cannot send an Authorization header, works as is. Protect your own routes as you see fit (a session cookie, a token in the query string).
Many open streams do not slow the web server down. Each cluster call runs on Turbine's threads and only comes back to the event loop to deliver its answer. As a rough order of magnitude (not a contract), thousands of reads per second from dozens of open streams leave the event loop as responsive as when it is idle. What can block it is your own code inside the stream: heavy computation or a blocking call without await.

Signals and shutdown¶

The web server owns Ctrl-C and SIGTERM. Stopping it stops Turbine: in-flight batches finish, workers checkpoint, and — in a cluster — the node announces its stop and hands leadership over if needed, before the process exits. It stays a voter so a rolling restart does not remove and re-add it.

During startup, Turbine may still be waiting for cluster admission. When the lifespan runs on Python's main thread, Ctrl-C and SIGTERM also request Turbine's shutdown while that wait is pending, and the web server still receives its signal. Startup then fails and joins the Turbine thread. After startup, the relay stays active for the lifespan: the first signal immediately makes /ready return 503 with status: "unavailable" and reason: "stopping". /health stays 200. Turbine keeps processing and answering in-process reads until the lifespan exits, so open responses can drain. Handlers are restored on exit, including failure or cancellation. The host's default and ignored signal policies remain in effect.

If the lifespan runs in another thread, the host must cancel its startup task when stopping, and withdraw readiness through POST /ready/stop before closing its listener: Python only allows signal handlers to be installed on the main thread.

Long-lived responses delay all of that. On shutdown, uvicorn first waits for open connections to finish, and only then runs the lifespan's exit — which is what stops Turbine. A server-sent-events stream never finishes on its own, so with uvicorn's defaults a single open browser tab holds the shutdown forever: Turbine never gets to its final checkpoint, and in Kubernetes the pod is killed once its grace period runs out. Nothing is lost (processing resumes from the last checkpoint), but the final checkpoint is skipped and everything since is processed again.

Bound the wait, so open streams are cut and Turbine stops cleanly:

uvicorn.run(api, host="0.0.0.0", port=8000, timeout_graceful_shutdown=5)

(--timeout-graceful-shutdown 5 on the uvicorn command line.) After a SIGTERM, open streams are closed after 5 seconds, then Turbine stops and writes its final checkpoint. In Kubernetes, leave a clear margin above that for Turbine's own shutdown — for instance terminationGracePeriodSeconds: 30 — so the pod is never killed in the middle of it. Include the preStop duration in that budget too. Browsers can reconnect an EventSource to another pod once the load balancer has withdrawn the stopping pod. A failed reconnection can leave an EventSource closed, so do not rely on automatic reconnection alone to hide routing failures. In Kubernetes, use the traffic withdrawal hook to let readiness and load-balancer updates propagate before uvicorn receives SIGTERM and closes its listener.

Uvicorn logs a CancelledError ("timeout graceful shutdown exceeded") for each stream it cuts: that is the bound doing its job, not a failure.

Ports and security¶

Two ports are open: the web server's (your routes) and Turbine's own API port (8400 by default), which the other nodes of a cluster use and which serves the console. Only the first needs to be public. Keep Turbine's port on the internal network and set a token on it (TURBINE_API_TOKEN) — see Securing the API. Your routes are ordinary FastAPI routes: protect them, and set their CORS policy, with FastAPI's own tools (CORSMiddleware, dependencies); Turbine's CORS setting only covers its own port.

Without FastAPI¶

lifespan is a thin wrapper. Any asyncio program can do the same with Turbine.running():

async with app.running() as cluster:
    nodes = await cluster.nodes()

In a FastAPI lifespan, app.running() is what lifespan(app) uses; you can enter it directly and put the client in the state under "turbine" yourself (see Combining with your own lifespan).

Startup waits for Turbine's API to serve and, in cluster mode, for the cluster to accept this app's subscription names, topics and declaration order. A refused declaration fails server startup. The internal API remains available during admission so nodes can communicate; that alone does not complete embedded startup. Startup does not wait for every partition to be healthy: use the health and readiness probes to monitor processing.

If the Turbine app thread fails after startup, the error is raised when leaving app.running() or during FastAPI lifespan shutdown, with the original error as its cause. An exception from your own context body takes precedence and receives an exception note describing the Turbine failure. App failures do not automatically interrupt the running web server; monitor the probes while it serves. Shutdown always stops Turbine and joins its thread.