Step 12 of 13

Web APIs with FastAPI

Build a typed JSON API with FastAPI: routes, Pydantic request and response models, errors, dependency injection, routers, lifespan resources and tests.

19 min 10 sections

You've called other people's APIs; now you'll build one. FastAPI is the most popular modern Python web framework for JSON APIs. You declare what each endpoint accepts and returns with type hints and Pydantic models, and FastAPI uses those declarations to parse and validate requests, serialise responses and generate interactive documentation, all from the same source. This lesson builds a small notes API, covering routes, models, errors, dependencies, project structure, startup resources and testing.

A first endpoint

bash
uv add "fastapi[standard]"
python
# src/notes_api/main.py
from fastapi import FastAPI

app = FastAPI(title="Notes API")


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}
bash
uv run fastapi dev src/notes_api/main.py
text
   FastAPI   Starting development server
             Serving at: http://127.0.0.1:8000
             API docs: http://127.0.0.1:8000/docs

@app.get("/health") registers the function for GET /health. The return value is converted to JSON. fastapi dev reloads the server whenever you save a file; in production you run fastapi run, which doesn't.

Open http://127.0.0.1:8000/docs: FastAPI has generated interactive documentation (OpenAPI) from your code, where you can call each endpoint from the browser. It stays accurate because it's derived from the same type hints that drive validation.

Path and query parameters

Parameters in the path are declared in the route and the function signature; any other function parameter becomes a query parameter. Type hints drive conversion and validation:

python
@app.get("/notes/{note_id}")
def get_note(note_id: int) -> dict:
    ...


@app.get("/notes")
def list_notes(tag: str | None = None, limit: int = 20, offset: int = 0) -> list[dict]:
    ...

GET /notes/abc never reaches your function: FastAPI responds with a 422 error explaining that note_id must be an integer. GET /notes?limit=5&tag=work calls list_notes(tag="work", limit=5, offset=0). Add constraints with Query:

python
from typing import Annotated
from fastapi import Query


@app.get("/notes")
def list_notes(
    tag: str | None = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> list[dict]:
    ...

Now limit=1000 is rejected before your code runs, which protects the database from a single huge request.

Request and response models

Request bodies are described with Pydantic models. Use separate models for input and output, because they differ: clients don't send an id or created_at, and the server shouldn't return internal fields.

python
from datetime import datetime

from pydantic import BaseModel, Field


class NoteCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    body: str = ""
    tags: list[str] = []


class NoteUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    body: str | None = None
    tags: list[str] | None = None


class Note(BaseModel):
    id: int
    title: str
    body: str
    tags: list[str]
    created_at: datetime
python
from datetime import UTC, datetime

from fastapi import status

NOTES: dict[int, Note] = {}
_next_id = 1


@app.post("/notes", status_code=status.HTTP_201_CREATED)
def create_note(data: NoteCreate) -> Note:
    global _next_id
    note = Note(id=_next_id, created_at=datetime.now(UTC), **data.model_dump())
    NOTES[note.id] = note
    _next_id += 1
    return note

FastAPI reads the JSON body, validates it against NoteCreate, and passes you a typed object. An empty title gets a 422 with a precise error:

json
{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "title"],
      "msg": "String should have at least 1 character",
      "input": "",
      "ctx": {"min_length": 1}
    }
  ]
}

The return annotation -> Note does the reverse: FastAPI validates and serialises the response through the Note model, and drops any field not in it. That makes response models a safety net: if your internal object has a password_hash or owner_email, it can't leak unless it's declared in the response model. Set status_code=201 for creation, as above.

Errors

Raise HTTPException for expected client errors; FastAPI turns it into a JSON response:

python
from fastapi import HTTPException


@app.get("/notes/{note_id}")
def get_note(note_id: int) -> Note:
    note = NOTES.get(note_id)
    if note is None:
        raise HTTPException(status_code=404, detail=f"note {note_id} not found")
    return note


@app.patch("/notes/{note_id}")
def update_note(note_id: int, changes: NoteUpdate) -> Note:
    note = get_note(note_id)
    updated = note.model_copy(update=changes.model_dump(exclude_unset=True))
    NOTES[note_id] = updated
    return updated


@app.delete("/notes/{note_id}", status_code=204)
def delete_note(note_id: int) -> None:
    if NOTES.pop(note_id, None) is None:
        raise HTTPException(status_code=404, detail=f"note {note_id} not found")

model_dump(exclude_unset=True) returns only the fields the client actually sent, which is what makes PATCH work as a partial update: a request with just {"title": "New"} leaves the body and tags alone. Unexpected exceptions (bugs) become a generic 500 without leaking a traceback to the client; they're logged on the server.

Dependencies

Many endpoints need the same things: the current user, a database session, pagination settings. FastAPI's dependency injection lets an endpoint declare what it needs, and FastAPI provides it per request:

python
from typing import Annotated

from fastapi import Depends, Header, HTTPException

API_KEYS = {"dev-key-123": "ana"}


def current_user(x_api_key: Annotated[str | None, Header()] = None) -> str:
    user = API_KEYS.get(x_api_key or "")
    if user is None:
        raise HTTPException(status_code=401, detail="missing or invalid API key")
    return user


CurrentUser = Annotated[str, Depends(current_user)]


@app.get("/me")
def me(user: CurrentUser) -> dict[str, str]:
    return {"user": user}

Header() reads the X-API-Key request header (FastAPI converts the underscore). Any endpoint that takes a CurrentUser parameter is now protected, and the check lives in one function. Dependencies can depend on other dependencies, can be async, and can use yield to clean up after the response, which is the standard way to hand out a database session:

python
def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

Dependencies are also what make testing easy, as you'll see below.

Structure: routers and lifespan

As the API grows, split endpoints into modules with APIRouter:

python
# src/notes_api/routes/notes.py
from fastapi import APIRouter

router = APIRouter(prefix="/notes", tags=["notes"])


@router.get("")
def list_notes() -> list[Note]:
    ...
python
# src/notes_api/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI

from notes_api.routes import notes


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=10)    # startup: create shared resources
    yield
    await app.state.http.aclose()                       # shutdown: release them


app = FastAPI(title="Notes API", lifespan=lifespan)
app.include_router(notes.router)

The lifespan function runs once: code before yield at startup, code after at shutdown. Create expensive shared resources there, such as connection pools and HTTP clients, rather than per request or at import time.

async def or def?

FastAPI supports both, and runs them differently:

  • async def endpoints run on the event loop. Use them when you await async libraries (httpx.AsyncClient, an async database driver). Never call blocking code inside them, or every request stalls (the async lesson explains why).
  • Plain def endpoints run in a thread pool, so blocking calls are fine. Use them with synchronous libraries.

If you're unsure, plain def is the safe default. Switch an endpoint to async def when everything it calls is async.

Testing

TestClient calls your app in-process, without a running server, using an httpx-compatible API:

python
# tests/test_notes.py
from fastapi.testclient import TestClient

from notes_api.main import app, current_user

client = TestClient(app)


def test_create_and_fetch_note():
    created = client.post("/notes", json={"title": "Shopping", "tags": ["home"]})
    assert created.status_code == 201
    note_id = created.json()["id"]

    fetched = client.get(f"/notes/{note_id}")
    assert fetched.json()["title"] == "Shopping"


def test_empty_title_is_rejected():
    response = client.post("/notes", json={"title": ""})
    assert response.status_code == 422


def test_me_with_overridden_user():
    app.dependency_overrides[current_user] = lambda: "test-user"
    try:
        assert client.get("/me").json() == {"user": "test-user"}
    finally:
        app.dependency_overrides.clear()

dependency_overrides replaces a dependency for the duration of a test: skip authentication, point the database session at a test database, swap a real email sender for a fake. Use with TestClient(app) as client: when your app has a lifespan, so startup and shutdown run.

Practice

Complete the notes API:

  1. Implement POST /notes, GET /notes (with tag, limit and offset), GET /notes/{id}, PATCH /notes/{id} and DELETE /notes/{id}.
  2. Every note belongs to the user who created it. Require the CurrentUser dependency on all note endpoints, and make sure users can only see and change their own notes (another user's note should be a 404, not a 403, so IDs don't reveal what exists).
  3. Write tests for: creating and listing; filtering by tag; pagination limits rejecting limit=0; a partial PATCH that leaves other fields alone; and user A getting a 404 for user B's note. Reset the in-memory store between tests with a fixture.
Show solution
python
# src/notes_api/main.py
from datetime import UTC, datetime
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException, Query
from pydantic import BaseModel, Field

app = FastAPI(title="Notes API")

API_KEYS = {"key-ana": "ana", "key-ben": "ben"}


def current_user(x_api_key: Annotated[str | None, Header()] = None) -> str:
    user = API_KEYS.get(x_api_key or "")
    if user is None:
        raise HTTPException(status_code=401, detail="missing or invalid API key")
    return user


CurrentUser = Annotated[str, Depends(current_user)]


class NoteCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    body: str = ""
    tags: list[str] = []


class NoteUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    body: str | None = None
    tags: list[str] | None = None


class Note(BaseModel):
    id: int
    title: str
    body: str
    tags: list[str]
    created_at: datetime


class StoredNote(Note):
    owner: str            # internal; never part of a response


class Store:
    def __init__(self) -> None:
        self.notes: dict[int, StoredNote] = {}
        self.next_id = 1


store = Store()


def owned_note(note_id: int, user: str) -> StoredNote:
    note = store.notes.get(note_id)
    if note is None or note.owner != user:
        raise HTTPException(status_code=404, detail=f"note {note_id} not found")
    return note


@app.post("/notes", status_code=201)
def create_note(data: NoteCreate, user: CurrentUser) -> Note:
    note = StoredNote(id=store.next_id, owner=user, created_at=datetime.now(UTC), **data.model_dump())
    store.notes[note.id] = note
    store.next_id += 1
    return note


@app.get("/notes")
def list_notes(
    user: CurrentUser,
    tag: str | None = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> list[Note]:
    mine = [n for n in store.notes.values() if n.owner == user and (tag is None or tag in n.tags)]
    return mine[offset : offset + limit]


@app.get("/notes/{note_id}")
def get_note(note_id: int, user: CurrentUser) -> Note:
    return owned_note(note_id, user)


@app.patch("/notes/{note_id}")
def update_note(note_id: int, changes: NoteUpdate, user: CurrentUser) -> Note:
    note = owned_note(note_id, user)
    updated = note.model_copy(update=changes.model_dump(exclude_unset=True))
    store.notes[note_id] = updated
    return updated


@app.delete("/notes/{note_id}", status_code=204)
def delete_note(note_id: int, user: CurrentUser) -> None:
    owned_note(note_id, user)
    del store.notes[note_id]
python
# tests/test_notes.py
import pytest
from fastapi.testclient import TestClient

from notes_api import main

client = TestClient(main.app)
ANA = {"X-API-Key": "key-ana"}
BEN = {"X-API-Key": "key-ben"}


@pytest.fixture(autouse=True)
def fresh_store(monkeypatch):
    monkeypatch.setattr(main, "store", main.Store())


def create(title: str, headers=ANA, **extra) -> dict:
    response = client.post("/notes", json={"title": title, **extra}, headers=headers)
    assert response.status_code == 201
    return response.json()


def test_create_and_list():
    create("One")
    create("Two")
    titles = [n["title"] for n in client.get("/notes", headers=ANA).json()]
    assert titles == ["One", "Two"]


def test_filter_by_tag():
    create("Work item", tags=["work"])
    create("Home item", tags=["home"])
    notes = client.get("/notes", params={"tag": "work"}, headers=ANA).json()
    assert [n["title"] for n in notes] == ["Work item"]


def test_limit_must_be_positive():
    assert client.get("/notes", params={"limit": 0}, headers=ANA).status_code == 422


def test_partial_patch_keeps_other_fields():
    note = create("Old", body="keep me", tags=["x"])
    patched = client.patch(f"/notes/{note['id']}", json={"title": "New"}, headers=ANA).json()
    assert (patched["title"], patched["body"], patched["tags"]) == ("New", "keep me", ["x"])


def test_other_users_note_is_not_found():
    note = create("Private", headers=BEN)
    assert client.get(f"/notes/{note['id']}", headers=ANA).status_code == 404


def test_owner_is_not_exposed():
    note = create("Mine")
    assert "owner" not in note


def test_requires_api_key():
    assert client.get("/notes").status_code == 401

StoredNote extends Note with the internal owner field, but every endpoint's return type is Note, so FastAPI's response model strips owner from the JSON, which the last-but-one test proves. The fresh_store fixture swaps in an empty Store for each test via monkeypatch, which works because the endpoints look up store as a module global at call time. Returning 404 for someone else's note avoids confirming that a given ID exists, a small but real information leak.

Key takeaways

  • FastAPI turns type hints and Pydantic models into request parsing, validation, response serialisation and OpenAPI docs at /docs.
  • Path and query parameters come from the function signature; constrain them with Query, and invalid input gets a 422 before your code runs.
  • Use separate input and output models; the response model is also a filter that stops internal fields from leaking.
  • Raise HTTPException for client errors, and share cross-cutting needs such as authentication and sessions through Depends.
  • Organise with APIRouter, create shared resources in lifespan, prefer def unless everything is async, and test with TestClient and dependency_overrides.

Check yourself

Quiz

0/4 correct · 4 left

  1. 1.A client sends GET /notes/abc to an endpoint declared as get_note(note_id: int). What happens?

  2. 2.Your internal note object has an owner_email field, but the endpoint's return type is a Note model without it. What does the client receive?

  3. 3.Why does PATCH use changes.model_dump(exclude_unset=True)?

  4. 4.How do you make tests bypass real API-key authentication?

Finished reading?

Mark it complete to fill in your roadmap.