PaginationStreaming¶
An activity log sent whole, without pages: one JSON object per line, read from the database a hundred events at a time by id, each batch sent as soon as it's read. A second endpoint sends the 25 newest events from a synchronous cursor, and closes it there. The strategy: PaginationStreaming.
Run it¶
From a clone of the repository:
make deps # .venv, with every extra
source .venv/bin/activate
uvicorn examples.pagination.streaming:app
Try it¶
Every event, one per line, as they come (-N turns curl's buffering off):
$ curl -iN http://localhost:8000/events | head -10
HTTP/1.1 200 OK
content-type: application/x-ndjson
Transfer-Encoding: chunked
{"id":1,"kind":"upload","at":"2026-10-29T09:01:00Z"}
{"id":2,"kind":"logout","at":"2026-10-29T09:02:00Z"}
{"id":3,"kind":"login","at":"2026-10-29T09:03:00Z"}
{"id":4,"kind":"upload","at":"2026-10-29T09:04:00Z"}
All 250 of them, from three queries of up to 100 rows; and the logins alone, from one:
$ curl -s http://localhost:8000/events | wc -l
250
$ curl -s "http://localhost:8000/events?kind=login" | head -3
{"id":3,"kind":"login","at":"2026-10-29T09:03:00Z"}
{"id":6,"kind":"login","at":"2026-10-29T09:06:00Z"}
{"id":9,"kind":"login","at":"2026-10-29T09:09:00Z"}
$ curl -s "http://localhost:8000/events?kind=login" | wc -l
83
The newest events end at 25, however many there are:
$ curl -s http://localhost:8000/events/latest | head -3
{"id":250,"kind":"upload","at":"2026-10-29T13:10:00Z"}
{"id":249,"kind":"login","at":"2026-10-29T13:09:00Z"}
{"id":248,"kind":"logout","at":"2026-10-29T13:08:00Z"}
$ curl -s http://localhost:8000/events/latest | wc -l
25
The parameters are checked before anything is streamed:
$ curl "http://localhost:8000/events?kind=anything"
{"detail":[{"type":"enum","loc":["query","kind"],"msg":"Input should be 'login', 'upload' or 'logout'","input":"anything","ctx":{"expected":"'login', 'upload' or 'logout'"}}]}
What to look at¶
- No pages: one request, one response, every event. The status and headers go out first, and
the lines follow as they're read, in chunks of
batch_sizelines;curl -Nshows them as they arrive. - Read by key:
batches(fetch, key=...)asks the database for 100 events, then for the 100 after the last id it got, and stops at the first batch shorter than 100: three queries for 250 events, each one through the index on id, as fast as the first. The whole log is never in memory. - Sync or async:
fetchhere is async; the cursor of/events/latestis a synchronous generator, which the stream reads in the thread pool, ten events at a time, so a blocking driver doesn't stop the event loop. max_items:/events/latestends at 25, and the cursor is closed there (itsfinallyruns) rather than read to the end.- A client that disconnects stops the stream: no more queries are made for it.
- Not cached:
@cachedoesn't keep streams, and each request reads the database again. Lists that are read often, a page at a time, are better cached by page (PaginationCursor).
The code¶
examples/pagination/streaming.py
"""An activity log streamed as NDJSON, one event per line, without pages and without loading it.
uvicorn examples.pagination.streaming:app
- `/events`: every event (or every event of a kind), read from the database 100 at a time by
id, each batch sent as soon as it's read. A client that stops reading stops the queries.
- `/events/latest`: the 25 newest, from a synchronous cursor that the stream reads in the thread
pool and closes at the 25th.
Streams aren't cached: each request reads the database again.
"""
from collections.abc import AsyncIterator, Iterator
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from enum import StrEnum
from typing import Annotated
import uvicorn
from fastapi import Depends, FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from fastapi_alcazaba.pagination.streaming import PaginationStreaming
class Kind(StrEnum):
LOGIN = "login"
UPLOAD = "upload"
LOGOUT = "logout"
class Event(BaseModel):
id: int
kind: Kind
at: datetime
class Database:
"""Stands in for the database (a session per request in a real app)."""
def __init__(self) -> None:
kinds, start = list(Kind), datetime(2026, 10, 29, 9, 0, tzinfo=UTC)
self.events = [
Event(id=i, kind=kinds[i % len(kinds)], at=start + timedelta(minutes=i))
for i in range(1, 251)
]
self.queries = 0
self.open_cursors = 0
async def events_after(
self,
after: int | None,
limit: int,
kind: Kind | None,
) -> list[Event]:
"""What a database does: WHERE id > :after AND kind = :kind ORDER BY id LIMIT :limit."""
self.queries += 1
found = [
event
for event in self.events
if (after is None or event.id > after) and (kind is None or event.kind == kind)
]
return found[:limit]
def newest(self) -> Iterator[Event]:
"""A synchronous driver's cursor, newest first: the rows come as they're read."""
self.queries += 1
self.open_cursors += 1
try:
yield from reversed(self.events)
finally:
self.open_cursors -= 1
@asynccontextmanager
async def lifespan(application: FastAPI) -> AsyncIterator[None]:
application.state.db = Database()
yield
app = FastAPI(lifespan=lifespan)
def get_db(request: Request) -> Database:
db: Database = request.app.state.db
return db
DatabaseDep = Annotated[Database, Depends(get_db)]
# 100 lines a chunk, and 100 rows a query.
stream = PaginationStreaming(batch_size=100)
latest = PaginationStreaming(batch_size=10, max_items=25)
@app.get("/events")
async def events(
db: DatabaseDep,
kind: Kind | None = None,
) -> StreamingResponse:
"""Every event, by id: a query per batch, past the last id of the batch before."""
async def fetch(
after: int | None,
limit: int,
) -> list[Event]:
return await db.events_after(after, limit, kind)
return stream.response(stream.batches(fetch, key=lambda event: event.id))
@app.get("/events/latest")
async def latest_events(db: DatabaseDep) -> StreamingResponse:
"""The 25 newest events: the stream stops reading the cursor there, and closes it."""
return latest.response(db.newest())
if __name__ == "__main__":
uvicorn.run("streaming:app", reload=True)