PaginationCursor

What it's for: feeds and infinite scroll, whose pages don't shift when items arrive, with signed cursors both ways that clients can't make up nor reuse with other filters.

An activity feed by cursor, newest first: each next_cursor gives the events right after the last one seen, however many were added since, and prev_cursor the ones before the first. The strategy: PaginationCursor.

Run it

From a clone of the repository:

make deps                 # .venv, with every extra
source .venv/bin/activate
uvicorn examples.pagination.cursor:app

Set CURSOR_SECRET (secrets.token_urlsafe(32)), the same in every process. Without it each start makes up a throwaway one, and the cursors stop working when the app restarts.

Try it

The three newest events, and where the next page starts. There's nothing before the first page, so prev_cursor is null:

$ curl "http://localhost:8000/events?limit=3"
{"items":[{"id":30,"created_at":"2026-10-29T09:30:00Z","text":"Event 30"},{"id":29,"created_at":"2026-10-29T09:29:00Z","text":"Event 29"},{"id":28,"created_at":"2026-10-29T09:28:00Z","text":"Event 28"}],"limit":3,"prev_cursor":null,"next_cursor":"WyJhIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyODowMCswMDowMCJ9LDI4XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.vZy63RavbiDn-h7tl5HDPQ"}

A new event arrives on top meanwhile:

$ curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{"text": "Hot news"}'
{"id":31,"created_at":"2026-10-30T19:40:40.783628Z","text":"Hot news"}

The next page still starts right after event 28: nothing is seen twice. It links to the pages on both sides of it, and is cached under its own key:

$ curl -i "http://localhost:8000/events?limit=3&cursor=WyJhIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyODowMCswMDowMCJ9LDI4XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.vZy63RavbiDn-h7tl5HDPQ"
HTTP/1.1 200 OK
content-type: application/json
link: </events?limit=3>; rel="first", </events?cursor=WyJiIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyNzowMCswMDowMCJ9LDI3XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.OSBP_vwXXNeYFKye264EoA&limit=3>; rel="prev", </events?cursor=WyJhIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyNTowMCswMDowMCJ9LDI1XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.GTbQ757EXZTDM_Hn0uzyHQ&limit=3>; rel="next"
etag: W/"ea02e62400d0fa258d315d4694476627"
cache-status: Alcazaba; fwd=miss; stored
cache-control: max-age=60

{"items":[{"id":27,"created_at":"2026-10-29T09:27:00Z","text":"Event 27"},{"id":26,"created_at":"2026-10-29T09:26:00Z","text":"Event 26"},{"id":25,"created_at":"2026-10-29T09:25:00Z","text":"Event 25"}],"limit":3,"prev_cursor":"WyJiIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyNzowMCswMDowMCJ9LDI3XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.OSBP_vwXXNeYFKye264EoA","next_cursor":"WyJhIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyNTowMCswMDowMCJ9LDI1XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.GTbQ757EXZTDM_Hn0uzyHQ"}

Back with prev_cursor: the events before 27, newest first, and this time a prev_cursor of its own, since event 31 came in above them:

$ curl "http://localhost:8000/events?limit=3&cursor=WyJiIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyNzowMCswMDowMCJ9LDI3XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.OSBP_vwXXNeYFKye264EoA"
{"items":[{"id":30,"created_at":"2026-10-29T09:30:00Z","text":"Event 30"},{"id":29,"created_at":"2026-10-29T09:29:00Z","text":"Event 29"},{"id":28,"created_at":"2026-10-29T09:28:00Z","text":"Event 28"}],"limit":3,"prev_cursor":"WyJiIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOTozMDowMCswMDowMCJ9LDMwXSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.22i-EGcTS9k24Y_eMsf1-g","next_cursor":"WyJhIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyODowMCswMDowMCJ9LDI4XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.vZy63RavbiDn-h7tl5HDPQ"}

A cursor changed by hand doesn't match its signature:

$ curl -i "http://localhost:8000/events?limit=3&cursor=WyJhIixbeyJ0IjoiZGF0ZXRpbWUiLCJ2IjoiMjAyNi0xMC0yOVQwOToyODowMCswMDowMCJ9LDI4XSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.vZy63RavbiDn-h7tl5HDPQx"
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"Invalid cursor"}

What to look at

• A unique sort key: (created_at, id), since two events may share a time.
• Both ways: next_cursor holds the key of the page's last event, prev_cursor that of its first. For the page before, the database reads the index the other way (oldest first, past the key), and paginate puts the rows back newest first.
• One row more: the database fetches params.fetch rows (the limit and one), which tells whether there's another page that way, and paginate(..., queried=True) keeps the page.
• Readable, not changeable: the cursor is base64 JSON, so clients can read the sort key, and signed, so they can't change it nor use it with other query parameters.
• Pages that don't move: a cursor's page is the same whatever is added later, so each one is cached under its own key; the first page is renewed every minute.

The code

examples/pagination/cursor.py

"""An activity feed by cursor, newest first: each `next_cursor` gives the events right after the
last one seen, however many were added since, and `prev_cursor` the ones before the first.

    uvicorn examples.pagination.cursor:app

`PaginationCursor` hides the sort key of the event at each edge of a page, `(created_at, id)`, in
a signed cursor: clients can't make one up, change it or use it with other filters. The database
fetches one row more than the page, which tells whether there's another one that way. Set
CURSOR_SECRET (secrets.token_urlsafe(32)), the same in every process, in the environment or a
.env file.
"""

import secrets
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from typing import Annotated

import uvicorn
from fastapi import Depends, FastAPI, status
from pydantic import BaseModel, Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

from fastapi_alcazaba import FastAPICache
from fastapi_alcazaba.backends.inmemory import InMemoryBackend
from fastapi_alcazaba.pagination.cursor import CursorPage, CursorParams, PaginationCursor
from fastapi_alcazaba.response_cache import cache

START = datetime(2026, 10, 29, 9, 0, tzinfo=UTC)


def throwaway_secret() -> SecretStr:
    """A random secret, for trying the example: it changes at every start and differs between
    processes."""
    return SecretStr(secrets.token_urlsafe(32))


class Settings(BaseSettings):
    """The configuration, from environment variables or a .env file."""

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")

    cursor_secret: SecretStr = Field(default_factory=throwaway_secret)


settings = Settings()


class Event(BaseModel):
    id: int
    created_at: datetime
    text: str


class NewEvent(BaseModel):
    text: str = Field(min_length=1, max_length=200)


def sort_key(event: Event) -> tuple[datetime, int]:
    """Unique: two events can share a time, never an id."""
    return event.created_at, event.id


class Database:
    """Stands in for the database: 30 events, a minute apart."""

    def __init__(self) -> None:
        self.events = [
            Event(id=n, created_at=START + timedelta(minutes=n), text=f"Event {n}")
            for n in range(1, 31)
        ]

    def page(self, params: CursorParams) -> list[Event]:
        """Newest first: SELECT * FROM events WHERE (created_at, id) < :after
        ORDER BY created_at DESC, id DESC LIMIT :fetch. For the page before:
        WHERE (created_at, id) > :before ORDER BY created_at, id LIMIT :fetch, nearest first."""
        if isinstance(params.before, tuple):  # the key of the first event of the page after
            found = sorted(self.events, key=sort_key)
            found = [event for event in found if sort_key(event) > params.before]
        else:
            found = sorted(self.events, key=sort_key, reverse=True)
            if isinstance(params.after, tuple):  # the key of the last event of the page before
                found = [event for event in found if sort_key(event) < params.after]
        return found[: params.fetch]

    def add(self, event: NewEvent) -> Event:
        saved = Event(id=len(self.events) + 1, created_at=datetime.now(UTC), text=event.text)
        self.events.append(saved)
        return saved


db = Database()
# Signed: clients can't make cursors up. Every process must have the same secret, or a cursor
# made by one is refused by the others.
cursors = PaginationCursor(settings.cursor_secret, default_limit=5, max_limit=20)


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    # Redis or Valkey in production, so that every process shares the pages.
    FastAPICache.init(InMemoryBackend(), prefix="feed", vary_headers=())  # public data
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/events")
@cache(expire=60)
async def events(params: Annotated[CursorParams, Depends(cursors)]) -> CursorPage[Event]:
    """One entry per cursor: the first page is renewed every minute, the others never change."""
    return cursors.paginate(db.page(params), params, key=sort_key, descending=True, queried=True)


@app.post("/events", status_code=status.HTTP_201_CREATED)
async def add_event(event: NewEvent) -> Event:
    """A new event goes on top: the next pages of every client stay as they were."""
    return db.add(event)


if __name__ == "__main__":
    uvicorn.run("cursor:app", reload=True)