Streaming, JSON Lines & SSE

import json
from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel


class Book(BaseModel):
    id: int
    title: str
    author: str | None = None


BOOKS = [Book(id=1, title="Dune", author="Frank Herbert"), Book(id=2, title="Emma"),
         Book(id=3, title="Ubik", author="Philip K. Dick")]

app = FastAPI()


@app.get("/books/stream")
async def stream_books() -> AsyncIterable[Book]:
    for book in BOOKS:
        yield book


@app.get("/books/stream-sync")
def stream_books_sync() -> Iterable[Book]:
    yield from BOOKS


@app.get("/books/stream-untyped")
async def stream_untyped():
    yield {"id": 9, "title": "Untyped", "extra": [1, 2]}


c = TestClient(app)
r = c.get("/books/stream")
print(r.status_code, r.headers["content-type"])
print(r.text, end="")
print([json.loads(line)["title"] for line in r.text.splitlines()])
print(c.get("/books/stream-sync").text == r.text)
print(c.get("/books/stream-untyped").text, end="")

200 application/jsonl
{"id":1,"title":"Dune","author":"Frank Herbert"}
{"id":2,"title":"Emma","author":null}
{"id":3,"title":"Ubik","author":"Philip K. Dick"}
['Dune', 'Emma', 'Ubik']
True
{"id": 9, "title": "Untyped", "extra": [1, 2]}

from typing import Annotated

from fastapi import FastAPI, Header
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.sse import EventSourceResponse, ServerSentEvent
from fastapi.testclient import TestClient

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=10)

REVIEWS = [{"id": i, "text": f"Review {i}"} for i in range(1, 6)]


@app.get("/books/{book_id}/reviews/stream", response_class=EventSourceResponse)
async def stream_reviews(
    book_id: int,
    last_event_id: Annotated[int | None, Header()] = None,
) -> AsyncIterable[ServerSentEvent]:
    start = last_event_id or 0
    for review in REVIEWS:
        if review["id"] <= start:
            continue
        yield ServerSentEvent(data=review, event="review", id=str(review["id"]))


c = TestClient(app)
r = c.get("/books/1/reviews/stream", headers={"Accept-Encoding": "gzip"})
print(r.headers.get("content-encoding"), [l for l in r.text.splitlines() if l.startswith("id:")])
r = c.get("/books/1/reviews/stream", headers={"Last-Event-ID": "3"})
print([l for l in r.text.splitlines() if l.startswith("id:")])
r = c.get("/books/1/reviews/stream", headers={"Last-Event-ID": "abc"})
print(r.status_code, r.json()["detail"][0]["msg"], r.json()["detail"][0]["loc"])

None ['id: 1', 'id: 2', 'id: 3', 'id: 4', 'id: 5']
['id: 4', 'id: 5']
422 Input should be a valid integer, unable to parse string as an integer ['header', 'last-event-id']

import asyncio
from collections import deque
from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import FastAPI, Header
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel


class ReviewIn(BaseModel):
    stars: int
    text: str


class ReviewOut(ReviewIn):
    id: int
    book_id: int


class ReviewFeed:
    """In-memory fan-out: a bounded replay buffer plus one queue per subscriber."""

    def __init__(self, history: int = 100) -> None:
        self.history: deque[ReviewOut] = deque(maxlen=history)
        self.subscribers: set[asyncio.Queue[ReviewOut]] = set()
        self.next_id = 1

    def publish(self, book_id: int, review: ReviewIn) -> ReviewOut:
        item = ReviewOut(id=self.next_id, book_id=book_id, **review.model_dump())
        self.next_id += 1
        self.history.append(item)
        for queue in self.subscribers:
            queue.put_nowait(item)
        return item

    async def subscribe(self, after_id: int | None) -> AsyncIterable[ReviewOut]:
        queue: asyncio.Queue[ReviewOut] = asyncio.Queue(maxsize=100)
        replay = [r for r in self.history if after_id is not None and r.id > after_id]
        self.subscribers.add(queue)  # register before replaying so nothing is missed
        try:
            for item in replay:
                yield item
            while True:
                yield await queue.get()
        finally:
            self.subscribers.discard(queue)


feed = ReviewFeed()
app = FastAPI()


@app.post("/books/{book_id}/reviews", status_code=201)
async def add_review(book_id: int, review: ReviewIn) -> ReviewOut:
    return feed.publish(book_id, review)


@app.get("/reviews/stream", response_class=EventSourceResponse)
async def review_stream(
    last_event_id: Annotated[int | None, Header()] = None,
) -> AsyncIterable[ServerSentEvent]:
    async for review in feed.subscribe(last_event_id):
        yield ServerSentEvent(data=review, event="review", id=str(review.id), retry=2000)

import json

import httpx
import uvicorn


async def main() -> None:
    server = uvicorn.Server(uvicorn.Config(app, port=18766, log_level="warning"))
    serving = asyncio.create_task(server.serve())
    while not server.started:
        await asyncio.sleep(0.01)
    async with httpx.AsyncClient(base_url="http://127.0.0.1:18766") as client:
        await client.post("/books/1/reviews", json={"stars": 5, "text": "missed while offline"})

        async def listen(label: str, headers: dict[str, str], count: int) -> None:
            async with client.stream("GET", "/reviews/stream", headers=headers) as r:
                seen = 0
                async for line in r.aiter_lines():
                    if line.startswith("data:"):
                        print(label, json.loads(line[5:])["text"])
                        seen += 1
                    if seen == count:
                        return

        first = asyncio.create_task(listen("A", {}, 1))
        await asyncio.sleep(0.1)
        print("subscribers:", len(feed.subscribers))
        await client.post("/books/1/reviews", json={"stars": 4, "text": "live review"})
        await first
        await client.post("/books/2/reviews", json={"stars": 3, "text": "sent during the outage"})
        await asyncio.wait_for(listen("A reconnected", {"Last-Event-ID": "2"}, 1), 3)
        await asyncio.sleep(0.1)
        print("subscribers after disconnects:", len(feed.subscribers))
    server.should_exit = True
    await serving


asyncio.run(main())

subscribers: 1
A live review
A reconnected sent during the outage
subscribers after disconnects: 0

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from fastapi.testclient import TestClient
from pydantic import BaseModel


class Prompt(BaseModel):
    text: str


@app.post("/assistant/stream", response_class=EventSourceResponse)
async def stream_answer(prompt: Prompt) -> AsyncIterable[ServerSentEvent]:
    for word in prompt.text.split():
        yield ServerSentEvent(data=word, event="token")
    yield ServerSentEvent(raw_data="[DONE]", event="done")


r = TestClient(app).post("/assistant/stream", json={"text": "Dune is great"})
print(r.headers["content-type"])
print(r.text)

text/event-stream; charset=utf-8
event: token
data: "Dune"

event: token
data: "is"

event: token
data: "great"

event: done
data: [DONE]

from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import FastAPI, Header
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()
REVIEWS = [f"Review {i}" for i in range(1, 6)]


@app.get("/reviews/stream", response_class=EventSourceResponse)
async def stream(last_event_id: Annotated[int | None, Header()] = None) -> AsyncIterable[ServerSentEvent]:
    start = last_event_id or 0
    # Send at most two events per connection, then end the stream: the client reconnects.
    for n, text in enumerate(REVIEWS[start:start + 2], start=start + 1):
        yield ServerSentEvent(data={"id": n, "text": text}, event="review", id=str(n), retry=200)

const source = new EventSource("http://127.0.0.1:8000/reviews/stream");
let opens = 0;
source.onopen = () => console.log(`open #${++opens}`);
source.addEventListener("review", (e) => {
  console.log(`review id=${e.lastEventId} data=${e.data}`);
  if (e.lastEventId === "5") { source.close(); console.log("closed"); clearInterval(keepAlive); }
});
source.onerror = () => console.log("error: connection lost, browser will retry");
const keepAlive = setInterval(() => {}, 1000); // Node exits while the reconnect timer is pending

open #1
review id=1 data={"id": 1, "text": "Review 1"}
review id=2 data={"id": 2, "text": "Review 2"}
error: connection lost, browser will retry
open #2
review id=3 data={"id": 3, "text": "Review 3"}
review id=4 data={"id": 4, "text": "Review 4"}
error: connection lost, browser will retry
open #3
review id=5 data={"id": 5, "text": "Review 5"}
closed

curl -sN -i -H "Last-Event-ID: 2" http://127.0.0.1:8000/reviews/stream

HTTP/1.1 200 OK
date: Fri, 02 Oct 2026 08:41:31 GMT
server: uvicorn
content-type: text/event-stream; charset=utf-8
cache-control: no-cache
x-accel-buffering: no
transfer-encoding: chunked

event: review
data: {"id": 3, "text": "Review 3"}
id: 3
retry: 200

event: review
data: {"id": 4, "text": "Review 4"}
id: 4
retry: 200

import asyncio
from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()
log: list[str] = []


@app.get("/clock", response_class=EventSourceResponse)
async def clock() -> AsyncIterable[ServerSentEvent]:
    log.append("stream opened")
    try:
        n = 0
        while True:
            n += 1
            yield ServerSentEvent(data={"tick": n}, id=str(n))
            await asyncio.sleep(0.05)
    except asyncio.CancelledError:
        log.append("cancelled (client went away)")
        raise
    finally:
        log.append("cleanup done")


async def main() -> None:
    import httpx
    import uvicorn

    server = uvicorn.Server(uvicorn.Config(app, port=18765, log_level="warning"))
    task = asyncio.create_task(server.serve())
    while not server.started:
        await asyncio.sleep(0.01)
    async with httpx.AsyncClient(base_url="http://127.0.0.1:18765") as client:
        async with client.stream("GET", "/clock") as response:
            async for line in response.aiter_lines():
                print("client got:", line)
                if line.startswith("id: 3"):
                    break  # leaving the block closes the connection
    await asyncio.sleep(0.3)
    print(log)
    server.should_exit = True
    await task


asyncio.run(main())

client got: data: {"tick": 1}
client got: id: 1
client got:
client got: data: {"tick": 2}
client got: id: 2
client got:
client got: data: {"tick": 3}
client got: id: 3
['stream opened', 'cancelled (client went away)', 'cleanup done']

import asyncio
from collections.abc import AsyncIterable
from typing import Annotated

import httpx
import uvicorn
from fastapi import Depends, FastAPI, HTTPException
from fastapi.sse import EventSourceResponse, ServerSentEvent

BOOKS = {1: "Dune"}
app = FastAPI()


def get_book(book_id: int) -> str:
    if book_id not in BOOKS:
        raise HTTPException(status_code=404, detail="Book not found")
    return BOOKS[book_id]


BookDep = Annotated[str, Depends(get_book)]


@app.get("/checked/{book_id}/stream", response_class=EventSourceResponse)
async def checked(book: BookDep) -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(data={"title": book})


@app.get("/late/{book_id}/stream", response_class=EventSourceResponse)
async def late(book_id: int) -> AsyncIterable[ServerSentEvent]:
    if book_id not in BOOKS:
        raise HTTPException(status_code=404, detail="Book not found")  # runs inside the stream
    yield ServerSentEvent(data={"title": BOOKS[book_id]})


@app.get("/failing", response_class=EventSourceResponse)
async def failing() -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(data="first")
    raise RuntimeError("database went away")


async def main() -> None:
    server = uvicorn.Server(uvicorn.Config(app, port=18768, log_level="critical"))
    serving = asyncio.create_task(server.serve())
    while not server.started:
        await asyncio.sleep(0.01)
    async with httpx.AsyncClient(base_url="http://127.0.0.1:18768") as c:
        for path in ("/checked/2/stream", "/late/2/stream", "/failing"):
            try:
                async with c.stream("GET", path) as r:
                    body = [line async for line in r.aiter_lines() if line]
                    print(path, r.status_code, body)
            except httpx.RemoteProtocolError as exc:
                print(path, "status was", r.status_code, "then:", type(exc).__name__)
    server.should_exit = True
    await serving


asyncio.run(main())

/checked/2/stream 404 ['{"detail":"Book not found"}']
/late/2/stream status was 200 then: RemoteProtocolError
/failing status was 200 then: RemoteProtocolError

location /reviews/stream {
    proxy_pass http://bookshelf_upstream;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_buffering off;       # redundant for SSE (the X-Accel-Buffering header does it), needed for JSON Lines
    proxy_read_timeout 1h;     # longer than the longest silence between events
}

import asyncio
from collections.abc import AsyncIterable, Iterator
from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from fastapi.testclient import TestClient

events: list[str] = []


class FakeSession:
    def __init__(self) -> None:
        self.closed = False

    def reviews(self) -> list[str]:
        if self.closed:
            raise RuntimeError("session is closed")
        return ["r1", "r2", "r3"]


def get_session() -> Iterator[FakeSession]:
    session = FakeSession()
    events.append("session opened")
    try:
        yield session
    finally:
        session.closed = True
        events.append("session closed")


SessionDep = Annotated[FakeSession, Depends(get_session)]
EarlySessionDep = Annotated[FakeSession, Depends(get_session, scope="function")]

app = FastAPI()


@app.get("/reviews/stream", response_class=EventSourceResponse)
async def stream(session: SessionDep) -> AsyncIterable[ServerSentEvent]:
    for row in session.reviews():
        events.append(f"yield {row}")
        yield ServerSentEvent(data=row)
    events.append("generator finished")


@app.get("/reviews/stream-early", response_class=EventSourceResponse)
async def stream_early(session: EarlySessionDep) -> AsyncIterable[ServerSentEvent]:
    events.append("generator started")
    await asyncio.sleep(0.05)  # simulate waiting for the next review
    for row in session.reviews():
        yield ServerSentEvent(data=row)


c = TestClient(app)
c.get("/reviews/stream")
print(events)
events.clear()
try:
    c.get("/reviews/stream-early")
except ExceptionGroup as group:
    print("ExceptionGroup ->", repr(group.exceptions[0]))
print(events)

['session opened', 'yield r1', 'yield r2', 'yield r3', 'generator finished', 'session closed']
ExceptionGroup -> RuntimeError('session is closed')
['session opened', 'generator started', 'session closed']

import asyncio
from collections.abc import AsyncIterable, Iterator
from contextlib import contextmanager
from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient

app = FastAPI()
opened: list[str] = []


class Db:
    @contextmanager
    def session(self) -> Iterator[list[str]]:
        opened.append("open")
        try:
            yield ["r1", "r2"]
        finally:
            opened.append("close")


def get_db() -> Db:
    return Db()


@app.get("/reviews/poll", response_class=StreamingResponse)
async def poll(db: Annotated[Db, Depends(get_db)]) -> AsyncIterable[str]:
    for _ in range(2):
        with db.session() as rows:  # short-lived session per poll, never held across waits
            yield ",".join(rows) + "\n"
        await asyncio.sleep(0.01)


c = TestClient(app)
print(c.get("/reviews/poll").text.splitlines(), opened)

['r1,r2', 'r1,r2'] ['open', 'close', 'open', 'close']

{
  "description": "Successful Response",
  "content": {
    "text/event-stream": {
      "itemSchema": {
        "type": "object",
        "properties": {
          "data": {"type": "string"},
          "event": {"type": "string"},
          "id": {"type": "string"},
          "retry": {"type": "integer", "minimum": 0}
        }
      }
    }
  }
}