PaginationPageNumber¶
A blog's archive by page number: numbered pages of posts, newest first, with the total and the number of pages, each page cached under its own key. The strategy: PaginationPageNumber.
Run it¶
From a clone of the repository:
make deps # .venv, with every extra
source .venv/bin/activate
uvicorn examples.pagination.page_number:app
Try it¶
The second page of three posts: the response says how many there are, and links to the pages around it:
$ curl -i "http://localhost:8000/posts?page=2&page_size=3"
HTTP/1.1 200 OK
content-type: application/json
link: </posts?page=1&page_size=3>; rel="first", </posts?page=1&page_size=3>; rel="prev", </posts?page=3&page_size=3>; rel="next", </posts?page=15&page_size=3>; rel="last"
etag: W/"f33562cc4b3edbd0316e1399d0de90a0"
cache-status: Alcazaba; fwd=miss; stored
cache-control: max-age=300
{"items":[{"id":42,"title":"Post 42"},{"id":41,"title":"Post 41"},{"id":40,"title":"Post 40"}],"total":45,"page":2,"page_size":3,"pages":15}
Asked again, it comes from the cache, without a query:
$ curl -si "http://localhost:8000/posts?page=2&page_size=3" | grep cache-status
cache-status: Alcazaba; hit; ttl=300
More than 20 posts a page is refused before the endpoint runs:
$ curl -i "http://localhost:8000/posts?page_size=21"
HTTP/1.1 422 Unprocessable Content
Publishing a post (POST /posts) shifts every page by one: it drops them all, by the posts
tag.
What to look at¶
- Only the page, and the count: the database fetches the page with
params.offsetandparams.limit, and counts the posts:pages.paginate(rows, params, total=count). - One entry per page and size: the page parameters are an argument of the endpoint, so
@cachekeeps each page apart. - The cost of numbers: each page counts every post, and a post published while someone reads shifts the pages after it. For feeds, see PaginationCursor.
The code¶
examples/pagination/page_number.py
"""A blog's archive by page number: numbered pages of posts, newest first, with the total and the
number of pages.
uvicorn examples.pagination.page_number:app
`PaginationPageNumber` reads `page` and `page_size` from the query (a 422 for a page size over
20), the database counts the posts and fetches only the page, and the response links to the
first, previous, next and last pages. Each page is cached under its own key; publishing a post
drops them all, by tag.
"""
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated
import uvicorn
from fastapi import Depends, FastAPI, status
from pydantic import BaseModel, Field
from fastapi_alcazaba import FastAPICache
from fastapi_alcazaba.backends.inmemory import InMemoryBackend
from fastapi_alcazaba.pagination.page_number import (
PageNumberPage,
PageNumberParams,
PaginationPageNumber,
)
from fastapi_alcazaba.response_cache import cache
class Post(BaseModel):
id: int
title: str
class NewPost(BaseModel):
title: str = Field(min_length=1, max_length=200)
class Database:
"""Stands in for the database: 45 posts, and how many queries it ran."""
def __init__(self) -> None:
self.posts = [Post(id=n, title=f"Post {n}") for n in range(1, 46)]
self.queries = 0
def count(self) -> int:
"""SELECT count(*) FROM posts"""
self.queries += 1
return len(self.posts)
def newest(self, offset: int, limit: int) -> list[Post]:
"""SELECT * FROM posts ORDER BY id DESC LIMIT :limit OFFSET :offset"""
self.queries += 1
return sorted(self.posts, key=lambda post: post.id, reverse=True)[offset : offset + limit]
def add(self, post: NewPost) -> Post:
saved = Post(id=len(self.posts) + 1, title=post.title)
self.posts.append(saved)
return saved
db = Database()
pages = PaginationPageNumber(default_page_size=10, max_page_size=20)
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
# Redis or Valkey in production, so that every process shares the pages.
FastAPICache.init(InMemoryBackend(), prefix="blog", vary_headers=()) # public data
yield
app = FastAPI(lifespan=lifespan)
@app.get("/posts")
@cache(expire=300, tags=["posts"])
async def posts(params: Annotated[PageNumberParams, Depends(pages)]) -> PageNumberPage[Post]:
"""Only the page, and the count: what a database does with OFFSET and LIMIT."""
rows = db.newest(params.offset, params.limit)
return pages.paginate(rows, params, total=db.count())
@app.post("/posts", status_code=status.HTTP_201_CREATED)
async def publish(post: NewPost) -> Post:
"""The new post goes first and shifts every page: drop them all."""
saved = db.add(post)
await FastAPICache.invalidate_tags("posts")
return saved
if __name__ == "__main__":
uvicorn.run("page_number:app", reload=True)