Pagination

What it's for: lists served a page at a time, by page number or by cursor, with limits that clients can't go past, and each page cached apart.

Each pagination strategy has an example of its own, in the menu. This one is a product catalogue paginated with PaginationPageNumber and PaginationCursor, and cached three ways: one entry per page, one entry for a whole short list sliced into pages, and one entry per cursor. Every list is tagged, so adding a product drops every page at once.

Run it

From a clone of the repository:

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

Cursors are signed: set CURSOR_SECRET (secrets.token_urlsafe(32)), the same in every process, or a cursor made by one process is refused by the others.

Try it

A page, computed once and then served from the cache, which links to the pages around it in the Link header:

$ curl -i "http://127.0.0.1:8000/products?page=2&page_size=3"
HTTP/1.1 200 OK
link: </products?page=1&page_size=3>; rel="first", </products?page=1&page_size=3>; rel="prev", </products?page=3&page_size=3>; rel="next", </products?page=20&page_size=3>; rel="last"
cache-status: Alcazaba; fwd=miss; stored
cache-control: max-age=300

{"items":[{"name":"Product 4","category":"vegetables","price":1.4,"id":4},{"name":"Product 5","category":"oil","price":1.5,"id":5},{"name":"Product 6","category":"fruit","price":1.6,"id":6}],"total":60,"page":2,"page_size":3,"pages":20}

$ curl -si "http://127.0.0.1:8000/products?page=2&page_size=3" | grep cache-status
cache-status: Alcazaba; hit; ttl=300

Pages have a maximum size, checked before anything is queried:

$ curl "http://127.0.0.1:8000/products?page_size=1000"
{"detail":[{"type":"less_than_equal","loc":["query","page_size"],"msg":"Input should be less than or equal to 100","input":"1000","ctx":{"le":100}}]}

By cursor, newest first: each next_cursor gives the products right after the last one seen, however many are added meanwhile, and prev_cursor the ones before the first. Cursors are signed, so a made-up one is refused:

$ curl "http://127.0.0.1:8000/products/newest?limit=2"
{"items":[{"name":"Product 60",...,"id":60},{"name":"Product 59",...,"id":59}],"limit":2,"prev_cursor":null,"next_cursor":"WyJhIiw1OSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.S2JSMl4UWMumn4AjE9_ghQ"}

$ curl "http://127.0.0.1:8000/products/newest?limit=2&cursor=WyJhIiw1OSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.S2JSMl4UWMumn4AjE9_ghQ"
{"items":[{"name":"Product 58",...,"id":58},{"name":"Product 57",...,"id":57}],"limit":2,"prev_cursor":"WyJiIiw1OCwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.hERhWzZZDoKBLL99aQKcgg","next_cursor":"WyJhIiw1NywiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.A01Jk7K5khFEXQC2OO9uLw"}

$ curl -si "http://127.0.0.1:8000/products/newest?limit=2" | grep link
link: </products/newest?limit=2>; rel="first", </products/newest?cursor=WyJhIiw1OSwiNDdERVFwajhIQlNhLV9USW1XLTVKQSIsMF0.S2JSMl4UWMumn4AjE9_ghQ&limit=2>; rel="next"

$ curl -i "http://127.0.0.1:8000/products/newest?cursor=abc"
HTTP/1.1 400 Bad Request

{"detail":"Invalid cursor"}

Adding a product drops every cached page:

$ curl -X POST http://127.0.0.1:8000/products -H "Content-Type: application/json" \
    -d '{"name": "Capers", "category": "vegetables", "price": 2.5}'
{"name":"Capers","category":"vegetables","price":2.5,"id":61}

$ curl -si "http://127.0.0.1:8000/products?page=2&page_size=3" | grep cache-status
cache-status: Alcazaba; fwd=miss; stored

What to look at

• The page parameters are arguments of the endpoint, so @cache keeps each page, size and category under its own key. The catalogue (the database session in a real app) is left out of the key with exclude_params, since it changes from one request to the next.
• Categories are an enum, not any string: clients can't create cache entries without end.
• Three strategies for three kinds of lists: a page per entry for big lists where most clients read the first pages; the whole list once (get_or_set) for short ones; a cursor for infinite scroll and clients that sync everything.
• Tags: every list is tagged products, and a write drops them all without knowing their keys.
• Public data: the cache is created with vary_headers=(), so pages are shared by everybody. Don't do that for anything that depends on who asks.
• Links made for each request: a cached page gets the Link header of the request it answers, with that request's other query parameters; the links aren't stored with the entry, where one client's parameters would reach everybody else.

The code

examples/pagination/main.py

"""A product catalogue paginated and cached by Alcazaba, with no other library.

    uvicorn examples.pagination.main:app

Three ways to cache paginated lists:

- `/products`: `@cache` on the endpoint, one entry per page. Each response is served straight from
  the cache. Best for big lists, where most clients only read the first pages.
- `/categories/{category}/products`: the whole list cached once with `get_or_set`, and every page
  sliced from it. One entry, whatever the page and page size. Best for short lists.
- `/products/newest`: by cursor, newest first, one entry per cursor. Products added meanwhile
  don't shift the pages, which suits infinite scroll and clients that sync everything.

They're tagged "products", so a write drops every page at once without knowing their keys.
"""

import secrets
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from enum import StrEnum
from typing import Annotated

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

from fastapi_alcazaba import (
    Cache,
    FastAPICache,
    get_cache,
)
from fastapi_alcazaba.backends.inmemory import InMemoryBackend
from fastapi_alcazaba.pagination.cursor import CursorPage, CursorParams, PaginationCursor
from fastapi_alcazaba.pagination.page_number import (
    PageNumberPage,
    PageNumberParams,
    PaginationPageNumber,
)
from fastapi_alcazaba.response_cache import cache


class Category(StrEnum):
    # An enum, not any string: clients can't create cache entries without end.
    FRUIT = "fruit"
    VEGETABLES = "vegetables"
    OIL = "oil"


class NewProduct(BaseModel):
    name: str
    category: Category
    price: float


class Product(NewProduct):
    id: int


class Catalog:
    """Stands in for the database (a session per request in a real app)."""

    def __init__(self) -> None:
        categories = list(Category)
        self.products = [
            Product(id=i, name=f"Product {i}", category=categories[i % 3], price=1 + i / 10)
            for i in range(1, 61)
        ]
        self.queries = 0

    def find(self, category: Category | None = None) -> list[Product]:
        self.queries += 1
        return [p for p in self.products if category is None or p.category == category]

    def page(
        self,
        params: PageNumberParams,
        category: Category | None = None,
    ) -> PageNumberPage[Product]:
        """What a database does: count, and fetch only the page (OFFSET/LIMIT)."""
        found = self.find(category)
        rows = found[params.offset : params.offset + params.limit]
        return pages.paginate(rows, params, total=len(found))

    def newest(self, params: CursorParams) -> CursorPage[Product]:
        """What a database does: WHERE id < :after ORDER BY id DESC LIMIT :fetch; for the page
        before, WHERE id > :before ORDER BY id LIMIT :fetch, nearest first."""
        self.queries += 1
        if isinstance(params.before, int):  # the id of the first product of the page after
            rows = sorted((p for p in self.products if p.id > params.before), key=lambda p: p.id)
        else:
            rows = sorted(self.products, key=lambda p: p.id, reverse=True)
            if isinstance(params.after, int):  # the id of the last product of the page before
                rows = [p for p in rows if p.id < params.after]
        page = rows[: params.fetch]
        return cursors.paginate(page, params, key=lambda p: p.id, descending=True, queried=True)

    def add(self, product: NewProduct) -> Product:
        saved = Product(id=len(self.products) + 1, **product.model_dump())
        self.products.append(saved)
        return saved


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


app = FastAPI(lifespan=lifespan)


def get_catalog(request: Request) -> Catalog:
    catalog: Catalog = request.app.state.catalog
    return catalog


CatalogDep = Annotated[Catalog, Depends(get_catalog)]


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()
pages = PaginationPageNumber(default_page_size=20, max_page_size=100)
# 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: set CURSOR_SECRET (secrets.token_urlsafe(32)).
cursors = PaginationCursor(settings.cursor_secret, default_limit=20)


@app.get("/products")
@cache(expire=300, tags=["products"], exclude_params=["catalog"])
async def products(
    params: Annotated[PageNumberParams, Depends(pages)],
    catalog: CatalogDep,
    category: Category | None = None,
) -> PageNumberPage[Product]:
    """One entry per page, page size and category: the page parameters are arguments, part of the
    key. The session is left out of it, since it changes from one request to the next."""
    return catalog.page(params, category)


@app.get("/categories/{category}/products")
async def category_products(
    category: Category,
    params: Annotated[PageNumberParams, Depends(pages)],
    catalog: CatalogDep,
    app_cache: Annotated[Cache, Depends(get_cache)],
) -> PageNumberPage[Product]:
    """One entry for the whole category: every page is sliced from it."""

    async def load() -> list[Product]:  # annotated: the entry comes back as products
        return catalog.find(category)

    found = await app_cache.get_or_set(f"category:{category}", load, expire=300, tags=["products"])
    return pages.paginate(found, params)


@app.get("/products/newest")
@cache(expire=300, tags=["products"], exclude_params=["catalog"])
async def newest_products(
    params: Annotated[CursorParams, Depends(cursors)],
    catalog: CatalogDep,
) -> CursorPage[Product]:
    """By cursor, newest first: each next_cursor gives the products right after the last one seen,
    however many were added since."""
    return catalog.newest(params)


@app.post("/products", status_code=status.HTTP_201_CREATED)
async def add_product(product: NewProduct, catalog: CatalogDep) -> Product:
    """The new product may land on any page: drop them all."""
    saved = catalog.add(product)
    await FastAPICache.invalidate_tags("products")
    return saved


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