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.
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
uv add "fastapi[standard]"# 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"}uv run fastapi dev src/notes_api/main.py 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:
@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:
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.
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: datetimefrom 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 noteFastAPI reads the JSON body, validates it against NoteCreate, and passes you a typed object. An empty title gets a 422 with a precise error:
{
"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:
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:
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:
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:
# src/notes_api/routes/notes.py
from fastapi import APIRouter
router = APIRouter(prefix="/notes", tags=["notes"])
@router.get("")
def list_notes() -> list[Note]:
...# 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 defendpoints run on the event loop. Use them when youawaitasync libraries (httpx.AsyncClient, an async database driver). Never call blocking code inside them, or every request stalls (the async lesson explains why).- Plain
defendpoints 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:
# 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:
- Implement
POST /notes,GET /notes(withtag,limitandoffset),GET /notes/{id},PATCH /notes/{id}andDELETE /notes/{id}. - Every note belongs to the user who created it. Require the
CurrentUserdependency 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). - 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
# 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]# 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 == 401StoredNote 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
HTTPExceptionfor client errors, and share cross-cutting needs such as authentication and sessions throughDepends. - Organise with
APIRouter, create shared resources inlifespan, preferdefunless everything is async, and test withTestClientanddependency_overrides.
Check yourself
Quiz
0/4 correct · 4 left
1.A client sends GET /notes/abc to an endpoint declared as get_note(note_id: int). What happens?
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.Why does PATCH use changes.model_dump(exclude_unset=True)?
4.How do you make tests bypass real API-key authentication?
Finished reading?
Mark it complete to fill in your roadmap.