Gå til innhold

Løsningsforslag

Det komplette APIet, i to varianter avhengig av hva slags ID-regime du velger.

For å kjøre dem:

cd code
uv run fastapi dev main.py
uv run fastapi dev main_autoid.py

main.py

APIet fra steg 3-7. Endepunkter for GET all, GET single, DELETE, og et PUT-endepunkt som håndterer både create og delete.

Det kunne vært skrevet enda mer fancy. Arv, mer bruk av Annotated... Men jeg har holdt meg til det mest grunnleggende for å fokusere på FastAPI spesifikt.

main.py
"""Solution key: the CRUD API participants build.

Written for an audience whose Python is not necessarily strong, so it avoids things
that are idiomatic but hard to read cold: no model inheritance, no walrus operator, no
`**kwargs` unpacking, and `Annotated` appears exactly once (on the path parameter,
where there's no good alternative).

Two deliberate design points, both from section 5:

1. There is no POST. Identity is a client-chosen slug, so `PUT /pokemon/{slug}` covers
   both create and replace, and a second write verb would earn nothing.
2. `PokemonUpdate` has no slug, because the URL already said which one. Identity
   arriving in one place only means it can never contradict itself.

Data lives in a module-level dict, so it vanishes every time the server reloads. That's
not an oversight -- it's the closing note of the workshop and the opening of the
follow-up workshop.

`main_autoid.py` is the other design (server-generated ids and POST), for the optional
demo in section 5.
"""

from enum import StrEnum
from typing import Annotated

from fastapi import FastAPI, HTTPException, Path, Response, status
from pydantic import BaseModel, Field

app = FastAPI(
    title="Pokédex",
    summary="A toy CRUD API, built to demonstrate rather too many FastAPI features.",

)

# The shape of an identifier: lowercase letters and digits, in hyphen-separated groups.
# URL-safe, unambiguous about capitals, and impossible to get a space or a slash into.
SLUG_PATTERN = r"^[a-z0-9]+(-[a-z0-9]+)*$"

# `Annotated` is the one piece of intimidating syntax we can't avoid: it means "a string,
# and here's some extra information about it". The extra information is the rule above,
# which FastAPI uses both to reject bad requests and to document the parameter.
SlugPath = Annotated[str, Path(pattern=SLUG_PATTERN, description="The Pokémon's slug.")]


class Type(StrEnum):
    """A closed set of options. Anything else is rejected before our code runs."""

    ELECTRIC = "electric"
    FIRE = "fire"
    WATER = "water"
    GRASS = "grass"
    FLYING = "flying"
    PSYCHIC = "psychic"
    STEEL = "steel"
    ICE = "ice"


class PokemonUpdate(BaseModel):
    """What a client sends us. No slug: the URL already said which Pokémon."""

    display_name: str = Field(
        min_length=1,
        max_length=64,
        description="Shown to humans. May change, and may collide with others.",
        examples=["Vulpix"],
    )
    type1: Type = Field(description="The primary type.")
    type2: Type | None = Field(
        default=None, description="The secondary type, for dual-type Pokémon."
    )


class Pokemon(BaseModel):
    """What we store and send back: the client's data plus the identity it lives under."""

    slug: str = Field(
        pattern=SLUG_PATTERN,
        description="Stable identifier, chosen by the client.",
        examples=["vulpix-alola"],
    )
    display_name: str
    type1: Type
    type2: Type | None = None


# Our entire database. It is a dict, it lives in memory, and it does not survive a
# restart. Section 8 has opinions about that.
datastore: dict[str, Pokemon] = {}


@app.get("/pokemon", summary="List Pokémon, optionally filtered by type")
async def get_all_pokemon(type: Type | None = None) -> list[Pokemon]:
    all_pokemon = list(datastore.values()) 
    if type is None:
        return all_pokemon
    return [p for p in all_pokemon if type in (p.type1, p.type2)]


@app.get(
    "/pokemon/{slug}",
    summary="Fetch a single Pokémon",
    responses={404: {"description": "No Pokémon with that slug"}},
)
async def get_pokemon(slug: SlugPath) -> Pokemon:
    if slug not in datastore:
        raise HTTPException(status.HTTP_404_NOT_FOUND, f"No Pokémon with slug {slug!r}")
    return datastore[slug]


@app.put(
    "/pokemon/{slug}",
    summary="Create or replace a Pokémon",
    responses={201: {"description": "Created a new Pokémon"}},
)
async def put_pokemon(
    slug: SlugPath, update: PokemonUpdate, response: Response
) -> Pokemon:
    """Create or replace -- the whole write side of the API.

    Send this twice and you get the same single Pokémon, which is exactly what PUT
    promises and exactly why we don't need a POST.
    """
    # 201 if we're creating, 200 if we're replacing. We can't know which until we look,
    # so the status code gets set here rather than on the decorator.
    if slug not in datastore:
        response.status_code = status.HTTP_201_CREATED

    stored = Pokemon(
        slug=slug,
        display_name=update.display_name,
        type1=update.type1,
        type2=update.type2,
    )
    datastore[slug] = stored
    return stored


@app.delete(
    "/pokemon/{slug}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Delete a Pokémon",
    responses={404: {"description": "No Pokémon with that slug"}},
)
async def delete_pokemon(slug: SlugPath) -> None:
    if slug not in datastore:
        raise HTTPException(status.HTTP_404_NOT_FOUND, f"No Pokémon with slug {slug!r}")
    del datastore[slug]

main_autoid.py

Designet for case B i steg 5 — en hendelseslogg der serveren bestemmer ID. Verdt å lese og sammenligne med main.py.

  • Den har POST for å håndtere creates.
  • POST-endepunktet returnerer en Location-header, sånn at klienten får vite hvor nyopprettede objekter bor.
  • PUT er kun for oppdateringer, og svarer 404 dersom IDen ikke finnes.
main_autoid.py
"""Presenter's file: the other design, for the optional live demo in section 5.

Only open this if someone asks "but what if the client can't pick the identifier?" --
which is the right question, and has a real answer. Some domains genuinely can't:

- incidents, log entries, orders: no natural name, and two can be identical
- anything where the client shouldn't get to choose (guessable URLs, tenant leakage)

When the *server* owns the identifier, PUT-as-upsert stops working: the client can't
name a URL it doesn't know yet. That's precisely when POST earns its place, and it comes
with a Location header, because the response is the client's only chance to learn where
the thing landed.

Run it beside main.py:
    uv run fastapi dev main_autoid.py --port 8001

Demo script:
    curl -i -X POST localhost:8001/incidents -H 'content-type: application/json' \
      -d '{"title": "Database down", "severity": "high"}'
    # -> 201, and a Location header. Run it again: a second incident, different id.
    #    Compare with PUT in main.py, where running it twice changes nothing.
"""

import uuid
from enum import StrEnum

from fastapi import FastAPI, HTTPException, Response, status
from pydantic import BaseModel, Field

app = FastAPI(
    title="Incidents (server-generated IDs)",
    summary="The design where POST is right: the server owns the identifier.",
)


class Severity(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class IncidentCreate(BaseModel):
    """What the client sends. No id -- it isn't theirs to choose."""

    title: str = Field(
        min_length=1,
        max_length=200,
        description="Free text. Two incidents may legitimately share a title.",
        examples=["Database down"],
    )
    severity: Severity


class Incident(BaseModel):
    """What we store and return: the client's data plus the id we assigned."""

    id: uuid.UUID = Field(description="Assigned by the server. Clients cannot choose it.")
    title: str
    severity: Severity


datastore: dict[uuid.UUID, Incident] = {}


@app.get("/incidents", summary="List all incidents")
async def list_incidents() -> list[Incident]:
    return list(datastore.values())


@app.get(
    "/incidents/{incident_id}",
    summary="Fetch one incident",
    responses={404: {"description": "No such incident"}},
)
async def get_incident(incident_id: uuid.UUID) -> Incident:
    if incident_id not in datastore:
        raise HTTPException(status.HTTP_404_NOT_FOUND, f"No incident {incident_id}")
    return datastore[incident_id]


@app.post(
    "/incidents",
    status_code=status.HTTP_201_CREATED,
    summary="File a new incident",
)
async def create_incident(to_create: IncidentCreate, response: Response) -> Incident:
    """The case POST is actually for: the client can't know the URL in advance.

    Not idempotent, and honestly so. Two identical POSTs mean two incidents, because
    "the database went down again" is a real thing that happens. Compare PUT in
    main.py, where two identical requests are indistinguishable from one.
    """
    incident = Incident(
        id=uuid.uuid4(), title=to_create.title, severity=to_create.severity
    )
    datastore[incident.id] = incident
    # The Location header now carries information the client had no way to compute --
    # this is the header doing its actual job, unlike on an upsert PUT.
    response.headers["Location"] = f"/incidents/{incident.id}"
    return incident


@app.put(
    "/incidents/{incident_id}",
    summary="Replace an existing incident",
    responses={404: {"description": "No such incident"}},
)
async def replace_incident(
    incident_id: uuid.UUID, update: IncidentCreate
) -> Incident:
    """PUT is still here, but it only replaces -- it cannot create.

    Worth saying out loud during the demo: this is the one arrangement where "POST
    creates, PUT updates" is genuinely true. It's true because the server owns the
    identifier, not because it's a rule about the verbs.
    """
    if incident_id not in datastore:
        raise HTTPException(status.HTTP_404_NOT_FOUND, f"No incident {incident_id}")
    updated = Incident(
        id=incident_id, title=update.title, severity=update.severity
    )
    datastore[incident_id] = updated
    return updated


@app.delete(
    "/incidents/{incident_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Delete an incident",
    responses={404: {"description": "No such incident"}},
)
async def delete_incident(incident_id: uuid.UUID) -> None:
    if incident_id not in datastore:
        raise HTTPException(status.HTTP_404_NOT_FOUND, f"No incident {incident_id}")
    del datastore[incident_id]