Lesson 6 — view, edit, and delete

What you will build

The three routes that finish CRUD: read one user, change one, remove one. The rules are not new — the service has known them since lesson 4 — but three questions arrive with them that a collection endpoint never has to answer: what a missing row looks like, what a partial edit means, and whether a repeated delete is an error.

What you need

Nothing new. examples/user_management/service.py grows three methods and examples/user_management/app.py grows three routes:

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

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

Look at the order inside update_user. The row is looked up first, so a missing user is a 404 even when the body also names an email somebody else already has: two errors are true at once, and the one that describes the resource the request named wins. Only then is the address checked for a collision — and the check excludes the row being edited, because otherwise every edit of a user's own address would report a duplicate of itself.

The lesson's own module drives all three:

examples/learn_traditional/traditional06_crud.py

"""Lesson 6 — reading one user, editing one, and deleting one.

Run it with:

    python -m learn_traditional.traditional06_crud
"""

from __future__ import annotations

import tempfile
from pathlib import Path

from httpx import Response
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"]

CHANGES = [
    (1, {"display_name": "  Ada,   Countess of Lovelace "}),
    (1, {"enabled": "yes"}),
    (1, {"nickname": "Ada"}),
    (1, {"email": "ada@example.com"}),
    (2, {"email": "ada@example.com"}),
]


def show(label: str, response: Response) -> None:
    """Print one response as the lesson reads it: the status, then the body."""
    body = response.json() if response.content else None
    print(f"{label} ->", response.status_code, body)


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()}
                )

            show("GET /users/1", client.get("/users/1"))
            show("GET /users/99", client.get("/users/99"))
            show("GET /users/abc", client.get("/users/abc"))

            for user_id, changes in CHANGES:
                show(
                    f"PATCH /users/{user_id} {changes}",
                    client.patch(f"/users/{user_id}", json=changes),
                )

            show("DELETE /users/3", client.delete("/users/3"))
            show("DELETE /users/3 again", client.delete("/users/3"))

            listing = client.get("/users").json()
            print(
                "GET /users ->",
                [item["email"] for item in listing["items"]],
                "of",
                listing["total"],
            )


if __name__ == "__main__":
    main()

How to run it

PowerShellPOSIX

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

PYTHONPATH=src:examples python -m learn_traditional.traditional06_crud

What it prints

GET /users/1 -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada', 'enabled': True, 'external_subject': None}
GET /users/99 -> 404 {'error': {'type': 'UserNotFoundError', 'message': 'user 99 was not found'}}
GET /users/abc -> 422 {'error': {'type': 'RequestValidationError', 'message': "the path parameter 'user_id' is not valid"}}
PATCH /users/1 {'display_name': '  Ada,   Countess of Lovelace '} -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada, Countess of Lovelace', 'enabled': True, 'external_subject': None}
PATCH /users/1 {'enabled': 'yes'} -> 422 {'error': {'type': 'UserValidationError', 'message': 'enabled must be a boolean'}}
PATCH /users/1 {'nickname': 'Ada'} -> 422 {'error': {'type': 'UserValidationError', 'message': 'unknown field: nickname'}}
PATCH /users/1 {'email': 'ada@example.com'} -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada, Countess of Lovelace', 'enabled': True, 'external_subject': None}
PATCH /users/2 {'email': 'ada@example.com'} -> 409 {'error': {'type': 'DuplicateEmailError', 'message': 'ada@example.com is already registered'}}
DELETE /users/3 -> 204 None
DELETE /users/3 again -> 404 {'error': {'type': 'UserNotFoundError', 'message': 'user 3 was not found'}}
GET /users -> ['ada@example.com', 'grace@example.com'] of 2

Eleven requests, and the interesting ones are the refusals:

Request	Status	Why
GET /users/99	404	the service raised UserNotFoundError, and the adapter turned it into a response
GET /users/abc	422	the path parameter is annotated int, so the framework refused it before the service was called
PATCH with enabled: "yes"	422	JSON has a boolean; a string that looks like one is not it
PATCH with nickname	422	an unknown field is refused rather than ignored
PATCH /users/2 with user 1's email	409	the collision check, excluding the row being edited
DELETE /users/3 twice	204, then 404	the second delete reported that the row was already gone

Note the two different 422s. GET /users/abc never reached the service — the path parameter is typed as int, so the framework caught it. The PATCH refusals did reach it: they are rules about users, and rules about users live in the service.

What PATCH means here

PATCH replaces the fields it names and leaves the rest alone. Three consequences follow, and all three are deliberate:

• An empty body is refused, not treated as "change nothing", because a request that asks for nothing is more likely to be a client bug than an intent.
• A field the service does not know is refused. Accepting {"nickname": "Ada"} silently would mean a client that misspelled display_name would get 200 and no change.
• Only the named fields are validated. {"display_name": "Ada"} does not have to carry a valid email, because it is not changing one.

PUT is the other half of that idea — a replacement that must carry every field — and it is this lesson's exercise.

DELETE twice: 204 or 404?

Both answers are defensible, and the choice is about what the client needs to hear.

Choice	Second DELETE	What the client learns
Strict — what this lesson does	404	whether this request removed the row, which is what an audit trail or a retry loop usually wants
Idempotent	204	that the row is gone, which is all a client that only wants it gone needs

The lesson picks strict because it composes with the rest of the API: a 404 already means "no such user" here, and answering 204 for a user that never existed would make /users/999 a route that reports success. Whichever you pick, pick it for every delete in the application, and say so on the page a client reads.

Common mistakes

• Returning None from a handler and expecting 200. None becomes 204 No Content, which is why DELETE is one line long. Returning a body from a 204 is itself a protocol error.
• Deciding 404 from the update's row count when the body can also fail validation. The order of the checks decides which error a client sees, so put the existence check first and the collision second.
• Forgetting that the collision check must exclude the row being edited. Without users.c.id != user_id, saving a user without changing its email returns 409, and the client cannot fix it.
• Coercing a JSON string into a boolean. "yes", "true", and 1 are not True; accepting them means a client sending "false" eventually enables an account.
• Deleting the row the client is about to need. A DELETE that reports 204 and then removes a user the client still references is a data-model question, not an HTTP one: soft-delete (enabled = 0) is what the enabled column is for when records have to survive.

Exercise

Add PUT /users/{user_id}: a full replacement that requires email, display_name, and enabled, refuses a body that leaves one out, reuses the same collision check, and still answers 404 for a user that does not exist.

Show the answer

examples/learn_traditional/traditional06_exercise.py

"""Lesson 6, exercise — a full replacement, and how it differs from a partial edit.

The answer to the lesson's exercise. In your project the replacement goes into ``service.py`` and
the route into ``app.py``; it is a subclass here so the answer stays runnable without editing the
project.
"""

from __future__ import annotations

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

from httpx import Response
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 ALLOWED_FIELDS, UserService

from xyberos import App, Inject
from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter, read_json_object
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"]

REPLACEMENT = {"email": "ada@example.com", "display_name": "Ada Lovelace", "enabled": False}


class UserRegistry(UserService):
    """``service.py``, plus the replacement the exercise asks for."""

    async def replace_user(
        self, user_id: int, replacement: Mapping[str, object]
    ) -> dict[str, object]:
        """Replace every field of one user, or refuse a body that leaves one out."""
        _required_replacement(replacement)
        return await self.update_user(user_id, replacement)


def _required_replacement(replacement: Mapping[str, object]) -> None:
    """A replacement names every field, because one it leaves out would be dropped silently."""
    unknown = sorted(set(replacement) - set(ALLOWED_FIELDS))
    if unknown:
        raise UserValidationError(f"unknown field: {unknown[0]}")
    missing = [field for field in ALLOWED_FIELDS if field not in replacement]
    if missing:
        raise UserValidationError(f"a replacement needs {', '.join(ALLOWED_FIELDS)}")


def create_registry_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(UserRegistry)

    @app.put("/users/{user_id}")
    async def replace_user(
        user_id: int, request: Request, registry: Annotated[UserRegistry, Inject()]
    ) -> dict[str, object]:
        payload = await read_json_object(request)
        return await registry.replace_user(user_id, payload)

    return app


def show(label: str, response: Response) -> None:
    """Print one response as the lesson reads it: the status, then the body."""
    body = response.json() if response.content else None
    print(f"{label} ->", response.status_code, body)


def main() -> None:
    """Create two users, then replace one three ways, and edit one partially."""
    with tempfile.TemporaryDirectory() as directory:
        settings = SqliteSettings(path=Path(directory) / "users.db")
        migrate(settings)

        app = create_registry_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()}
                )

            show("PUT /users/1", client.put("/users/1", json=REPLACEMENT))
            show(
                "PUT /users/1 without enabled",
                client.put("/users/1", json={"email": "ada@example.com", "display_name": "Ada"}),
            )
            show("PUT /users/99", client.put("/users/99", json=REPLACEMENT))
            show(
                "PATCH /users/2",
                client.patch("/users/2", json={"display_name": "Rear Admiral Grace Hopper"}),
            )


if __name__ == "__main__":
    main()

PUT /users/1 -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': False, 'external_subject': None}
PUT /users/1 without enabled -> 422 {'error': {'type': 'UserValidationError', 'message': 'a replacement needs email, display_name, enabled'}}
PUT /users/99 -> 404 {'error': {'type': 'UserNotFoundError', 'message': 'user 99 was not found'}}
PATCH /users/2 -> 200 {'id': 2, 'email': 'grace@example.com', 'display_name': 'Rear Admiral Grace Hopper', 'enabled': True, 'external_subject': None}

The replacement is three lines because the write itself is not new: it calls the same `update_user`, so the collision rule, the normalisation, and the `404` are shared rather than copied. What `PUT` adds is the *contract* — every field must be present — and that belongs at the edge of the operation, before anything is read or written. The last line is the contrast worth keeping: `PATCH` still edits one field on its own, and the same user comes back with only the name changed.

Next

Lesson 7 takes the schema away from metadata.create_all. Migrations: writing a revision, running it at deploy time, and why the file this lesson's code relies on is a stand-in that has to go.