Swap the hardcoded list for the real database, then spend most of the module on the unhappy path — because that is where most backend work lives.
Module 02 made you build endpoints over a Python list on purpose. This is the module where that decision pays for itself: the URLs do not move, the response shapes do not move, only the inside of each function changes.
places list from Module 02 with real SQLAlchemy queries against the models and get_db dependency from Module 05.db: Session = Depends(get_db) to the signature, build a select() statement, and hand the result to a response model.Here is one endpoint, fully, before and after. Read the two blocks side by side and note how little of the HTTP layer is involved in the change.
# BEFORE - app/main.py, the Module 02 version
places = [
{"id": 1, "name": "Blue Bottle", "neighbourhood": "Hayes Valley", "rating": 4},
]
@app.get("/places")
def list_places():
return places # storage IS the endpoint's local variable
# AFTER - app/routers/places.py
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.db import get_db # Module 05: yields a Session, closes it after
from app.models import Place # Module 05: the mapped class
from app.schemas import PlaceOut # new in this module
router = APIRouter(prefix="/places", tags=["places"])
@router.get("", response_model=list[PlaceOut])
def list_places(db: Session = Depends(get_db)):
stmt = select(Place).order_by(Place.id)
return db.scalars(stmt).all() # storage is now a separate system
Two things in that block are new and worth naming. Depends(get_db) is FastAPI asking the dependency system for a session before your function runs and closing it after — you never open or close one yourself. And db.scalars(stmt) rather than db.execute(stmt): execute hands back Row tuples like (<Place 1>,), while scalars unwraps the first column of each row, which is the Place object you actually wanted. Beginners hit AttributeError: 'Row' object has no attribute 'name' here constantly.
db.get(Place, place_id) beats select().where(id == ...) for lookups by primary key. It checks the session's identity map first and only queries if it must. Use it for every "fetch one by id" and save select() for anything with a filter.db.commit(). db.add(place) only stages the object. If you return before committing, the finally in get_db closes the session, the transaction rolls back, and your endpoint returns a 201 describing a row that does not exist. This is the single most common bug in this module.This is the most important design idea in the module, and it is worth more than everything else here combined. The thing a client is allowed to send you and the thing you are willing to send back are two different shapes, and they must be two different classes.
app/schemas.py, imported by routers — never by the models file.Create, Update, Out. The Out one carries model_config = ConfigDict(from_attributes=True) so an ORM object can be serialised through it.# app/schemas.py
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class PlaceCreate(BaseModel):
model_config = ConfigDict(extra="forbid") # reject unknown keys outright
name: str = Field(min_length=1, max_length=120)
neighbourhood: str = Field(min_length=1, max_length=120)
rating: int = Field(ge=1, le=5)
class PlaceUpdate(BaseModel):
"""Every field optional - this is the body of a PATCH."""
model_config = ConfigDict(extra="forbid")
name: str | None = Field(default=None, min_length=1, max_length=120)
neighbourhood: str | None = Field(default=None, min_length=1, max_length=120)
rating: int | None = Field(default=None, ge=1, le=5)
class PlaceOut(BaseModel):
model_config = ConfigDict(from_attributes=True) # accept an ORM object, not just a dict
id: int
name: str
neighbourhood: str
rating: int
created_at: datetime
Now the concrete leak, because "it is tidier" is not an argument. Suppose you skip PlaceOut and return the ORM object directly — FastAPI will happily serialise every column it finds. Three months later Module 07 adds a users table with password_hash, and someone adds an internal_notes column to places for the moderation queue. The moment that migration runs, both fields appear in the JSON body of an endpoint nobody edited. No code changed. No test failed — your tests assert that name is correct, not that password_hash is absent. The API leaked because a schema changed.
With PlaceOut the same migration changes nothing, because the response is built from an explicit list of five fields. That is the whole argument: an output model makes leaking a field require a deliberate edit to a file called schemas.py, instead of being the default outcome of adding a column.
Input models do the mirror-image job. PlaceCreate has no id and no created_at, so a client that sends {"id": 1, "name": "..."} is not merely ignored — with extra="forbid" it is rejected with a 422 naming the offending key. Fields the server owns must be impossible for a client to set. Without that, Place(**request_body) lets anyone choose their own primary key, and later, their own owner_id.
rating; PATCH must not; the response includes id which neither request has. Any single class that served all three would need every field optional, which means it validates nothing.response_model= is enforcement, not documentation. FastAPI runs the returned object through that model and drops anything not declared. Return a dict with an extra key and it silently disappears from the response — which is the behaviour you want.exclude_unset=True. payload.model_dump(exclude_unset=True) gives you only the keys the client actually sent, so you can tell "field omitted" apart from "field explicitly set to null". Without it, every PATCH wipes the fields it did not mention.schemas.py must not import models.py and vice versa. The moment they know about each other you have lost the separation and the leak comes back through a shared base class.Field(...) constraints for simple limits, @field_validator for one field, @model_validator for rules spanning several.if at the top of the route function. Move it into the schema instead.ValueError inside a validator; FastAPI converts the resulting Pydantic failure into a 422 response automatically.# app/schemas.py (continued)
from pydantic import field_validator, model_validator
class PlaceCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str = Field(min_length=1, max_length=120)
neighbourhood: str = Field(min_length=1, max_length=120)
rating: int = Field(ge=1, le=5)
@field_validator("name", "neighbourhood")
@classmethod
def not_only_whitespace(cls, v: str) -> str:
cleaned = v.strip()
if not cleaned:
raise ValueError("must not be blank")
return cleaned # validators can NORMALISE, not only reject
class TagCreate(BaseModel):
name: str = Field(min_length=1, max_length=40)
@field_validator("name")
@classmethod
def normalise(cls, v: str) -> str:
return v.strip().lower() # "Coffee " and "coffee" become one tag
class PlaceUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str | None = Field(default=None, min_length=1, max_length=120)
rating: int | None = Field(default=None, ge=1, le=5)
@model_validator(mode="after")
def at_least_one_field(self): # cross-field: needs the whole object
if not self.model_dump(exclude_unset=True):
raise ValueError("send at least one field to update")
return self
Note that min_length=1 does not catch " ". A single space is one character, so the constraint passes and a place called " " lands in your database. That gap is exactly why field_validator exists, and it is a good illustration of the general rule: declarative constraints handle shape, custom validators handle meaning.
Send {"name": " ", "neighbourhood": "Mission", "rating": 7} and FastAPI returns 422 Unprocessable Content with this body. Learn to read it — you will see it more often than any other error in your career:
{
"detail": [
{
"type": "value_error",
"loc": ["body", "name"],
"msg": "Value error, must not be blank",
"input": " ",
"ctx": {"error": "must not be blank"}
},
{
"type": "less_than_equal",
"loc": ["body", "rating"],
"msg": "Input should be less than or equal to 5",
"input": 7,
"ctx": {"le": 5}
}
]
}
detail is a list — every failure is reported at once, not just the first. loc is a path: ["body", "name"] means the name key of the request body, and for a nested list it would read ["body", "places", 0, "rating"]. The "Value error, " prefix on your own message is Pydantic's, not yours; it surprises everyone once. Depending on your exact FastAPI and Pydantic versions each entry may also carry a "url" key pointing at the Pydantic error docs — harmless, and safe to ignore.
{"name": with no closing brace and FastAPI cannot even parse it — you get a 422 from FastAPI with "type": "json_invalid", which is one of the few places the two blur. The distinction to hold onto: your validator rejecting a value is 422; a request that never made it as far as your validator is 400.
Every backend engineer is eventually asked "why is this a 400 and not a 409?" in code review. Having an answer — a consistent one, written down — is a large part of what separates a working API from a professional one.
HTTPException calls; the body shape in one place, a global exception handler.raise HTTPException(status_code=..., detail=...) in routes, and @app.exception_handler(...) to catch everything that escapes.| Situation | Code | Why that one |
|---|---|---|
GET /places/999, no such row | 404 | The URL names a resource that does not exist. Nothing is wrong with the request itself. |
Body has rating: 7 | 422 | Parsed fine, failed a rule. The client can fix it by changing a value. |
| Malformed request, or a business rule with no better code | 400 | The generic "you sent something wrong". Reach for it last, not first. |
| Creating a place that already exists; adding a tag twice | 409 | The request is valid but conflicts with the resource's current state. Retrying unchanged will fail identically. |
| No credentials, or expired ones | 401 | "I do not know who you are." Module 07 builds this. |
| Known user, not allowed to do this | 403 | "I know who you are and the answer is still no." Logging in again will not help. |
Your code raised KeyError | 500 | Your bug. The client did nothing wrong and can do nothing about it. |
| Database unreachable | 503 | Temporarily broken, not permanently. Signals "retry later" — see the next section. |
The 404/409 pair is the one people get wrong. Absent is 404; already present is 409. Both are about the state of the resource rather than the shape of the request, which is what separates them from 422.
# app/errors.py
import logging
from fastapi import Request
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from sqlalchemy.exc import IntegrityError, OperationalError
from starlette.exceptions import HTTPException as StarletteHTTPException
log = logging.getLogger("cityguide")
CODES = {400: "bad_request", 401: "unauthenticated", 403: "forbidden",
404: "not_found", 409: "conflict", 422: "validation_failed",
500: "internal_error", 503: "service_unavailable"}
def body(status: int, message: str, details=None):
return {"error": {"code": CODES.get(status, "error"),
"message": message,
"details": details}}
def install(app):
@app.exception_handler(StarletteHTTPException)
async def handled(request: Request, exc):
return JSONResponse(exc.status_code, body(exc.status_code, exc.detail))
@app.exception_handler(RequestValidationError)
async def invalid(request: Request, exc):
return JSONResponse(422, body(422, "Request failed validation",
jsonable_encoder(exc.errors())))
@app.exception_handler(IntegrityError)
async def conflict(request: Request, exc):
log.warning("integrity error on %s: %s", request.url.path, exc.orig)
return JSONResponse(409, body(409, "That resource already exists"))
@app.exception_handler(Exception) # the net under everything
async def unhandled(request: Request, exc):
log.exception("unhandled %s on %s %s",
type(exc).__name__, request.method, request.url.path)
return JSONResponse(500, body(500, "Something went wrong on our side"))
Call errors.install(app) once in main.py. Now every failure — including one you never anticipated — leaves your service as the same JSON shape, so a client can write one error-handling branch instead of one per endpoint. Note that the raw Pydantic errors go into details: they are genuinely useful to a developer, and unlike a stack trace they describe the request, not your source code.
The IntegrityError handler deserves attention. A unique constraint — add one on (name, neighbourhood) with an Alembic migration, using what Module 05 taught — is enforced by Postgres, not by Python. When it fires, SQLAlchemy raises IntegrityError from db.commit(), which without a handler is a 500: your API reporting its own bug when the client sent a perfectly reasonable duplicate. Translating it to 409 is the fix. Even better, catch it locally where you can say something specific:
@router.post("", response_model=PlaceOut, status_code=201)
def create_place(payload: PlaceCreate, db: Session = Depends(get_db)):
place = Place(**payload.model_dump())
db.add(place)
try:
db.commit()
except IntegrityError:
db.rollback() # REQUIRED - the session is unusable until you do
raise HTTPException(
status_code=409,
detail=f"A place named '{payload.name}' already exists in {payload.neighbourhood}",
)
db.refresh(place) # pull back id and created_at from the database
return place
app.use((err, req, res, next) => ...) registered last, with zod or joi schemas parsed at the top of each handler and their failures thrown into it. Django REST Framework uses serializers, which are the same idea as Pydantic schemas but conventionally used for both input and output at once — a long-running argument in that community, and the reason DRF codebases often grow a separate read serializer anyway. Rails uses strong parameters (params.require(:place).permit(:name, :rating)), which is an allowlist for input only; Rails leaves output filtering to a view layer such as Jbuilder or a serializer gem. Same three jobs everywhere: allowlist the input, validate it, and control the output separately.
Every previous module assumed Postgres answers. It will not always answer, and what your API does in that minute is a real design decision — not an edge case to be handled "later".
sqlalchemy.exc.OperationalError, return 503 with a short honest message, and put the detail in your logs instead of the response.Do this now, deliberately, with your server running:
$ docker compose stop db
$ curl -i localhost:8000/places
HTTP/1.1 500 Internal Server Error
content-type: text/plain; charset=utf-8
Internal Server Error
Meanwhile your server terminal fills with forty lines ending in something like:
sqlalchemy.exc.OperationalError: (psycopg.OperationalError) connection failed:
connection to server at "localhost" (127.0.0.1), port 5432 failed:
Connection refused
Is the server running on that host and accepting TCP/IP connections?
(Background on this error at: https://sqlalche.me/e/20/e3q8)
Two problems, and the second one is the serious one. First, the client got a 500, which means "my bug" — so their monitoring pages an engineer who then discovers there is no bug. It should be a 503, which means "not my fault, not permanent, try again". Second: that plain-text body is what a default FastAPI app returns, but flip one flag — FastAPI(debug=True), which people leave on far more often than they admit — and Starlette returns the entire traceback to the client as an HTML page: your file paths, your source lines, your library versions, and error text naming your database host, port and user.
A stack trace in a response is useless to the caller, who cannot fix your code, and valuable to an attacker, who now has a map of your filesystem and dependency versions. Those are the only two audiences. Add the handler:
# app/errors.py - add alongside the others
@app.exception_handler(OperationalError)
async def db_unavailable(request: Request, exc):
log.error("database unreachable on %s %s", request.method,
request.url.path, exc_info=exc) # full detail, server side only
return JSONResponse(
503,
body(503, "The service cannot reach its database right now. Try again shortly."),
headers={"Retry-After": "10"},
)
$ curl -i localhost:8000/places
HTTP/1.1 503 Service Unavailable
retry-after: 10
content-type: application/json
{"error":{"code":"service_unavailable","message":"The service cannot reach its database right now. Try again shortly.","details":null}}
Retry-After is the machine-readable half of the message. A well-written client backs off for that many seconds instead of hammering a database that is already struggling.Exception handler still lets uvicorn log the traceback. Starlette sends your JSON response and then re-raises so the server records it. That is the behaviour you want: structured JSON outward, full detail inward./health returns 200 while every real endpoint returns 503, it is decorative. Module 10 makes it actually run a query.Exception handler is invisible in tests by default. pytest's TestClient re-raises server exceptions instead of returning your 500 response, so a test asserting response.status_code == 500 fails with the original traceback. You need TestClient(app, raise_server_exceptions=False). Module 09 covers this properly — note it now so it does not cost you an hour later.
/places/{id}/tags, /users/{id}/lists — where the path expresses the ownership.@router.post("/{place_id}/tags", response_model=PlaceOut, status_code=201)
def attach_tag(place_id: int, payload: TagCreate, db: Session = Depends(get_db)):
place = db.get(Place, place_id)
if place is None:
raise HTTPException(404, f"No place with id {place_id}")
tag = db.scalar(select(Tag).where(Tag.name == payload.name))
if tag is None: # payload.name is already lowercased
raise HTTPException(404, f"No tag named '{payload.name}'")
if tag in place.tags: # 409, not 400 - state conflict
raise HTTPException(409, f"Place {place_id} already has tag '{tag.name}'")
place.tags.append(tag) # SQLAlchemy writes the join row
db.commit()
db.refresh(place)
return place
Read place.tags.append(tag) carefully. You never mention the place_tags join table. The relationship you declared in Module 05 knows about it, and appending to the Python list is what makes SQLAlchemy emit the INSERT INTO place_tags at commit time. This is the ORM earning its keep — and also the exact place where it stops being obvious what SQL runs, which is the subject of the next section.
# app/routers/users.py
class SavedListOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
places: list[PlaceOut]
@computed_field # derived, not stored - from pydantic
@property
def place_count(self) -> int:
return len(self.places)
@router.get("/{user_id}/lists", response_model=list[SavedListOut])
def user_lists(user_id: int, db: Session = Depends(get_db)):
if db.get(User, user_id) is None:
raise HTTPException(404, f"No user with id {user_id}")
stmt = select(SavedList).where(SavedList.user_id == user_id).order_by(SavedList.name)
return db.scalars(stmt).all()
db.get(User, ...) check, /users/9999/lists returns [] — an empty list is a truthful answer to "the lists of a user who does not exist" only if you squint. It tells the client their user id is fine when it is not.POST /places/7/tags with {"place_id": 7} in the body gives you two sources of truth that can disagree. The URL wins; leave place_id out of the schema entirely.DELETE /places/{id}/tags/{name}, not a flag in a PATCH. The link is the resource. Model it that way and the status codes fall out on their own: 204 on success, 404 if it was not attached.The endpoint you just wrote is slow in a way you cannot see from its code. This is the most common performance bug in every ORM in every language, and it is not exotic — it is what the obvious code does by default.
echo=True, count the SELECTs, then add selectinload() or joinedload() to the statement.Turn on SQL logging in your engine — temporarily, this is far too noisy to leave on:
# app/db.py
engine = create_engine(settings.database_url, echo=True)
Now request a user with three saved lists and watch the terminal:
$ curl -s localhost:8000/users/1/lists > /dev/null
SELECT users.id, users.email FROM users WHERE users.id = %(pk_1)s
SELECT saved_lists.id, saved_lists.name, ... FROM saved_lists
WHERE saved_lists.user_id = %(user_id_1)s ORDER BY saved_lists.name
SELECT places.id, ... FROM places JOIN saved_list_places ON ...
WHERE %(param_1)s = saved_list_places.list_id # list 1
SELECT places.id, ... FROM places JOIN saved_list_places ON ...
WHERE %(param_1)s = saved_list_places.list_id # list 2
SELECT places.id, ... FROM places JOIN saved_list_places ON ...
WHERE %(param_1)s = saved_list_places.list_id # list 3
Five queries for three lists. Nothing in your endpoint asked for those last three — they were emitted by Pydantic reading .places on each object while building the response. A user with fifty lists produces fifty-two queries. Each is a network round trip to Postgres, and round trips, not query complexity, are what make this endpoint slow.
from sqlalchemy.orm import selectinload
stmt = (
select(SavedList)
.where(SavedList.user_id == user_id)
.options(selectinload(SavedList.places)) # <- the entire fix
.order_by(SavedList.name)
)
SELECT saved_lists.id, ... FROM saved_lists WHERE saved_lists.user_id = %(user_id_1)s
SELECT places.id, ..., saved_list_places.list_id FROM saved_list_places
JOIN places ON places.id = saved_list_places.place_id
WHERE saved_list_places.list_id IN (%(p1)s, %(p2)s, %(p3)s)
Two queries, and it stays two whether the user has three lists or three hundred. That is the shape you want: a constant number of queries, not a number that grows with your data.
selectinload() | joinedload() | |
|---|---|---|
| SQL emitted | A second SELECT with IN (...) | One SELECT with a LEFT OUTER JOIN |
| Best for | Collections — one-to-many, many-to-many | Scalars — many-to-one, one-to-one |
| Why | Parent rows are not duplicated, so nothing is transferred twice | One round trip, and a single related row adds no duplication |
| Trap | Two round trips instead of one | Joining a collection multiplies parent rows; combine it with LIMIT and you get the wrong number of parents |
len(self.places) looks like a memory operation and is a database query. The only reliable detection is counting the SQL — which is why echo=True for one request is a habit worth building now.lazy="raise" on relationships once you are eager-loading deliberately. Then any accidental lazy load throws InvalidRequestError instead of quietly issuing a query, and N+1 becomes a loud failure in development rather than a slow endpoint in production.echo=True output is just reading SQL. Without that module this section would be five unreadable blocks of noise.EXPLAIN ANALYZE answers a different question. It tells you why one query is slow. N+1 is a problem of query count, and every one of those fifty-two queries is individually fast. Do not go looking for it in a query plan.select_related() (a JOIN, for foreign keys) and prefetch_related() (a second query, for collections) — the same split as joinedload versus selectinload. Rails calls it includes() and ships a bullet gem that shouts at you in development when you forget. Node's Prisma has include, and Sequelize has eager loading. Any tool that turns rows into objects has this bug available by default.
limit and offset query parameters with a default and an enforced maximum, plus a total count in the response envelope.An unbounded GET /places is fine with 40 rows, fine with 4,000, and takes your service down at 4,000,000 — Postgres streams every row, SQLAlchemy builds four million Python objects, Pydantic serialises them, and your process runs out of memory while holding a connection. The fix is a default, not a feature request.
from typing import Generic, Literal, TypeVar
from fastapi import Query
from sqlalchemy import func
T = TypeVar("T")
class Page(BaseModel, Generic[T]):
items: list[T]
total: int
limit: int
offset: int
@router.get("", response_model=Page[PlaceOut])
def list_places(
db: Session = Depends(get_db),
limit: int = Query(default=20, ge=1, le=100), # default AND ceiling
offset: int = Query(default=0, ge=0),
sort: Literal["id", "name", "rating", "created_at"] = "id",
order: Literal["asc", "desc"] = "asc",
):
column = getattr(Place, sort)
column = column.desc() if order == "desc" else column.asc()
total = db.scalar(select(func.count()).select_from(Place))
items = db.scalars(
select(Place).order_by(column, Place.id).limit(limit).offset(offset)
).all()
return Page(items=items, total=total, limit=limit, offset=offset)
Three details in that block are load-bearing. le=100 is the ceiling: without it a client sends ?limit=999999 and you have an unbounded endpoint again with extra steps. Literal[...] on sort is an allowlist — getattr(Place, sort) on an arbitrary string is how someone sorts by a column you never meant to expose, and FastAPI rejects anything off the list with a 422 before your code runs. And .order_by(column, Place.id) adds the primary key as a tiebreaker: sorting by rating alone leaves rows with equal ratings in an order Postgres may change between calls, so page 2 can repeat a row from page 1 and skip another entirely.
Now the honest paragraph about offset. OFFSET 100000 does not skip ahead — the database produces one hundred thousand rows, sorts them, throws them away, and returns the next twenty. Cost grows with how deep you page, so page 1 is instant and page 5,000 times out. It is also unstable: insert a row while a user is reading, and every subsequent page shifts by one, so they see a duplicate. Cursor (or keyset) pagination fixes both by replacing "skip 100,000" with "give me rows after this specific one" — WHERE (rating, id) < (:last_rating, :last_id) ORDER BY rating DESC, id DESC LIMIT 20 — which an index can jump straight to, at the same cost on page 5,000 as on page 1. The trade-off is that you lose the ability to jump to an arbitrary page number, which is why offset survives in admin tables and cursors dominate infinite-scroll feeds. Do not implement it today. Know the name and know when you will need it.
total later without breaking every client. Wrap it in an object from the first version, even when the only key is items.COUNT(*) is a second query and it is not free on a large table, since Postgres must scan to count. Once that hurts, the usual answers are to cache it, return an approximate count, or drop it entirely and return only "is there a next page".ORDER BY has no defined order. Rows often come back in insertion order by accident, right up until an UPDATE moves one. Pagination without a deterministic sort is not pagination.places list. Rewrite every endpoint with db: Session = Depends(get_db) and select(). Do not change a single URL.app/schemas.py with PlaceCreate / PlaceUpdate / PlaceOut (plus tag and list equivalents), add the field constraints and the two validators, and set response_model on every route.app/errors.py, call install(app) from main.py, and add an Alembic migration creating a unique constraint on (name, neighbourhood) so 409 has something to fire on.POST /places/{id}/tags and GET /users/{id}/lists. Run the second one with echo=True, count the SELECTs, add selectinload(), count again. Write both numbers in your notes.limit / offset / sort / order and the Page envelope to every list endpoint.docker compose stop db mid-session and confirm the client gets structured JSON and no traceback; POST the same place twice and confirm 409; POST rating: 7 and confirm 422. Only then docker compose start db.Every one of these is pass/fail on the status code alone. Run them in order, with the server running and docker compose ps showing the database Up.
$ curl -s -X POST localhost:8000/places \
-H "Content-Type: application/json" \
-d '{"name":"Blue Bottle","neighbourhood":"Hayes Valley","rating":4}' \
-w '\n%{http_code}\n'
{"id":1,"name":"Blue Bottle","neighbourhood":"Hayes Valley","rating":4,
"created_at":"2026-03-01T10:14:22.113456+00:00"}
201 # 201 Created, and the body has an id you did not send
$ curl -s localhost:8000/places -w '\n%{http_code}\n'
{"items":[{"id":1,"name":"Blue Bottle",...}],"total":1,"limit":20,"offset":0}
200 # note the envelope - NOT a bare array
$ curl -s localhost:8000/places/999 -w '\n%{http_code}\n'
{"error":{"code":"not_found","message":"No place with id 999","details":null}}
404 # the URL names something that does not exist
$ curl -s -X POST localhost:8000/places \
-H "Content-Type: application/json" \
-d '{"name":"Blue Bottle","neighbourhood":"Hayes Valley","rating":4}' \
-w '\n%{http_code}\n'
{"error":{"code":"conflict","message":"A place named 'Blue Bottle' already exists in Hayes Valley","details":null}}
409 # the SAME request as the first one. Valid, but conflicting.
$ curl -s -X POST localhost:8000/places \
-H "Content-Type: application/json" \
-d '{"name":" ","neighbourhood":"Mission","rating":7}' \
-w '\n%{http_code}\n'
{"error":{"code":"validation_failed","message":"Request failed validation","details":[
{"type":"value_error","loc":["body","name"],"msg":"Value error, must not be blank",
"input":" ","ctx":{"error":"must not be blank"}},
{"type":"less_than_equal","loc":["body","rating"],
"msg":"Input should be less than or equal to 5","input":7,"ctx":{"le":5}}]}
422 # BOTH failures reported, not just the first
$ curl -s -X POST localhost:8000/places \
-H "Content-Type: application/json" \
-d '{"id":99,"name":"Ritual","neighbourhood":"Mission","rating":5}' \
-w '\n%{http_code}\n'
422 # extra="forbid" rejects a client-chosen id. loc is ["body","id"]
$ curl -s "localhost:8000/places?limit=999" -w '\n%{http_code}\n'
422 # the ceiling holds. le=100 is enforced by FastAPI, not by you
$ docker compose stop db
$ curl -s -i localhost:8000/places | head -3
HTTP/1.1 503 Service Unavailable
retry-after: 10
content-type: application/json
# structured JSON, no traceback, no connection string
$ curl -s localhost:8000/places
{"error":{"code":"service_unavailable","message":"The service cannot reach its database right now. Try again shortly.","details":null}}
$ docker compose start db # put it back before you move on
Write your answers in notes/module-06.md first, in full sentences, before opening any of these. The gap between your answer and the given one is the actual content of this module.
1. Why are the input and output schemas separate classes? Give a concrete way one shared class leaks data.
Because they are different contracts with different threats. Concretely: return the ORM Place directly, or reuse one schema for both directions, and then add an internal_notes column for the moderation queue — or reach Module 07 and add password_hash to users. The next response includes it. No route changed, no test failed, because your tests assert what is present rather than what is absent. An explicit PlaceOut makes leaking require someone to edit schemas.py on purpose. The mirror-image job is on the input side: PlaceCreate omits id and created_at, and extra="forbid" turns a client trying to set them into a 422 rather than a silently ignored field — which matters far more once a row has an owner_id.
2. Your database is unreachable. What does your API return, and why is a stack trace unacceptable?
503 with a short JSON body and a Retry-After header, with the full OperationalError logged server-side. Not 500: 500 means "my bug", which sends someone hunting a bug that does not exist, and it tells a client the request will never work when in fact retrying in ten seconds probably will. The stack trace has exactly two audiences. The caller cannot act on it — they cannot patch your code. An attacker can: it hands over your file paths, source lines, library versions and, in the error text itself, your database host, port, user and database name. Free reconnaissance, published by your own server. The useful information belongs in your logs, where the people who can act on it are looking.
3. Which of your endpoints issues the most SQL queries, and how do you know rather than guess?
GET /users/{id}/lists, and you know because you set echo=True on the engine, made one request, and counted the SELECT lines — five for a user with three lists, fifty-two for a user with fifty. You do not know it by reading the code, and that is the lesson: len(self.places) during response serialisation looks like a memory operation and is a network round trip to Postgres. selectinload(SavedList.places) takes it to a constant two, regardless of list count. Note what does not find this: EXPLAIN ANALYZE answers "why is this one query slow", and every one of those fifty-two queries is individually fast. N+1 is a problem of count, so you must count.
4. Give a concrete request that should return each of 400, 404, 409 and 422, and say what distinguishes them.
400: a request your server cannot make sense of at all, or a business rule with no better code — a body that is not JSON, or a truncated one. 404: GET /places/999 when no row 999 exists — the request is perfect, the resource is absent. 409: POST /places with a name and neighbourhood that already exist, or POST /places/1/tags for a tag already attached — valid request, conflicts with current state, and retrying it unchanged fails identically. 422: POST /places with rating: 7 — parsed fine, broke a rule, and the client can fix it by changing a value. The dividing line: 400 and 422 are about the request; 404 and 409 are about the resource's state. Absent is 404, already present is 409.
5. You wrote db.add(place), returned it, got a 201 — and the row is not in the database. What happened?
You never called db.commit(). add() only stages the object in the session's pending set; the INSERT is emitted when the session flushes, and it is made permanent only by a commit. When your function returned, the finally block in get_db closed the session, which rolled back the open transaction. The 201 was truthful about your code's intent and false about the world. A related version of the same trap: commit but forget db.refresh(place), and place.id and place.created_at may be unavailable or stale, because the database generated those values and your Python object has not been told. Commit writes; refresh reads back.
6. Why catch IntegrityError instead of running a SELECT first to check for duplicates?
Because check-then-insert is a race, not a check. Two requests arrive at the same moment, both run the SELECT, both see nothing, both INSERT — one succeeds and the other raises IntegrityError anyway, so you needed the handler regardless and now you also do a pointless extra query on every write. The unique constraint in Postgres is the only part of this that is genuinely atomic. The general principle is worth more than the example: prefer letting the database enforce an invariant and translating its failure, over trying to predict that failure in application code. One caveat — after an IntegrityError the session is in a failed state, so db.rollback() before you raise, or every later query in that request fails with PendingRollbackError.