PaginationOffset¶
An access log of a thousand lines, read newest first by offset and limit, without ever being counted: each page says whether there's another after it. The strategy: PaginationOffset.
Run it¶
From a clone of the repository:
make deps # .venv, with every extra
source .venv/bin/activate
uvicorn examples.pagination.offset:app
Try it¶
The three newest lines, and whether there are more:
$ curl -i "http://localhost:8000/logs?limit=3"
HTTP/1.1 200 OK
content-type: application/json
link: </logs?offset=0&limit=3>; rel="first", </logs?offset=3&limit=3>; rel="next"
etag: W/"64cc8e327a8a5779379fa48acb2c8286"
cache-status: Alcazaba; fwd=miss; stored
cache-control: max-age=10
{"items":[{"id":1000,"level":"error","message":"GET /products/30"},{"id":999,"level":"info","message":"GET /products/29"},{"id":998,"level":"info","message":"GET /products/28"}],"offset":0,"limit":3,"has_more":true,"total":null}
The next link is the following page:
$ curl -i "http://localhost:8000/logs?offset=3&limit=3"
HTTP/1.1 200 OK
content-type: application/json
link: </logs?offset=0&limit=3>; rel="first", </logs?offset=0&limit=3>; rel="prev", </logs?offset=6&limit=3>; rel="next"
etag: W/"7f266a69ae7b4ed942e5ef096485b258"
cache-status: Alcazaba; fwd=miss; stored
cache-control: max-age=10
{"items":[{"id":997,"level":"info","message":"GET /products/27"},{"id":996,"level":"info","message":"GET /products/26"},{"id":995,"level":"info","message":"GET /products/25"}],"offset":3,"limit":3,"has_more":true,"total":null}
The errors alone: twenty of them, so from the nineteenth on there are only two, has_more is
false and the links have no next. Asked again within ten seconds, the page comes from the cache:
$ curl -i "http://localhost:8000/logs?level=error&offset=18"
HTTP/1.1 200 OK
content-type: application/json
link: </logs?level=error&offset=0&limit=20>; rel="first", </logs?level=error&offset=0&limit=20>; rel="prev"
etag: W/"b7de8434b1ce71cd4dffef81cfa5b626"
cache-status: Alcazaba; fwd=miss; stored
cache-control: max-age=10
{"items":[{"id":100,"level":"error","message":"GET /products/3"},{"id":50,"level":"error","message":"GET /products/50"}],"offset":18,"limit":20,"has_more":false,"total":null}
$ curl -si "http://localhost:8000/logs?level=error&offset=18" | grep cache-status
cache-status: Alcazaba; hit; ttl=10
Pages hold at most 100 lines, checked before anything is queried:
$ curl -i "http://localhost:8000/logs?limit=500"
HTTP/1.1 422 Unprocessable Content
content-type: application/json
{"detail":[{"type":"less_than_equal","loc":["query","limit"],"msg":"Input should be less than or equal to 100","input":"500","ctx":{"le":100}}]}
What to look at¶
- One row more, no count: the database is asked for
limit + 1rows (params.fetch); the extra one isn't returned, it only says that there's a next page. Nothing counts the whole log, which on a big table takes longer than the page itself. - No total, no last page:
totalis null and the links are first, previous and, whilehas_more, next. An endpoint that counts passestotal=instead ofqueried=True, and its pages have the total and a link to the last one. - Lines written meanwhile shift the pages: a line logged while page one is read pushes its last line into page two, which repeats it. To follow a list that grows at the top without that, page by key: PaginationKeyset or PaginationCursor.
- A short expiry: the log changes all the time, so each page is cached for ten seconds,
under its own key (
offset,limitandlevel). The level is an enum: clients can't create cache entries without end.
The code¶
examples/pagination/offset.py
"""An access log paginated by offset, without the total: a table too big to count at every
request, and that grows while it's read.
uvicorn examples.pagination.offset:app
`/logs?offset=0&limit=20` reads one row more than the page, which says whether there's a next one
(`has_more`), instead of counting the whole log: the newest lines first, all of them or those of
one level. Each page is cached for a few seconds, under its own key.
"""
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
from pydantic import BaseModel
from fastapi_alcazaba import FastAPICache
from fastapi_alcazaba.backends.inmemory import InMemoryBackend
from fastapi_alcazaba.pagination.offset import OffsetPage, OffsetParams, PaginationOffset
from fastapi_alcazaba.response_cache import cache
class Level(StrEnum):
# An enum, not any string: clients can't create cache entries without end.
INFO = "info"
WARNING = "warning"
ERROR = "error"
class Line(BaseModel):
id: int
level: Level
message: str
def level_of(number: int) -> Level:
if number % 50 == 0:
return Level.ERROR
return Level.WARNING if number % 10 == 0 else Level.INFO
class Database:
"""Stands in for the database (a session per request in a real app)."""
def __init__(self) -> None:
self.lines = [
Line(id=i, level=level_of(i), message=f"GET /products/{i % 97}") for i in range(1, 1001)
]
self.queries = 0
def page(
self,
params: OffsetParams,
level: Level | None = None,
) -> OffsetPage[Line]:
"""What a database does: WHERE level = :level ORDER BY id DESC LIMIT :fetch OFFSET
:offset, one row more than the page and no count."""
self.queries += 1
newest = [line for line in reversed(self.lines) if level is None or line.level == level]
rows = newest[params.offset : params.offset + params.fetch]
return offsets.paginate(rows, params, queried=True)
@asynccontextmanager
async def lifespan(application: FastAPI) -> AsyncIterator[None]:
FastAPICache.init(InMemoryBackend(), prefix="logs", vary_headers=())
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)]
offsets = PaginationOffset(default_limit=20, max_limit=100)
@app.get("/logs")
@cache(expire=10, exclude_params=["db"])
async def logs(
params: Annotated[OffsetParams, Depends(offsets)],
db: DatabaseDep,
level: Level | None = None,
) -> OffsetPage[Line]:
"""The newest lines first, a page at a time: `has_more` says whether to ask for the next."""
return db.page(params, level)
if __name__ == "__main__":
uvicorn.run("offset:app", reload=True)