Lesson 5 — validation and pagination

What you will build

The same collection endpoint, one turn of the screw: it now reads ?limit= and ?offset= from the query string. That sounds like two numbers, and it is the first place in this path where text from outside meets a rule inside, so it is where the shape of validation gets decided.

What you need

Nothing new. The http and database extras from lesson 4 are all this lesson uses.

The code

One file changes. examples/user_management/app.py grows a private helper and passes two arguments to the service:

examples/user_management/app.py

"""Compose the application and expose it as an ASGI app."""

from __future__ import annotations

from collections.abc import Mapping
from pathlib import Path
from typing import Annotated

from starlette.applications import Starlette
from starlette.datastructures import QueryParams
from starlette.requests import Request
from starlette.responses import RedirectResponse, Response
from starlette.templating import Jinja2Templates

from user_management.errors import DuplicateEmailError, UserValidationError
from user_management.identity import PrincipalVerifier, require_role
from user_management.service import DEFAULT_LIMIT, UserService
from xyberos import App, Inject, Settings
from xyberos.database import DatabaseProvider, SqliteSettings, sqlite_provider
from xyberos.http import HttpAdapter, read_json_object

#: The role the administrative routes require. One constant, so the route table reads as a list.
ADMIN = "admin"

#: Where the pages live. Templates are application code: the framework routes and serializes, and a
#: template engine is something an application adds.
TEMPLATES = Path(__file__).parent / "templates"

#: The header HTMX sends. It is how one route answers a browser and a script differently.
HX_REQUEST = "HX-Request"


def create_app(
    *,
    settings: SqliteSettings | None = None,
    database: DatabaseProvider | None = None,
    secret: str | None = None,
) -> App:
    """Compose the application.

    Pass ``settings`` to let the application own and start its database, or ``database`` to supply
    an already-started provider, which the application then owns and closes on shutdown. ``secret``
    is the identity provider's shared secret; without one the verifier refuses every request.
    """
    app = App(settings=Settings.load(environ={}, overrides={"app": {"name": "user_management"}}))
    if database is None:
        app.provide(DatabaseProvider, factory=sqlite_provider(settings), eager=True)
    else:
        app.provide(DatabaseProvider, instance=database)
    app.provide(UserService)
    app.provide(PrincipalVerifier, instance=PrincipalVerifier(secret))

    def administrator(request: Request, verifier: PrincipalVerifier) -> None:
        """Refuse a caller who is not an administrator, before anything else happens.

        One call at the top of a route, and its cost is that it can be forgotten: there is no marker
        on the handler that a reviewer can grep for except this line. That is the trade the ADR
        describes — the application owns authorization, so the route table is where it is audited.
        """
        require_role(verifier.verify(request), ADMIN)

    @app.get("/health")
    async def health(database: Annotated[DatabaseProvider, Inject()]) -> dict[str, str]:
        await database.health()
        return {"status": "ready"}

    @app.post("/users")
    async def create_user(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        administrator(request, verifier)
        payload = await read_json_object(request)
        return await users.create_user(payload.get("email"), payload.get("display_name"))

    @app.get("/users")
    async def list_users(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        administrator(request, verifier)
        query = request.query_params
        return await users.list_users(
            limit=_whole_number(query, "limit", default=DEFAULT_LIMIT),
            offset=_whole_number(query, "offset", default=0),
        )

    @app.get("/users/{user_id}")
    async def get_user(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        administrator(request, verifier)
        return await users.get_user(user_id)

    @app.patch("/users/{user_id}")
    async def update_user(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        administrator(request, verifier)
        payload = await read_json_object(request)
        return await users.update_user(user_id, payload)

    @app.delete("/users/{user_id}")
    async def delete_user(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> None:
        administrator(request, verifier)
        await users.delete_user(user_id)

    @app.get("/me")
    async def read_me(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        principal = verifier.verify(request)
        return await users.find_own_user(principal)

    @app.post("/me")
    async def claim_account(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        principal = verifier.verify(request)
        payload = await read_json_object(request)
        return await users.create_user(
            payload.get("email"),
            payload.get("display_name"),
            external_subject=principal.subject,
        )

    @app.patch("/me")
    async def update_me(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        principal = verifier.verify(request)
        payload = await read_json_object(request)
        return await users.update_own_profile(principal, payload)

    templates = Jinja2Templates(directory=str(TEMPLATES))

    async def users_page(
        request: Request, users: UserService, *, status_code: int = 200, error: str | None = None
    ) -> Response:
        page = await users.list_users()
        return templates.TemplateResponse(
            request,
            "users.html",
            {"users": page["items"], "error": error, "form": await _submitted_form(request)},
            status_code=status_code,
        )

    @app.get("/pages/users")
    async def show_users(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> Response:
        administrator(request, verifier)
        return await users_page(request, users)

    @app.post("/pages/users")
    async def create_user_from_form(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> Response:
        administrator(request, verifier)
        form = await request.form()
        try:
            await users.create_user(form.get("email"), form.get("display_name"))
        except (DuplicateEmailError, UserValidationError) as error:
            return await users_page(request, users, status_code=422, error=str(error))
        return RedirectResponse("/pages/users", status_code=303)

    @app.post("/pages/users/{user_id}/toggle")
    async def toggle_user(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> Response:
        administrator(request, verifier)
        current = await users.get_user(user_id)
        updated = await users.update_user(user_id, {"enabled": not current["enabled"]})
        if request.headers.get(HX_REQUEST):
            return templates.TemplateResponse(request, "_row.html", {"user": updated})
        return RedirectResponse("/pages/users", status_code=303)

    return app


async def _submitted_form(request: Request) -> Mapping[str, object] | None:
    """Return what the client typed, so a refused form can be shown back to them."""
    if request.method != "POST":
        return None
    form = await request.form()
    return {key: str(value) for key, value in form.items()}


def _whole_number(query: QueryParams, name: str, *, default: int) -> int:
    """Read one query string value as a whole number, or refuse the request.

    A query string is text: `?limit=abc` arrives as `"abc"`. The conversion belongs here, where the
    request is untrusted, and the *bounds* stay in the service, which knows what a page may be.
    """
    raw = query.get(name)
    if raw is None:
        return default
    try:
        return int(raw)
    except ValueError:
        raise UserValidationError(f"{name} must be a whole number") from None


def build_asgi_app(app: App) -> Starlette:
    """Build an ASGI application for a composed application."""
    return HttpAdapter(app).asgi_app


def app_factory() -> Starlette:
    """Create the application's ASGI app for an ASGI server.

    The identity secret is read from the environment by the verifier, and a deployment that has not
    set it refuses every request rather than trusting one.

    Usage: ``uvicorn --app-dir examples --factory user_management.app:app_factory``
    """
    return build_asgi_app(create_app())

The handler still does not know what a valid page is: it turns two strings into two integers, and the service refuses the integers that make no sense. examples/user_management/service.py did not change — its bounds were in place from lesson 4, because they are rules about the operation rather than about a request.

And the lesson's own module drives it:

examples/learn_traditional/traditional05_pagination.py

"""Lesson 5 — validating query parameters, and paging the collection.

Run it with:

    python -m learn_traditional.traditional05_pagination
"""

from __future__ import annotations

import tempfile
from pathlib import Path

from user_management.app import create_app
from user_management.identity import PrincipalVerifier
from user_management.migrate import migrate

from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client

SECRET = "a-secret-the-deployment-keeps"
#: The administrator's assertion, presented once and carried by every request.
ADMIN_HEADERS = PrincipalVerifier(SECRET).headers("admin@idp", "admin")

EMAILS = [
    "ada@example.com",
    "grace@example.com",
    "katherine@example.com",
    "marie@example.com",
    "radia@example.com",
]

PAGES = ["?limit=2", "?limit=2&offset=2", "?limit=2&offset=4"]
REFUSALS = ["?limit=abc", "?limit=0", "?limit=101", "?offset=-1"]


def main() -> None:
    with tempfile.TemporaryDirectory() as directory:
        settings = SqliteSettings(path=Path(directory) / "users.db")
        migrate(settings)

        app = create_app(settings=settings, secret=SECRET)
        with create_test_client(HttpAdapter(app)) as client:
            client.headers.update(ADMIN_HEADERS)
            for email in EMAILS:
                client.post(
                    "/users", json={"email": email, "display_name": email.split("@")[0].title()}
                )

            for query in PAGES:
                response = client.get(f"/users{query}")
                page = response.json()
                emails = [item["email"] for item in page["items"]]
                print(f"GET /users{query} ->", response.status_code, emails, "of", page["total"])

            for query in REFUSALS:
                response = client.get(f"/users{query}")
                print(f"GET /users{query} ->", response.status_code, response.json()["error"])


if __name__ == "__main__":
    main()

How to run it

PowerShellPOSIX

$env:PYTHONPATH = "src;examples"
python -m learn_traditional.traditional05_pagination

PYTHONPATH=src:examples python -m learn_traditional.traditional05_pagination

What it prints

GET /users?limit=2 -> 200 ['ada@example.com', 'grace@example.com'] of 5
GET /users?limit=2&offset=2 -> 200 ['katherine@example.com', 'marie@example.com'] of 5
GET /users?limit=2&offset=4 -> 200 ['radia@example.com'] of 5
GET /users?limit=abc -> 422 {'type': 'UserValidationError', 'message': 'limit must be a whole number'}
GET /users?limit=0 -> 422 {'type': 'UserValidationError', 'message': 'limit must be between 1 and 100'}
GET /users?limit=101 -> 422 {'type': 'UserValidationError', 'message': 'limit must be between 1 and 100'}
GET /users?offset=-1 -> 422 {'type': 'UserValidationError', 'message': 'offset must not be negative'}

Three pages of five users, then four refusals. Look at why each refusal happened:

Request	Refused by	Because
?limit=abc	the handler, in app.py	"abc" is not a number, and only the handler knows it was text
?limit=0	the service, in service.py	zero is a number, and no page may be empty
?limit=101	the service	a page larger than 100 is a client asking for the table
?offset=-1	the service	an offset before the first row has no meaning

That split is the lesson. int() is not validation — it is parsing, and it belongs where the untrusted text arrives. The bounds are rules about the operation, so they belong where the operation lives, and that is what keeps them true when the caller is a CLI or a background job instead of a request.

Both kinds of refusal come back as the same 422 body, because UserValidationError is a RequestValidationError whichever layer raised it. A client does not have to care which layer said no.

Why the last page is not an error

?offset=4 asks for rows five onwards, and one row is left. That is a correct page, not a problem: the client asked for up to two rows starting at four, and there were fewer than two. total is how it learns that, because the only alternative is guessing from a short page — and a full last page is indistinguishable from a page with more behind it.

Three answers are worth having for a collection, and this one is the middle:

• Return everything. Fine until it is not, and then a request times out that used to work.
• Return a page with its total, and let the client ask for more. What this endpoint does. The client can page forwards, jump to an offset, and show "1–2 of 5" without a second request.
• Return a cursor. The right answer for a table that changes under the reader, because an offset drifts when rows are inserted. It is also more machinery than a first release needs; the pages here are small and ordered by id.

Common mistakes

• Reading limit with request.query_params.get("limit", 20). It returns the string "20" when the parameter is present and an integer when it is not, so the type depends on the client. Default in the handler, convert in the handler, and the service gets an integer either way.
• Validating a query parameter by catching everything. except Exception around the conversion turns a typo in the handler into a 422 blaming the client. int() raises ValueError; catch that.
• Believing a 422 is enough. Range checks on limit protect the database, and the offset still needs a bound if the table is large: ?offset=1000000000 is a legitimate-looking request that makes SQLite walk a long way. The bigger the table, the more it matters.
• Making the page size a client-only decision. A limit of ten thousand is a denial-of-service request with a friendly name, which is why the maximum lives in the service and not in documentation.
• Forgetting that the offset is a position, not a page number. ?limit=2&offset=3 is a page that skips a row. Either the client computes offset = page * limit, or the API takes a page number and converts — but the two must not be mixed.

Exercise

Add an exact-match ?email= filter to the collection: absent means no filter, a blank value is refused, and total counts the rows the filter matched rather than the whole table. The match must be normalised the same way create_user normalises what it stores, or ?email=ADA@example.com finds nothing.

Show the answer

examples/learn_traditional/traditional05_exercise.py

"""Lesson 5, exercise — an exact-match email filter that pages like any other collection.

The answer to the lesson's exercise. In your project this is one more argument on `list_users` and
one more branch in the same handler. It is a separate route here, on `/directory` rather than under
`/users/`, so the answer stays runnable while the project keeps changing underneath it.
"""

from __future__ import annotations

import tempfile
from pathlib import Path
from typing import Annotated

from httpx import Response
from sqlalchemy import func, select
from starlette.requests import Request
from user_management.app import create_app
from user_management.errors import UserValidationError
from user_management.identity import PrincipalVerifier
from user_management.migrate import migrate
from user_management.service import DEFAULT_LIMIT, UserService
from user_management.tables import users

from xyberos import App, Inject
from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client

SECRET = "a-secret-the-deployment-keeps"
#: The administrator's assertion, presented once and carried by every request.
ADMIN_HEADERS = PrincipalVerifier(SECRET).headers("admin@idp", "admin")

EMAILS = ["ada@example.com", "grace@example.com", "katherine@example.com"]

FILTERS = ["", "?email=ADA@example.com", "?email=marie@example.com", "?email="]


class UserDirectory(UserService):
    """``service.py``, plus the one query the exercise asks for."""

    async def find_users(
        self, email: object, *, limit: int = DEFAULT_LIMIT, offset: int = 0
    ) -> dict[str, object]:
        """Return one page of users with exactly this address, and how many there are."""
        wanted = _required_email(email)
        page = select(users).where(users.c.email == wanted).order_by(users.c.id)
        count = select(func.count()).select_from(users).where(users.c.email == wanted)
        async with self._database.transaction() as connection:
            total = int((await connection.execute(count)).scalar_one())
            rows = (await connection.execute(page.limit(limit).offset(offset))).mappings().all()
        return {
            "items": [dict(row) for row in rows],
            "total": total,
            "limit": limit,
            "offset": offset,
        }


def _required_email(email: object) -> str:
    """Normalise the filter the way `create_user` normalises the value it stores."""
    if not isinstance(email, str) or not email.strip():
        raise UserValidationError("email must not be blank")
    return email.strip().lower()


def create_directory_app(*, settings: SqliteSettings) -> App:
    """The lesson's application, with the exercise's route added to it."""
    app = create_app(settings=settings, secret=SECRET)
    app.provide(UserDirectory)

    @app.get("/directory")
    async def find_users(
        request: Request,
        users: Annotated[UserService, Inject()],
        directory: Annotated[UserDirectory, Inject()],
    ) -> dict[str, object]:
        email = request.query_params.get("email")
        if email is None:
            return await users.list_users()
        return await directory.find_users(email)

    return app


def report(query: str, response: Response) -> None:
    """Print one response: the addresses it matched, or the refusal it returned."""
    body = response.json()
    if response.status_code == 200:
        addresses = [item["email"] for item in body["items"]]
        print(f"GET /directory{query} ->", response.status_code, addresses, "of", body["total"])
    else:
        print(f"GET /directory{query} ->", response.status_code, body["error"]["message"])


def main() -> None:
    """Create three users, then filter the collection four ways."""
    with tempfile.TemporaryDirectory() as directory:
        settings = SqliteSettings(path=Path(directory) / "users.db")
        migrate(settings)

        app = create_directory_app(settings=settings)
        with create_test_client(HttpAdapter(app)) as client:
            client.headers.update(ADMIN_HEADERS)
            for email in EMAILS:
                client.post(
                    "/users", json={"email": email, "display_name": email.split("@")[0].title()}
                )

            for query in FILTERS:
                report(query, client.get(f"/directory{query}"))


if __name__ == "__main__":
    main()

GET /directory -> 200 ['ada@example.com', 'grace@example.com', 'katherine@example.com'] of 3
GET /directory?email=ADA@example.com -> 200 ['ada@example.com'] of 1
GET /directory?email=marie@example.com -> 200 [] of 0
GET /directory?email= -> 422 email must not be blank

The answer's route is `/directory` rather than something under `/users/`, because a literal segment there is matched by the per-user route a later lesson adds: `/users/find` would be read as a user id and refused as one. Route order is decided by registration, and the application registers its own routes before an answer can add to them. Two details carry the lesson forward. The **count is filtered too** — `total` describes the collection the client is paging through, so a filtered page that reported `3` would make the client keep asking for pages that cannot exist. And the **filter is normalised at the edge of the service**, next to the rule it mirrors: the column stores `ada@example.com`, so a filter that compared `ADA@example.com` literally would return an empty page and look like a missing row rather than a missing `lower()`.

Next

Lesson 6 finishes CRUD: reading one user, editing one, deleting one — and the difference between a write that repeats safely and a write that must not.