Every request opened two database sessions, and closed none

Quick recapType hints· 2 min

A type hint says what a value should be: def read_note(pk: int) -> NoteOut:. Python itself ignores them at run time; tools read them. FastAPI reads them to check and convert every value a request brings.

Quick recapDecorators· 2 min

A decorator wraps or registers a function. @router.get("/notes/") above def list_notes(): registers the function as the handler for that route, and returns it unchanged.

See the problem

The FastAPI Notes API has a get_db() function that opens a database session. The endpoint calls it, and so does the helper that finds the logged-in user:

def current_user(token: str):
    db = get_db()
    ...

@router.post("/notes/")
def create_note(data: dict, token: str):
    db = get_db()
    me = current_user(token)
    ...

Each request opens two sessions. Nothing ever closes them. Under load the pool runs dry, and every request waits for a session that never comes back. And data: dict takes any JSON, so a note with "owner_id": 2 is saved as Ben’s.

Define the task

Write run(endpoint): the heart of FastAPI’s Depends, for one request.

• Input: an endpoint function. Its parameters may have defaults like db=Depends(get_db); dependencies can have dependencies of their own.
• Output: what the endpoint returns, called with each dependency’s value.
• The rules: a dependency used twice in one request runs once, and both get the same value. A dependency that yields gives what it yields, and the code after its yield runs once the endpoint is done, last opened first.
• The hard part: that clean-up must run even when the endpoint raises.

Example: with get_db used by both the endpoint and current_user, the log is ["open", "user", "close"].

Pause and think

⏸ Pause and think

Before you read on: two functions each asked for a session, and each got a new one. How could they share one, and who should close it?

Give it two minutes with pen and paper before you open a hint.

Hints open one at a time:

Hint 1 of 3

Don’t let each function make its own. Let them declare what they need, and have something hand it over.

Hint 2 of 3

If that something remembers what it already made in this request, it can give the same session to both.

Hint 3 of 3

A generator pauses at yield. Resume it after the endpoint, in a finally, and the code after yield closes the session.

Understand the idea

FastAPI reads type hints. Each endpoint parameter says where it comes from and what it is:

• pk: int in the path /notes/{pk}/ is a path parameter, turned into an int.
• page: int = 1 not in the path is a query parameter: ?page=2.
• data: NoteIn, a Pydantic model, is the JSON body.

If a value doesn’t fit, FastAPI answers 422 with every problem, before your code runs. The runnable Notes API is in examples/fastapi/; start it with ./run.sh examples fastapi and open /docs.

Pydantic models shape what comes in and what goes out.

class NoteIn(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True)
    title: str = Field(min_length=1, max_length=200)
    body: str = ""

Unknown keys like owner_id are ignored. The return type, -> NoteOut, is the response model: only its fields are sent, so owner_id or a password hash in the row stays on the server.

Dependencies hand each endpoint what it needs. db: Annotated[Session, Depends(get_db)] says “call get_db and pass me the result”. A dependency can depend on others; FastAPI solves them all, runs each once per request, and passes the same value to everyone who asked. A dependency that yields is set-up and clean-up in one: the code before yield runs before the endpoint, the code after runs once the response is done. This lesson’s editor builds exactly that.

Routers group routes. router = APIRouter() holds the notes routes; app.include_router(router, prefix="/api") adds them, like a Flask blueprint.

The docs write themselves. FastAPI builds an OpenAPI schema from the hints and models, and shows it at /docs. TestClient sends requests in the same process, for tests.

One request through FastAPI

⤢ Full view

POST /api/notes/ in FastAPI

Parts:

• Client (the request is here)
• Pydantic: NoteIn
• Depends: get_db, current_user
• create_note()
• Response model: NoteOut

Arrows:

• Client sends requests to Pydantic: NoteIn
• Pydantic: NoteIn sends requests to Depends: get_db, current_user
• Depends: get_db, current_user sends requests to create_note()
• create_note() sends requests to Response model: NoteOut
• Response model: NoteOut sends requests to Client

Step 1 of 7. The client sends POST /api/notes/ with {"title": " ", "body": 5} and a Bearer token.

◀ BackPlayNext ▶Reset Ask me first Show as text

★Declare what you need; let the framework hand it over

Endpoints don’t open sessions or find users themselves. They declare them as parameters, and FastAPI builds each one once per request and cleans it up afterwards. That is dependency injection.

Climb the ladder

Three versions of run. The tests share a dependency, use one that yields, and raise inside the endpoint.

Okay

Call each dependency where it is named

The idea: For each Depends default, call its function (solving its own dependencies), and take the first value of a generator.

The endpoint gets its values, and plain dependencies work.

import inspect


class Depends:
    def __init__(self, dependency):
        self.dependency = dependency


def run(endpoint):
    def solve(func):
        kwargs = {}
        for name, param in inspect.signature(func).parameters.items():
            if isinstance(param.default, Depends):
                kwargs[name] = solve(param.default.dependency)  # called every time it is named
        value = func(**kwargs)
        if inspect.isgenerator(value):
            value = next(value)  # take what it yields; the code after yield never runs
        return value

    return solve(endpoint)

Sessions per request

Two

Closed

Never

Why

Each parameter gets a fresh call, and generators are never resumed.

Where it breaks: get_db runs for the endpoint and again for current_user: two sessions. The code after yield never runs, so neither is closed.

Better

Cache per request

The idea: Keep a dict of values made in this request; reuse a dependency’s value if it is there.

The waste in Okay: it made the same thing twice.

Highlighted lines are new in this rung.

import inspect


class Depends:
    def __init__(self, dependency):
        self.dependency = dependency


def run(endpoint):
    cache = {}  # one value per dependency, for this request

    def solve(func):
        kwargs = {}
        for name, param in inspect.signature(func).parameters.items():
            if isinstance(param.default, Depends):
                dep = param.default.dependency
                if dep not in cache:
                    cache[dep] = solve(dep)
                kwargs[name] = cache[dep]
        value = func(**kwargs)
        if inspect.isgenerator(value):
            value = next(value)  # still never finishes the generator
        return value

    return solve(endpoint)

Sessions per request

Onewas Two

Closed

Never

Why

The second request for get_db finds the first one’s session in the cache.

What changed from the previous rung

import inspect


class Depends:
    def __init__(self, dependency):
        self.dependency = dependency


def run(endpoint):
    cache = {}  # one value per dependency, for this request

    def solve(func):
        kwargs = {}
        for name, param in inspect.signature(func).parameters.items():
            if isinstance(param.default, Depends):
                kwargs[name] = solve(param.default.dependency)  # called every time it is named
                dep = param.default.dependency
                if dep not in cache:
                    cache[dep] = solve(dep)
                kwargs[name] = cache[dep]
        value = func(**kwargs)
        if inspect.isgenerator(value):
            value = next(value)  # take what it yields; the code after yield never runs
            value = next(value)  # still never finishes the generator
        return value

    return solve(endpoint)

Where it breaks: the session is shared but still never closed. Every request leaks one connection.

Best

Cache, and finish every generator

The idea: Keep each generator after its first value; in a finally, resume them in reverse order so the code after yield runs.

The waste in Better: it opened things without closing them.

The reverse order matters: a user found with a session must be finished with before the session closes.

Highlighted lines are new in this rung.

import inspect


class Depends:
    def __init__(self, dependency):
        self.dependency = dependency


def run(endpoint):
    cache = {}
    open_generators = []

    def solve(func):
        kwargs = {}
        for name, param in inspect.signature(func).parameters.items():
            if isinstance(param.default, Depends):
                dep = param.default.dependency
                if dep not in cache:
                    cache[dep] = solve(dep)
                kwargs[name] = cache[dep]
        value = func(**kwargs)
        if inspect.isgenerator(value):
            open_generators.append(value)
            value = next(value)  # the code before yield runs now
        return value

    try:
        return solve(endpoint)
    finally:
        for gen in reversed(open_generators):  # last opened, first closed
            next(gen, None)  # the code after yield runs now, even if the endpoint raised

Sessions per request

One

Closed

Always, last firstwas Never

Why

Set-up runs once, and clean-up runs once, in the reverse order, whatever the endpoint does.

What changed from the previous rung

import inspect


class Depends:
    def __init__(self, dependency):
        self.dependency = dependency


def run(endpoint):
    cache = {}  # one value per dependency, for this request
    cache = {}
    open_generators = []

    def solve(func):
        kwargs = {}
        for name, param in inspect.signature(func).parameters.items():
            if isinstance(param.default, Depends):
                dep = param.default.dependency
                if dep not in cache:
                    cache[dep] = solve(dep)
                kwargs[name] = cache[dep]
        value = func(**kwargs)
        if inspect.isgenerator(value):
            value = next(value)  # still never finishes the generator
            open_generators.append(value)
            value = next(value)  # the code before yield runs now
        return value

    return solve(endpoint)
    try:
        return solve(endpoint)
    finally:
        for gen in reversed(open_generators):  # last opened, first closed
            next(gen, None)  # the code after yield runs now, even if the endpoint raised

When Best isn’t best: FastAPI also sends the endpoint’s exception into the generator, so get_db can roll back. Write try: yield db except: db.rollback(); raise. For app-wide things, like a connection pool, use the lifespan, not a per-request dependency.

Write it yourself

Write run(endpoint) from the task. Press Run tests to check your code.

• Read parameters with inspect.signature(func).parameters; a Depends default names the dependency.
• Keep a dict for this request: solve each dependency once, and reuse its value.
• If a value is a generator (inspect.isgenerator), keep it and use next(gen). In a finally, call next(gen, None) on each, last first.

Depends, built by hand

Ctrl + Enter runs

import inspect class Depends: def __init__(self, dependency): self.dependency = dependency def run(endpoint): kwargs = {} for name, param in inspect.signature(endpoint).parameters.items(): if isinstance(param.default, Depends): kwargs[name] = param.default.dependency() return endpoint(**kwargs)

Run testsReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: An endpoint with no dependencies → 'ok'
2. ○Not run yet: A plain dependency → 42
3. ○Not run yet: get_db used by the endpoint and by current_user → ['ana uses db', ['open', 'user', 'close']]
4. ○Not run yet: The endpoint raises → ['open', 'user', 'close', 'raised']
5. ○Not run yet: Extra test
6. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

run()Run onceVisualize

Compare

Way	Sessions per request	Closed	Good	Not so good	Use it when
Okay: call where named	One per use	Never	Simple	Leaks, and repeats work	Plain values with no clean-up
Better: cache per request	One	Never	Shares values	Leaks every request	Never with resources
Best: cache and finish generators	One	Always, last first	What FastAPI does	More moving parts	Understanding Depends; in real code, use FastAPI’s

Common mistakes and edge cases

!A blocking call in async def

async def endpoints run on the event loop. A blocking call there, like time.sleep, requests.get or a sync database query, freezes every request. Use plain def for blocking code (FastAPI runs it in a thread), or await async libraries. The advanced lesson on async covers this.

• data: dict instead of a model. Nothing is checked, and every key gets through. Use a Pydantic model.
• Returning the database row with no response model. Every column is sent, including ones you didn’t mean to. Set the return type to an output model.
• Creating a session at import time. One session shared by every request and thread. Use a yield dependency.
• Catching validation in the endpoint. FastAPI already answers 422 with details. Add your own rules in the model with Field and validators.
• Mutable defaults in models. tags: list[str] = [] is safe in Pydantic, which copies it, but not in plain classes. Prefer Field(default_factory=list) for clarity.

Say it in the interview

Questions to ask first

• Do we need async, for many slow calls at once, or is sync code fine?
• Who calls the API, and how do they log in: cookies or tokens?
• Is the OpenAPI schema part of the contract with other teams?

A 60-second walkthrough

“FastAPI reads type hints. Path, query and body parameters come from the signature, and Pydantic models check the body, so bad input gets a 422 listing every problem. The return type is the response model, so only its fields go out. Shared needs, like a database session or the current user, are dependencies. FastAPI solves them once per request and shares each value. A yield dependency closes its session after the response, even on errors. Routes are grouped with APIRouter. The OpenAPI docs are built from the code, and tests use TestClient with overridden dependencies.”

Likely follow-ups

• def or async def? async def when everything inside awaits; def for blocking code, which FastAPI runs in a threadpool.
• How do you swap the database in tests? app.dependency_overrides[get_db] = get_test_db.
• Why 422 and not 400? The request was well-formed JSON, but its content didn’t fit the schema. FastAPI uses 422 for that.
• FastAPI vs Flask? FastAPI checks types, writes docs and supports async by default; Flask is smaller and older, with a huge set of extensions.

Practice

Three problems, from easy to medium. Each has hints, tests you can run here, and its own Okay, Better and Best answers.

1. Typed path parameters (medium)

Write match_path(template, path, types). Match a path like "/notes/7/" against a template like "/notes/{pk}/". Return None if the route doesn’t match: a different length or a different fixed part. Otherwise convert each {name} with types.get(name, str). If one fails, return {"errors": [...]} with {"loc": ["path", name], "msg": "Input should be a valid int"}.

Example: match_path("/notes/{pk}/", "/notes/abc/", {"pk": int}) gives the error, not None: the route matched, the value didn’t.

Hint 1 of 3

Split both on "/" after stripping the slashes at the ends.

Hint 2 of 3

A different number of parts, or a fixed part that differs, means another route: None, so 404.

Hint 3 of 3

Wrap the conversion in try/except ValueError, and use the type’s __name__ in the message.

Typed path parameters

Ctrl + Enter runs

def match_path(template: str, path: str, types: dict) -> dict | None: return None

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: /notes/7/ → {'pk': 7}
2. ○Not run yet: /tags/7/ → None
3. ○Not run yet: /notes/abc/ → {'errors': [{'loc': ['path', 'pk'], 'msg': 'Input should be a valid int'}]}
4. ○Not run yet: /notes/7/edit/ → None
5. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

match_path()Run onceVisualize

Show the Okay, Better and Best answers

Okay

Take the text

The idea: Pair each template part with a path part; keep the values as text.

It pulls out the value.

def match_path(template: str, path: str, types: dict) -> dict | None:
    names = template.strip("/").split("/")
    parts = path.strip("/").split("/")
    values = {}
    for name, part in zip(names, parts, strict=False):
        if name.startswith("{"):
            values[name[1:-1]] = part  # always text
    return values

/tags/7/

Matches

/notes/abc/

pk = abc

Why

It never compares the fixed parts or converts the values.

Where it breaks: any route of the same shape matches, and pk is "7", so pk + 1 fails later.

Better

Check the route, convert the values

The idea: Refuse different lengths and fixed parts; convert each value with its type.

The waste in Okay: it trusted the shape alone.

Highlighted lines are new in this rung.

def match_path(template: str, path: str, types: dict) -> dict | None:
    names = template.strip("/").split("/")
    parts = path.strip("/").split("/")
    if len(names) != len(parts):
        return None
    values = {}
    for name, part in zip(names, parts, strict=True):
        if name.startswith("{"):
            key = name[1:-1]
            values[key] = types.get(key, str)(part)  # int("abc") raises ValueError
        elif name != part:
            return None
    return values

/tags/7/

No matchwas Matches

/notes/abc/

Crasheswas pk = abc

Why

Only the right route matches, and values have their types.

What changed from the previous rung

def match_path(template: str, path: str, types: dict) -> dict | None:
    names = template.strip("/").split("/")
    parts = path.strip("/").split("/")
    if len(names) != len(parts):
        return None
    values = {}
    for name, part in zip(names, parts, strict=False):
    for name, part in zip(names, parts, strict=True):
        if name.startswith("{"):
            values[name[1:-1]] = part  # always text
            key = name[1:-1]
            values[key] = types.get(key, str)(part)  # int("abc") raises ValueError
        elif name != part:
            return None
    return values

Where it breaks: int("abc") raises, which becomes a 500. The client deserves a 422 that names the bad part.

Best

404 for another route, 422 for a bad value

The idea: Return None when the route differs; collect conversion errors with their location.

The waste in Better: it let a client’s typo crash the server.

Highlighted lines are new in this rung.

def match_path(template: str, path: str, types: dict) -> dict | None:
    names = template.strip("/").split("/")
    parts = path.strip("/").split("/")
    if len(names) != len(parts):
        return None  # 404: a different route
    values, errors = {}, []
    for name, part in zip(names, parts, strict=True):
        if name.startswith("{"):
            key = name[1:-1]
            try:
                values[key] = types.get(key, str)(part)
            except ValueError:
                errors.append(
                    {"loc": ["path", key], "msg": f"Input should be a valid {types[key].__name__}"}
                )
        elif name != part:
            return None
    return (
        {"errors": errors} if errors else values
    )  # 422 when the route matched but a type is wrong

/tags/7/

No match

/notes/abc/

422, with wherewas Crashes

Why

It tells “no such route” apart from “right route, wrong value”, as FastAPI does.

What changed from the previous rung

def match_path(template: str, path: str, types: dict) -> dict | None:
    names = template.strip("/").split("/")
    parts = path.strip("/").split("/")
    if len(names) != len(parts):
        return None
    values = {}
        return None  # 404: a different route
    values, errors = {}, []
    for name, part in zip(names, parts, strict=True):
        if name.startswith("{"):
            key = name[1:-1]
            values[key] = types.get(key, str)(part)  # int("abc") raises ValueError
            try:
                values[key] = types.get(key, str)(part)
            except ValueError:
                errors.append(
                    {"loc": ["path", key], "msg": f"Input should be a valid {types[key].__name__}"}
                )
        elif name != part:
            return None
    return values
    return (
        {"errors": errors} if errors else values
    )  # 422 when the route matched but a type is wrong

When Best isn’t best: FastAPI does this from pk: int, and Starlette’s {pk:int} converter can even skip the route for non-numbers. Use the framework; this shows what loc in a 422 means.

2. A response model (easy)

Write dump(row, fields). Return only the listed fields. A field the row lacks is None. A datetime becomes ISO text, so it can go in JSON.

Example: a row with a password_hash and fields ["id", "title"] gives only id and title.

Hint 1 of 3

Build a new dict from the field list, not from the row.

Hint 2 of 3

row.get(name) gives None for a missing key.

Hint 3 of 3

isinstance(value, datetime) then value.isoformat().

A response model

Ctrl + Enter runs

from datetime import datetime def dump(row: dict, fields: list[str]) -> dict: return dict(row)

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: A row with a password hash → {'id': 7, 'title': 'Tea'}
2. ○Not run yet: body missing from the row → {'id': 7, 'title': 'Tea', 'body': None}
3. ○Not run yet: A full row, with a datetime → {'id': 7, 'title': 'Tea', 'body': None, 'updated_at': '2026-11-08T09:30:00+00:00'}
4. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

dump()Run onceVisualize

Show the Okay, Better and Best answers

Okay

The whole row

The idea: Copy the row.

Nothing is ever missing.

def dump(row: dict, fields: list[str]) -> dict:
    return dict(row)  # everything the row has, secrets too

Secrets

Sent

Dates

Not JSON

Why

It sends whatever the database has.

Where it breaks: the password hash and owner id go to the client, and a datetime can’t be turned into JSON.

Better

Only the listed fields

The idea: Build the dict from the field list.

The waste in Okay: it let the table decide the API.

Highlighted lines are new in this rung.

def dump(row: dict, fields: list[str]) -> dict:
    return {name: row[name] for name in fields}  # KeyError if a field is missing

Secrets

Kept backwas Sent

Dates

Not JSON

Why

The output says what it sends, not the table.

What changed from the previous rung

def dump(row: dict, fields: list[str]) -> dict:
    return dict(row)  # everything the row has, secrets too
    return {name: row[name] for name in fields}  # KeyError if a field is missing

Where it breaks: a row without body raises KeyError, and dates still can’t be sent.

Best

Listed fields, with defaults and JSON types

The idea: Use row.get for each field, and turn datetimes into ISO text.

The waste in Better: it assumed every row was complete.

Highlighted lines are new in this rung.

from datetime import datetime


def dump(row: dict, fields: list[str]) -> dict:
    out = {}
    for name in fields:
        value = row.get(name)  # a missing field is null
        out[name] = value.isoformat() if isinstance(value, datetime) else value  # JSON-ready
    return out

Secrets

Kept back

Dates

ISO textwas Not JSON

Why

Every response has the same shape, ready for JSON.

What changed from the previous rung

from datetime import datetime


def dump(row: dict, fields: list[str]) -> dict:
    return {name: row[name] for name in fields}  # KeyError if a field is missing
    out = {}
    for name in fields:
        value = row.get(name)  # a missing field is null
        out[name] = value.isoformat() if isinstance(value, datetime) else value  # JSON-ready
    return out

When Best isn’t best: a Pydantic response model does this and more: types, nested models, aliases. Set from_attributes=True to read straight from SQLAlchemy rows.

3. Query parameters with every error (medium)

Write read_query(params) for GET /notes/. params holds the query as text. page is an int, at least 1, default 1. size is an int from 1 to 100, default 20. sort is one of title, -title, updated or -updated, default -updated. Return (values, []), or ({}, errors) with every problem, each {"loc": ["query", name], "msg": ...}. Use FastAPI’s words:

• "Input should be a valid integer"
• "Input should be greater than or equal to 1"
• "Input should be less than or equal to 100"
• "Input should be one of -title, -updated, title, updated"

Hint 1 of 3

Loop over page and size with their limits; collect an error and move on instead of returning.

Hint 2 of 3

int() inside try/except ValueError tells you when text isn’t a number.

Hint 3 of 3

Return the values only when the error list is empty.

Query parameters with every error

Ctrl + Enter runs

SORTS = {"title", "-title", "updated", "-updated"} def read_query(params: dict[str, str]) -> tuple[dict, list[dict]]: return {"page": 1, "size": 20, "sort": "-updated"}, []

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: No query → [{'page': 1, 'size': 20, 'sort': '-updated'}, []]
2. ○Not run yet: read_query({'page': '2', 'sort': 'title'}) → [{'page': 2, 'size': 20, 'sort': 'title'}, []]
3. ○Not run yet: read_query({'page': 'abc'}) → [{}, [{'loc': ['query', 'page'], 'msg': 'Input should be a valid integer'}]]
4. ○Not run yet: Three mistakes at once → [{}, [{'loc': ['query', 'page'], 'msg': 'Input should be greater than or equal to 1'}, {'loc': ['query', 'size'], 'msg': 'Input should be less than or equal to 100'}, {'loc': ['query', 'sort'], 'msg': 'Input should be one of -title, -updated, title, updated'}]]
5. ○Not run yet: Extra test
6. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

read_query()Run onceVisualize

Show the Okay, Better and Best answers

Okay

int() and hope

The idea: Convert page and size with int(), and take sort as it is.

Right for valid input.

def read_query(params: dict[str, str]) -> tuple[dict, list[dict]]:
    return {
        "page": int(params.get("page", 1)),
        "size": int(params.get("size", 20)),
        "sort": params.get("sort", "-updated"),
    }, []

?page=abc

Crashes

?size=100000

Allowed

Why

It trusts the query string.

Where it breaks: ?page=abc is a 500, and ?size=100000 loads every note at once.

Better

Check the numbers are numbers

The idea: Return an error for the first value that isn’t digits.

The waste in Okay: a typo crashed the server.

Highlighted lines are new in this rung.

def read_query(params: dict[str, str]) -> tuple[dict, list[dict]]:
    values = {"page": 1, "size": 20, "sort": "-updated"}
    for name in ("page", "size"):
        if name in params:
            if not params[name].isdigit():
                return {}, [{"loc": ["query", name], "msg": "Input should be a valid integer"}]
            values[name] = int(params[name])
    if "sort" in params:
        values["sort"] = params["sort"]
    return values, []  # stops at the first error, and has no limits

?page=abc

422was Crashes

?size=100000

Allowed

Why

Bad text is a client error now.

What changed from the previous rung

def read_query(params: dict[str, str]) -> tuple[dict, list[dict]]:
    return {
        "page": int(params.get("page", 1)),
        "size": int(params.get("size", 20)),
        "sort": params.get("sort", "-updated"),
    }, []
    values = {"page": 1, "size": 20, "sort": "-updated"}
    for name in ("page", "size"):
        if name in params:
            if not params[name].isdigit():
                return {}, [{"loc": ["query", name], "msg": "Input should be a valid integer"}]
            values[name] = int(params[name])
    if "sort" in params:
        values["sort"] = params["sort"]
    return values, []  # stops at the first error, and has no limits

Where it breaks: it stops at the first problem, has no limits on page or size, and takes any sort, which could be any column.

Best

Limits, an allow list, and every error

The idea: Check each value’s type and range, and sort against an allow list; collect all errors in FastAPI’s words.

The waste in Better: it checked type, not meaning, one error at a time.

Highlighted lines are new in this rung.

SORTS = {"title", "-title", "updated", "-updated"}


def read_query(params: dict[str, str]) -> tuple[dict, list[dict]]:
    values: dict = {"page": 1, "size": 20, "sort": "-updated"}
    errors: list[dict] = []

    def error(name: str, msg: str) -> None:
        errors.append({"loc": ["query", name], "msg": msg})

    for name, low, high in (("page", 1, None), ("size", 1, 100)):
        if name not in params:
            continue
        try:
            number = int(params[name])
        except ValueError:
            error(name, "Input should be a valid integer")
            continue
        if number < low:
            error(name, f"Input should be greater than or equal to {low}")
        elif high is not None and number > high:
            error(name, f"Input should be less than or equal to {high}")
        else:
            values[name] = number
    if "sort" in params:
        if params["sort"] in SORTS:
            values["sort"] = params["sort"]
        else:
            error("sort", f"Input should be one of {', '.join(sorted(SORTS))}")
    return ({} if errors else values), errors

?page=abc

422

?size=100000

422was Allowed

Why

This is what Query(ge=1, le=100) and a Literal type give you.

What changed from the previous rung

SORTS = {"title", "-title", "updated", "-updated"}


def read_query(params: dict[str, str]) -> tuple[dict, list[dict]]:
    values = {"page": 1, "size": 20, "sort": "-updated"}
    for name in ("page", "size"):
        if name in params:
            if not params[name].isdigit():
                return {}, [{"loc": ["query", name], "msg": "Input should be a valid integer"}]
            values[name] = int(params[name])
    values: dict = {"page": 1, "size": 20, "sort": "-updated"}
    errors: list[dict] = []

    def error(name: str, msg: str) -> None:
        errors.append({"loc": ["query", name], "msg": msg})

    for name, low, high in (("page", 1, None), ("size", 1, 100)):
        if name not in params:
            continue
        try:
            number = int(params[name])
        except ValueError:
            error(name, "Input should be a valid integer")
            continue
        if number < low:
            error(name, f"Input should be greater than or equal to {low}")
        elif high is not None and number > high:
            error(name, f"Input should be less than or equal to {high}")
        else:
            values[name] = number
    if "sort" in params:
        values["sort"] = params["sort"]
    return values, []  # stops at the first error, and has no limits
        if params["sort"] in SORTS:
            values["sort"] = params["sort"]
        else:
            error("sort", f"Input should be one of {', '.join(sorted(SORTS))}")
    return ({} if errors else values), errors

When Best isn’t best: in FastAPI, write size: Annotated[int, Query(ge=1, le=100)] = 20 and sort: Literal["title", "-title", "updated", "-updated"] = "-updated". The 422 and the docs come with it.

Check yourself

A few quick questions. Ones you miss are remembered, so you can find your weak spots later.

1. Multiple choiceGET /notes/abc/ hits an endpoint with pk: int. What does FastAPI answer?

   404 Not Found500 Internal Server Error422, saying pk should be a valid integer200, with pk as the text abc

   Check answer
2. Spot the bugWhat is wrong with this dependency?

   def get_db():
       db = SessionLocal()
       return db

   The session is never closedDependencies must be asyncIt must take the request as an argumentNothing

   Check answer
3. Multiple choiceThe endpoint and current_user both use Depends(get_db). How many sessions does one request open?

   Two, one for eachOne, shared by bothNone until the first queryOne per database table

   Check answer
4. Pick the patternThe database row has owner_id and password_hash. How do you keep them out of the response?

   Delete them from the row before returningUse async defReturn a plain dictSet the return type to an output model without those fields

   Check answer
5. Fill in the answerIn tests, swap get_db for a test session with app.____[get_db] = get_test_db

   Your answerCheck answer

Flashcards

Add these to your review deck. Each card comes back a little later every time you get it right.

• Card 1Where does a FastAPI parameter come from?Show answer
• Card 2What does a 422 from FastAPI mean?Show answer
• Card 3What is a response model?Show answer
• Card 4What does a yield dependency do?Show answer
• Card 5def or async def for an endpoint?Show answer
• Card 6What is APIRouter?Show answer

Add 6 cards to my review deck

Cheat sheet

• Path {pk} + pk: int; query page: int = 1; body data: NoteIn. Bad input: 422 with every problem.
• Input models ignore unknown keys; the return type is the response model, so only its fields go out.
• Depends: once per request, shared; yield for set-up and clean-up, closed last first, even on errors.
• APIRouter + include_router(prefix=...); TestClient; app.dependency_overrides in tests.
• def for blocking code, async def only if everything inside awaits.

Go deeper

• FastAPI: tutorial, dependencies with yield and response models.
• Pydantic: models and fields.
• The runnable example: examples/fastapi/ in the repository, with 8 tests.
• Next on the Frameworks roadmap: FastAPI async, and the mistake that freezes every request.

Can you do these now?

If you’re not sure of any of these, scroll back to that part before moving on.

• Declare path, query and body parameters with type hints, and read FastAPI’s 422 errors
• Use Pydantic models for input and for response models that hide fields
• Write dependencies with Depends and yield, and explain how they are shared and cleaned up
• Split routes with APIRouter, and test with TestClient

✓Mark as completeI already know this

Status: Not started