Lesson 4 — create and list users¶

What you will build¶

The application every remaining lesson in this path grows. It is four files, and the point of the split is that each one owns exactly one thing: the columns, the rules, the responses, and the composition that ties them together.

Lesson 3 put a service and a route in one file to keep the idea small. This lesson keeps the idea and gives it a shape you can add to for nine more lessons.

What you need¶

The two extras from lessons 1–3, together:

PowerShellPOSIX

python -m pip install "xyberos[http,database]"

python -m pip install 'xyberos[http,database]'

The project¶

File	Owns	Never contains
`tables.py`	the columns, and only the columns	SQL, or a rule
`errors.py`	domain errors, each one knowing its status code	a route
`service.py`	rules and transaction boundaries	an HTTP type
`app.py`	providers, routes, and translation	SQL
`migrate.py` + `migrations/`	putting the schema in a database, as a revision	a rule, or a route

The files below are the project as the path leaves it, so app.py already contains the query-string handling lesson 5 adds, and the schema arrives through the migrations lesson 7 explains. Later lessons read the earlier pages' code the same way.

examples/user_management/tables.py is the whole schema, read in one screen:

examples/user_management/tables.py

"""SQLAlchemy Core table definitions owned by the application, not the framework."""

from __future__ import annotations

from sqlalchemy import Boolean, Column, Integer, MetaData, String, Table, text

metadata = MetaData()

users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("email", String(320), nullable=False, unique=True),
    Column("display_name", String(200), nullable=False),
    Column("enabled", Boolean, nullable=False, server_default=text("1")),
    Column("external_subject", String(200), nullable=True, unique=True),
)

examples/user_management/errors.py is the vocabulary the rest of the application speaks. Two of these are used before the lesson ends; UserNotFoundError is waiting for lesson 6:

examples/user_management/errors.py

"""Domain errors that the HTTP adapter turns into responses."""

from __future__ import annotations

from xyberos import HttpError, RequestValidationError


class UserValidationError(RequestValidationError):
    """Raised when user input is invalid; exposed to the client as a 422."""


class DuplicateEmailError(HttpError):
    """Raised when an email is already registered; exposed to the client as a 409."""

    def __init__(self, email: str) -> None:
        super().__init__(f"{email} is already registered", status_code=409)
        self.email = email


class UserNotFoundError(HttpError):
    """Raised when a user does not exist; exposed to the client as a 404."""

    def __init__(self, user_id: int) -> None:
        super().__init__(f"user {user_id} was not found", status_code=404)
        self.user_id = user_id


class AccountNotLinkedError(HttpError):
    """Raised when a verified principal belongs to no local account; exposed as a 404."""

    def __init__(self, subject: str) -> None:
        super().__init__(f"no local account is linked to {subject}", status_code=404)
        self.subject = subject


class AccountAlreadyLinkedError(HttpError):
    """Raised when a verified principal already has a local account; exposed as a 409."""

    def __init__(self, subject: str) -> None:
        super().__init__(f"{subject} is already linked to an account", status_code=409)
        self.subject = subject

examples/user_management/service.py is where the rules live. Every method either returns a row or raises a domain error, and no method knows what a request is:

examples/user_management/service.py

"""The user service: business rules and transaction boundaries, no HTTP types."""

from __future__ import annotations

from collections.abc import Mapping
from typing import Annotated

from sqlalchemy import delete, func, insert, select, update

from user_management.errors import (
    AccountAlreadyLinkedError,
    AccountNotLinkedError,
    DuplicateEmailError,
    UserNotFoundError,
    UserValidationError,
)
from user_management.identity import Principal
from user_management.tables import users
from xyberos import Inject
from xyberos.database import DatabaseProvider

MAX_EMAIL_LENGTH = 320
MAX_DISPLAY_NAME_LENGTH = 200
DEFAULT_LIMIT = 20
MAX_LIMIT = 100
ALLOWED_FIELDS = ("email", "display_name", "enabled")
#: What a user may change about themselves. Everything else is somebody else's decision.
SELF_FIELDS = ("display_name",)


class UserService:
    """Owns user rules and keeps every write inside one short transaction."""

    def __init__(self, database: Annotated[DatabaseProvider, Inject()]) -> None:
        self._database = database

    async def create_user(
        self,
        email: object,
        display_name: object,
        *,
        external_subject: str | None = None,
    ) -> dict[str, object]:
        """Insert one user, or raise because the input is wrong or the email is taken.

        ``external_subject`` links the new account to a verified principal. Both checks happen in
        the same transaction as the insert, and the order of the three refusals is deliberate: a
        conflict about the request's target first, then the payload, then a value it names.
        """
        async with self._database.transaction() as connection:
            if external_subject is not None:
                linked = (
                    await connection.execute(
                        select(users.c.id).where(users.c.external_subject == external_subject)
                    )
                ).first()
                if linked is not None:
                    raise AccountAlreadyLinkedError(external_subject)
            normalized_email = _validated_email(email)
            normalized_name = _validated_display_name(display_name)
            taken = (
                await connection.execute(
                    select(users.c.id).where(users.c.email == normalized_email)
                )
            ).first()
            if taken is not None:
                raise DuplicateEmailError(normalized_email)
            result = await connection.execute(
                insert(users).values(
                    email=normalized_email,
                    display_name=normalized_name,
                    external_subject=external_subject,
                )
            )
            inserted = result.inserted_primary_key
            if inserted is None:
                # SQLAlchemy types this as optional, and an insert with a generated key reports one.
                raise AssertionError("the insert reported no primary key")
            user_id = int(inserted[0])
            row = (
                (await connection.execute(select(users).where(users.c.id == user_id)))
                .mappings()
                .one()
            )
        return dict(row)

    async def list_users(self, *, limit: int = DEFAULT_LIMIT, offset: int = 0) -> dict[str, object]:
        """Return one page of users, newest row last, with the size of the whole table."""
        page_limit = _validated_limit(limit)
        page_offset = _validated_offset(offset)
        statement = select(users).order_by(users.c.id).limit(page_limit).offset(page_offset)
        async with self._database.transaction() as connection:
            total = int(
                (await connection.execute(select(func.count()).select_from(users))).scalar_one()
            )
            rows = (await connection.execute(statement)).mappings().all()
        return {
            "items": [dict(row) for row in rows],
            "total": total,
            "limit": page_limit,
            "offset": page_offset,
        }

    async def get_user(self, user_id: int) -> dict[str, object]:
        """Return one user, or raise because there is no such user."""
        async with self._database.transaction() as connection:
            row = (
                (await connection.execute(select(users).where(users.c.id == user_id)))
                .mappings()
                .first()
            )
        if row is None:
            raise UserNotFoundError(user_id)
        return dict(row)

    async def update_user(self, user_id: int, changes: Mapping[str, object]) -> dict[str, object]:
        """Apply what the client sent, and nothing else, to one existing user."""
        values = _validated_changes(changes)
        async with self._database.transaction() as connection:
            exists = (
                await connection.execute(select(users.c.id).where(users.c.id == user_id))
            ).first()
            if exists is None:
                raise UserNotFoundError(user_id)
            if "email" in values:
                taken = (
                    await connection.execute(
                        select(users.c.id).where(
                            users.c.email == values["email"], users.c.id != user_id
                        )
                    )
                ).first()
                if taken is not None:
                    raise DuplicateEmailError(str(values["email"]))
            await connection.execute(update(users).where(users.c.id == user_id).values(**values))
            row = (
                (await connection.execute(select(users).where(users.c.id == user_id)))
                .mappings()
                .one()
            )
        return dict(row)

    async def delete_user(self, user_id: int) -> None:
        """Delete one user, or raise because there is no such user."""
        async with self._database.transaction() as connection:
            result = await connection.execute(delete(users).where(users.c.id == user_id))
            if result.rowcount == 0:
                raise UserNotFoundError(user_id)

    async def find_own_user(self, principal: Principal) -> dict[str, object]:
        """Return the account a verified principal belongs to, or raise because there is none.

        The subject comes from the principal and nowhere else: there is no user id in this
        signature, so a request cannot ask for somebody else's account by naming it.
        """
        async with self._database.transaction() as connection:
            row = (
                (
                    await connection.execute(
                        select(users).where(users.c.external_subject == principal.subject)
                    )
                )
                .mappings()
                .first()
            )
        if row is None:
            raise AccountNotLinkedError(principal.subject)
        return dict(row)

    async def update_own_profile(
        self, principal: Principal, changes: Mapping[str, object]
    ) -> dict[str, object]:
        """Change what a user may change about themselves, and nothing else."""
        unknown = sorted(set(changes) - set(SELF_FIELDS))
        if unknown:
            raise UserValidationError(f"unknown field: {unknown[0]}")
        if not changes:
            raise UserValidationError(f"provide at least one of {', '.join(SELF_FIELDS)}")
        display_name = _validated_display_name(changes["display_name"])
        async with self._database.transaction() as connection:
            exists = (
                await connection.execute(
                    select(users.c.id).where(users.c.external_subject == principal.subject)
                )
            ).first()
            if exists is None:
                raise AccountNotLinkedError(principal.subject)
            await connection.execute(
                update(users)
                .where(users.c.external_subject == principal.subject)
                .values(display_name=display_name)
            )
            row = (
                (
                    await connection.execute(
                        select(users).where(users.c.external_subject == principal.subject)
                    )
                )
                .mappings()
                .one()
            )
        return dict(row)


def _validated_changes(changes: Mapping[str, object]) -> dict[str, object]:
    """Keep the fields the client may change, and refuse the ones it may not name."""
    unknown = sorted(set(changes) - set(ALLOWED_FIELDS))
    if unknown:
        raise UserValidationError(f"unknown field: {unknown[0]}")
    values: dict[str, object] = {}
    if "email" in changes:
        values["email"] = _validated_email(changes["email"])
    if "display_name" in changes:
        values["display_name"] = _validated_display_name(changes["display_name"])
    if "enabled" in changes:
        values["enabled"] = _validated_flag(changes["enabled"])
    if not values:
        raise UserValidationError(f"provide at least one of {', '.join(ALLOWED_FIELDS)}")
    return values


def _validated_flag(enabled: object) -> bool:
    if not isinstance(enabled, bool):
        raise UserValidationError("enabled must be a boolean")
    return enabled


def _validated_email(email: object) -> str:
    if not isinstance(email, str) or not email.strip():
        raise UserValidationError("email must be a non-empty string")
    normalized = email.strip().lower()
    local, separator, domain = normalized.partition("@")
    if not separator or not local or "." not in domain.strip("."):
        raise UserValidationError("email must look like name@example.com")
    if len(normalized) > MAX_EMAIL_LENGTH:
        raise UserValidationError(f"email must be at most {MAX_EMAIL_LENGTH} characters")
    return normalized


def _validated_display_name(display_name: object) -> str:
    if not isinstance(display_name, str) or not display_name.strip():
        raise UserValidationError("display_name must be a non-empty string")
    collapsed = " ".join(display_name.split())
    if len(collapsed) > MAX_DISPLAY_NAME_LENGTH:
        raise UserValidationError(
            f"display_name must be at most {MAX_DISPLAY_NAME_LENGTH} characters"
        )
    return collapsed


def _validated_limit(limit: int) -> int:
    if not 1 <= limit <= MAX_LIMIT:
        raise UserValidationError(f"limit must be between 1 and {MAX_LIMIT}")
    return limit


def _validated_offset(offset: int) -> int:
    if offset < 0:
        raise UserValidationError("offset must not be negative")
    return offset

examples/user_management/app.py composes the application: it registers the providers, then the routes, and hands each request 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 lesson scripts start from an empty file and call migrate(settings) first, which applies the application's revisions to it. That is the same call the deploy step makes — lesson 7 shows the files it runs — and it is why the temporary database in the script below ends up with a real users table:

examples/user_management/migrate.py

"""Apply the application's migrations.

This is the deploy step: a command someone runs against a database, never something the application
does to itself at startup.

Usage: python examples/user_management/migrate.py [path/to/app.db]
"""

from __future__ import annotations

import sys
from pathlib import Path

from xyberos.database import SqliteSettings, run_migrations

MIGRATIONS = Path(__file__).parent / "migrations"


def migrate(settings: SqliteSettings, *, revision: str = "head") -> None:
    """Apply every revision up to ``revision`` to this database."""
    file_path = settings.file_path
    if file_path is not None:
        file_path.parent.mkdir(parents=True, exist_ok=True)
    run_migrations(MIGRATIONS, url=settings.sync_url, revision=revision)


def main(argv: list[str]) -> int:
    """Apply the migrations, for a deployment step or a local reset."""
    settings = SqliteSettings(path=argv[1] if len(argv) > 1 else "var/app.db")
    migrate(settings)
    print(f"migrated {settings.sync_url} to head")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))

What the lesson runs¶

examples/learn_traditional/traditional04_users.py

"""Lesson 4 — creating and listing users through the application's own routes.

Run it with:

    python -m learn_traditional.traditional04_users
"""

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")

USERS = [
    {"email": "  ADA@Example.com ", "display_name": "Ada   Lovelace"},
    {"email": "ada@example.com", "display_name": "Ada Again"},
    {"email": "not-an-email", "display_name": "Nobody"},
    {"email": "grace@example.com", "display_name": "Grace Hopper"},
]


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)
            print("health:", client.get("/health").json())

            for payload in USERS:
                response = client.post("/users", json=payload)
                print("POST /users ->", response.status_code, response.json())

            listing = client.get("/users")
            print("GET /users ->", listing.status_code)
            print(listing.json())


if __name__ == "__main__":
    main()

How to run it¶

PowerShellPOSIX

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

PYTHONPATH=src:examples python -m learn_traditional.traditional04_users

The script borrows the trick from lesson 3 — a database in a temporary directory, created and thrown away — and drives the application through create_test_client, so nothing is typed into a terminal that the suite does not also run.

What it prints¶

health: {'status': 'ready'}
POST /users -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': None}
POST /users -> 409 {'error': {'type': 'DuplicateEmailError', 'message': 'ada@example.com is already registered'}}
POST /users -> 422 {'error': {'type': 'UserValidationError', 'message': 'email must look like name@example.com'}}
POST /users -> 200 {'id': 2, 'email': 'grace@example.com', 'display_name': 'Grace Hopper', 'enabled': True, 'external_subject': None}
GET /users -> 200
{'items': [{'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': None}, {'id': 2, 'email': 'grace@example.com', 'display_name': 'Grace Hopper', 'enabled': True, 'external_subject': None}], 'total': 2, 'limit': 20, 'offset': 0}

Four requests, four answers, and three of them are the lesson:

The first POST sends " ADA@Example.com " and "Ada Lovelace". The stored row is ada@example.com and Ada Lovelace, because normalising input is the service's job and not the caller's good manners. A client that sends ada@example.com later is the same user.
The second POST is that same address again, and it is refused with 409. The check happens inside the write transaction, and the column is unique as well: the first is the rule, the second is what happens when two requests race and both pass the check.
The third POST is not an email, and it is refused with 422. Both refusals come back in the shape lesson 2 taught: an error object with a type and a message.
The fourth POST succeeds, and GET /users returns both rows.

Where each rule lives¶

Input	Rule	Result
`" ADA@Example.com "`	trim, then lowercase	stored as `ada@example.com`
`"not-an-email"`	one `@`, a local part, a dotted domain	`422`
`"Ada Lovelace"`	collapse runs of whitespace	stored as `Ada Lovelace`
a taken address	`select` inside the write transaction	`409`
`limit=0`	between 1 and 100	`422`

Three of these rules are one line of code each, and they are all in examples/user_management/service.py rather than in the route. That is not tidiness: the rules belong to the operation, so the same method can be called from a CLI, a background job, or a test without a request existing.

Why the collection is a page, not a list¶

GET /users returns items, total, limit, and offset, and the limit defaults to 20. The alternative — returning a bare list — is the shape that quietly works until a table grows, and then becomes a response nobody planned for. A collection endpoint that cannot be bounded is a bug waiting for data.

Lesson 5 accepts ?limit= and ?offset= from the client, which means turning untrusted text into integers and refusing the values that make no sense. Until then the defaults apply.

Common mistakes¶

Validating in the route. The route's job is to read a request and call the service. Once a rule is in the route, the CLI cannot reuse it and the test has to build a request to check a rule.
Checking for a duplicate outside the transaction. select outside the with block, then insert inside it, is a race: two requests can both find nothing and both insert. Do both inside one transaction, and keep the unique constraint as the backstop.
Trusting payload.get("email") to be a string. It is whatever the JSON body held — a number, a list, None — which is why the service takes object and checks with isinstance before it touches it.
Leaving the service to raise an HTTP-shaped error. UserValidationError is a RequestValidationError and DuplicateEmailError is an HttpError, but the service only raises them; the adapter turns them into 422 and 409. A service that builds responses cannot be called from a job.
Creating the schema in the application. The tables arrive from migrations/ (lesson 7), applied by a command someone ran against the database, not by the application as it starts.

Exercise¶

Add set_enabled(user_id, enabled) to the service and a route that calls it — POST /users/{user_id}/deactivate. A user that does not exist must be a 404, and the response must be the updated row.

Show the answer

examples/learn_traditional/traditional04_exercise.py

"""Lesson 4, exercise — deactivate a user, and the 404 when there is no such user.

The answer to the lesson's exercise. In your project the `set_enabled` method goes into
``service.py`` and the route into ``app.py``; they are a subclass here so the answer stays runnable.
"""

from __future__ import annotations

import tempfile
from pathlib import Path
from typing import Annotated

from sqlalchemy import select, update
from user_management.app import create_app
from user_management.errors import UserNotFoundError
from user_management.identity import PrincipalVerifier
from user_management.migrate import migrate
from user_management.service import 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")


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

    async def set_enabled(self, user_id: int, enabled: bool) -> dict[str, object]:
        """Enable or disable one user, or raise because there is no such user."""
        async with self._database.transaction() as connection:
            result = await connection.execute(
                update(users).where(users.c.id == user_id).values(enabled=enabled)
            )
            if result.rowcount == 0:
                raise UserNotFoundError(user_id)
            row = (
                (await connection.execute(select(users).where(users.c.id == user_id)))
                .mappings()
                .one()
            )
        return dict(row)


def create_admin_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(UserAdmin)

    @app.post("/users/{user_id}/deactivate")
    async def deactivate(user_id: int, admin: Annotated[UserAdmin, Inject()]) -> dict[str, object]:
        return await admin.set_enabled(user_id, enabled=False)

    return app


def main() -> None:
    """Create one user, deactivate it, then deactivate a user that does not exist."""
    with tempfile.TemporaryDirectory() as directory:
        settings = SqliteSettings(path=Path(directory) / "users.db")
        migrate(settings)

        app = create_admin_app(settings=settings)
        with create_test_client(HttpAdapter(app)) as client:
            client.headers.update(ADMIN_HEADERS)
            created = client.post(
                "/users", json={"email": "ada@example.com", "display_name": "Ada Lovelace"}
            )
            print("created:", created.status_code, created.json())

            deactivated = client.post("/users/1/deactivate")
            print("deactivated:", deactivated.status_code, deactivated.json())

            missing = client.post("/users/99/deactivate")
            print("missing:", missing.status_code, missing.json())


if __name__ == "__main__":
    main()

created: 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': None}
deactivated: 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': False, 'external_subject': None}
missing: 404 {'error': {'type': 'UserNotFoundError', 'message': 'user 99 was not found'}}

The `update` reports how many rows it touched, and `rowcount == 0` is the only trustworthy way to learn that the user was not there: a `select` first and an `update` second is the same race as the duplicate check. `UserNotFoundError` is raised inside the transaction, so a missing user costs one round trip and no partial write.

Next¶

Lesson 5 takes the page apart: ?limit= and ?offset= from the query string, what a bad value gets, and why the bound is enforced twice.