APIs with FastAPI

Bodies with pydantic models

A POST usually sends a JSON body. Describe its shape with a pydantic model: a class whose annotated fields say the type of each value, plus constraints with Field. Use the model as a parameter's type, and FastAPI reads the body, converts it and validates it, all before your function runs.

import asyncio
from datetime import date
import httpx
from fastapi import FastAPI
from pydantic import BaseModel, Field, field_validator

TODAY = date(2025, 6, 30)

class HikeIn(BaseModel):
    trail: str = Field(min_length=1, max_length=80)
    day: date
    km: float = Field(gt=0, le=100)

    @field_validator("day")
    @classmethod
    def not_in_future(cls, value):
        if value > TODAY:
            raise ValueError("can't be in the future")
        return value

class HikeOut(HikeIn):
    id: int

api = FastAPI()
HIKES = {}

@api.post("/hikes", status_code=201)
def add_hike(hike: HikeIn) -> HikeOut:
    new = HikeOut(id=len(HIKES) + 1, **hike.model_dump())
    HIKES[new.id] = new
    return new

async def main():
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=api), base_url="http://test") as client:
        good = await client.post("/hikes", json={"trail": "Red Mesa", "day": "2025-06-14", "km": "11.5"})
        print(good.status_code, good.json())
        bad = await client.post("/hikes", json={"trail": "", "day": "2025-07-14", "km": -2})
        print(bad.status_code)
        for problem in bad.json()["detail"]:
            print(" ", problem["loc"], problem["msg"])

asyncio.run(main())

Every problem is reported at once, each with its location (["body", "km"]) and a message, which is far more useful to the developer calling your API than a bare "invalid request". status_code=201 makes a successful POST answer 201 Created.

Rules that Field can't express, like "not after today", go in a method marked @field_validator. It receives the already converted value (here a real date, not text), raises ValueError with a short message when the value breaks the rule, and returns the value otherwise. pydantic adds that message to the same list of problems, so the caller sees one consistent kind of error, whichever rule failed.

The return annotation -> HikeOut is the response model: FastAPI converts the result to that shape and sends only its fields. That's an important safety net. If a database row has a password_hash column and the response model doesn't, the hash can never leak into a reply, even if someone returns the whole row by mistake.

Errors and dependencies

When an endpoint finds a problem that validation can't, it raises HTTPException. FastAPI turns it into a JSON error reply with the status you choose:

from fastapi import HTTPException

@api.get("/hikes/{hike_id}")
def get_hike(hike_id: int) -> HikeOut:
    if hike_id not in HIKES:
        raise HTTPException(status_code=404, detail="Hike not found")
    return HIKES[hike_id]

Checks that many endpoints share, like "is there a valid API key?" or "read the page and limit", go in a dependency: a function whose parameters FastAPI fills from the request, just like an endpoint's. Depends(...) asks for its result:

from fastapi import Depends, Header, Query

def require_key(x_api_key: str | None = Header(default=None)):
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Missing or wrong API key")

def paging(limit: int = Query(10, ge=1, le=50), offset: int = Query(0, ge=0)):
    return {"limit": limit, "offset": offset}

@api.delete("/hikes/{hike_id}", status_code=204, dependencies=[Depends(require_key)])
def delete_hike(hike_id: int): ...

@api.get("/hikes")
def list_hikes(page: dict = Depends(paging)): ...

Header() reads a request header (the parameter x_api_key matches the header x-api-key), and Query(10, ge=1, le=50) adds limits to a query parameter. Dependencies can depend on other dependencies, which is how real apps build "the current user" from "the token" from "the header", each piece tested on its own.

async def or def?

FastAPI accepts both. An async def endpoint runs on the event loop, so while it awaits something slow, like a call to another service with an async client, the server handles other requests. A plain def endpoint runs in a worker thread, so blocking code inside it (a regular database driver, a big file read) doesn't freeze everyone else. The rule of thumb: use async def when everything slow in the endpoint is awaited; otherwise use def. Never call blocking code from an async def endpoint. The async clients lesson goes deeper.

(The academy has no threads, so here plain def endpoints run directly, one request at a time. Your code doesn't need to change.)

Documentation for free

From the routes, parameters and models, FastAPI writes an OpenAPI description of the whole API, as the OpenAPI lesson did by hand. api.openapi() returns it as a dictionary, and on a server the pages /docs and /redoc show it as interactive documentation where people can try every endpoint.

On your own computer, put the app in main.py, pip install "fastapi[standard]", and start a development server with fastapi dev main.py. In production, a server such as Uvicorn runs it, usually inside a container, as the containers lesson shows.

import asyncio, hashlib, secrets
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

class SignUp(BaseModel):
    email: str = Field(pattern=r"^[^@\s]+@[^@\s]+$")
    name: str = Field(min_length=1)
    password: str = Field(min_length=12)

class Account(BaseModel):
    id: int
    email: str
    name: str

api = FastAPI()
ROWS = {}

@api.post("/accounts", status_code=201)
def sign_up(form: SignUp) -> Account:
    if any(row["email"] == form.email for row in ROWS.values()):
        raise HTTPException(status_code=409, detail="That email is already registered")
    salt = secrets.token_hex(8)
    row = {"id": len(ROWS) + 1, "email": form.email, "name": form.name, "salt": salt,
           "password_hash": hashlib.sha256((salt + form.password).encode()).hexdigest()}
    ROWS[row["id"]] = row
    return row  # the whole row, by mistake: the response model filters it

async def main():
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=api), base_url="http://test") as client:
        body = {"email": "ada@example.com", "name": "Ada", "password": "a long trail passphrase"}
        first = await client.post("/accounts", json=body)
        print(first.status_code, first.json())
        again = await client.post("/accounts", json=body)
        print(again.status_code, again.json())

asyncio.run(main())