FastAPI 2 - De Basics

Na de FastAPI en Pydantic introductie uit de vorige les gaan we verder met de FastAPI tutorial.

Herhaling

• FastAPI (en Pydantic) werken op basis van type annotations.
• Gebruikt decorators om path functies te bepalen (zoals Flask).
• Automatische data validatie en conversie (coercion).
• Automatisch gegenereerde OpenAPI Spec and Swagger UI.
• uv add fastapi[standard], uv run fastapi dev

Path Parameters

@app.get("/items/{item_id}")
def read_item(item_id: int):
    print(f"Type of item_id is {type(item_id)}")
    return {"item_id": item_id}

Query Parameters

@app.get("/items/")
def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

Een default waarde maakt de parameter optioneel in de OAS.

Pydantic & Request Bodies

• Data validatie en serialisatie (= omzetten van Python naar bvb. JSON).
• Model en Field.
• Gebruik van Annotated.

main.py

from datetime import datetime
from typing import Annotated
from fastapi import FastAPI
from pydantic import UUID4, BaseModel, ConfigDict, Field, PositiveInt
from pydantic.alias_generators import to_camel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

class LineItem(BaseModel):
    item: Item
    quantity: PositiveInt

class Order(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel)

    order_date: datetime
    customer_uuid: Annotated[ UUID4, Field(examples=["d4a93ebc-65b2-48e0-a804-310347129f07"])]
    line_items: list[LineItem]

app = FastAPI()

@app.post("/orders/{order_type}")
def create_item(order_type: str, order: Order):
    print(f"Received order of type {order_type}: {order}")
    return order

Query Parameter Validatie

https://fastapi.tiangolo.com/tutorial/query-params-str-validations/

Query()

Een query parameter is op zich altijd een gewone string (een substring van de volledige URL). Met een Python base type annotatie (int, bool, …) krijgen we wel conversie, maar voor extra validatie kunnen we geen volledig Pydantic model als type toekennen.

We zouden (via Annotated) gebruik kunnen maken van pydantic.Field. FastAPI voorziet echter specifiek voor dit geval zijn eigen fastapi.Query functie. Deze werkt op precies dezelfde manier als pydantic.Field, maar met extra functionaliteit m.b.t. query parameters.

In het volgende voorbeeld wordt de (optionele!) query parameter q beperkt tot 5 karakters.

main.py

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
def read_items(q: Annotated[str | None, Query(max_length=5)] = None):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Extra Parameters

De Query functie biedt nog tal van andere handige parameters. We bekijken er nog enkele in het volgende voorbeeld.

main.py

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
def read_items(
    query: Annotated[
        str | None,
        Query(
1            description="Only return items matching this query.",
2            alias="q",
3            deprecated=True,
            min_length=3,
        ),
    ] = None,
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if query:
        results.update({"q": query})
    return results

1

Met description kan een beschrijving worden meegegeven. Deze wordt overgenomen in de OAS en getoond in de Swagger UI.

2

q is geen goede naam voor een Python variable. We gebruiken liever bvb. query. Maar in de API spec willen we wel graag q blijven gebruiken. Dat kan met alias.

3

Als we een parameter op termijn willen verwijderen uit de API kan deze als deprecated worden aangegeven. Opnieuw wordt dit overgenomen in de OAS en duidelijk weergegeven in de Swagger UI.

Custom Validatie

Net als bij een volledige Pydantic model kan het zijn dat we nog extra validatie willen uitvoeren op een query parameter. Net als een Pydantic model Field kunnen we gebruik maken van pydantic.AfterValidator, met een functie die als argument (en return value) de desbetreffende query parameter krijgt.

main.py

from typing import Annotated
from fastapi import FastAPI, HTTPException, status
from pydantic import AfterValidator

app = FastAPI()

data = {
    "isbn-9781529046137": "The Hitchhiker's Guide to the Galaxy",
    "imdb-tt0371724": "The Hitchhiker's Guide to the Galaxy",
    "isbn-9781439512982": "Isaac Asimov: The Complete Stories, Vol. 2",
}

def check_valid_id(id: str):
    if not id.startswith(("isbn-", "imdb-")):
        raise ValueError('Invalid ID format, it must start with "isbn-" or "imdb-"')
    return id

@app.get("/items/")
def read_items(
    id: Annotated[str | None, AfterValidator(check_valid_id)] = None,
):
    if id:
        item = data.get(id)
    else:

        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Item not found"
        )
    return {"id": id, "name": item}

Path Parameter Validatie

fastapi.Path is het equivalent van fastapi.Query maar dan voor path parameters.

main.py

from typing import Annotated
from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
def get_item(
    item_id: Annotated[
        int, Path(description="The ID of the item to get", gt=0, le=1000)
    ],
):
    results = {"item_id": item_id}
    return results

Query Parameter Modellen

Een groep gerelateerde query parameters kunnen we eenmalig definiëren als een Pydantic model. FastAPI zal dan automatisch elke veld (field) van het model als een afzonderlijke query parameter beschouwen. Wel moet de type hint in de path operation functie geannoteerd worden met een (eventueel lege) Query functie. FastAPI kan anders immers niet bepalen of het om een Query of Request Body gaat.

main.py

from typing import Annotated, Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()

class FilterParams(BaseModel):
    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []

@app.get("/items/")
def read_items(filter_query: Annotated[FilterParams, Query()]):
    return filter_query

Cookies

Response Cookies

Om een cookie mee te sturen in een response gebruik je een Response parameter in de path functie. Naast cookies kan je hiermee bijvoorbeeld ook extra response headers instellen.

main.py

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/cookie/")
def give_cookie(response: Response):
    response.set_cookie(key="nicecookie", value="lotus")
    response.headers["X-API-Source"] = "Cookiemonster"
    return {"message": "Have a cookie!"}

Request Cookies

Een cookie uit de request maak je beschikbaar in de path functie met Cookie, op dezelfde manier als Query en Path:

main.py

from typing import Annotated
from fastapi import Cookie, FastAPI, Response

app = FastAPI()

@app.get("/cookie/")
def give_cookie(response: Response):
    response.set_cookie(key="nicecookie", value="lotus")
    response.headers["X-API-Source"] = "Cookiemonster"
    return {"message": "Have a cookie!"}

@app.get("/readcookie")
def read_cookie(nicecookie: Annotated[str | None, Cookie(include_in_schema=False)]):
    return {"message": f"Got your cookie: {nicecookie}!"}

Cookies worden in principe automatisch opgeslagen en terug meegestuurd door clients (in ons geval de Swagger UI in een browser). Dus we willen deze niet opnemen in de OAS. Dat kan via de parameter include_in_schema=False. Deze is ook beschikbaar in andere functies zoals Query.

Request Headers

Request headers maak je beschikbaar in de path operation functie via functie argumenten in combinatie met de Header functie. Die laatste werkt op dezelfde manier als bvb. Query of Path.

main.py

from typing import Annotated
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
def read_items(
    custom_header: Annotated[str, Header()],
    user_agent: Annotated[str | None, Header(include_in_schema=False)] = None,
):
    return {"User-Agent": user_agent, "Custom-Header": custom_header}

HTTP headers gebruiken doorgaans een koppelteken (hyphen, -) tussen de woorden, zoals bvb. User-Agent. Maar dit teken mag niet gebruikt worden in Python namen (bvb. variabelen). FastAPI zet deze daarom automatisch om in een liggend streepje (underscore, _).

Een header als User-Agent verwachten we eigenlijk altijd en willend we bijgevolg niet opnemen in de OAS. Dat kan opnieuw met de parameter include_in_schema=False.

Extra: Header Model

Net als bij query parameters kan een groep headers gedefinieerd worden als een Pydantic model en zal FastAPI deze automatisch als afzonderlijke header velden interpreteren.

main.py

from typing import Annotated
from fastapi import FastAPI, Header
from pydantic import BaseModel

app = FastAPI()

class CommonHeaders(BaseModel):
    host: str
    save_data: bool
    if_modified_since: str | None = None
    traceparent: str | None = None
    x_tag: list[str] = []

@app.get("/items/")
def read_items(headers: Annotated[CommonHeaders, Header()]):
    return headers

Respons Model

https://fastapi.tiangolo.com/tutorial/response-model/

Return Value Type Hint

Tot nu toe keken we vooral naar het modelleren van de request. Met FastAPI (en Pydantic) kunnen we ook op dezelfde manier de respons definiëren, m.b.v. een type hint voor de functie return value!

main.py

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list[str] = []

@app.post("/items/")
def create_item(item: Item) -> Item:
    return item

FastAPI gebruikt de type hint om de OAS te vervolledigen. De type-checker van je editor zal ook controleren of we daadwerkelijk een Item object gereturned wordt.

Response_model

Het kan gebeuren dat de data waarover we beschikken in de functie (nog) niet overeenkomt met het type dat we in de respons willen. FastAPI kan de juiste conversie automatisch uitvoeren. I.p.v. een type hint voor de return value (die dus niet meer zou kloppen) gebruiken we dan de response_model parameter in de path operation decorator.

main.py

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list[str] = []

@app.post("/items/")
def create_item(item: Item) -> Item:
    return item

@app.get("/items/", response_model=list[Item])
def read_items():
    return [
        {"name": "Portal Gun", "price": 42.0},
        {"name": "Plumbus", "price": 32.0},
    ]

De read_items returned dus een list van dict’s maar FastAPI zet dit om in een list van Item’s (die om hun beurt dan weer worden omgezet naar de juiste JSON representatie in de response body).

Response_model als Filter

FastAPI zal de return value van de functie ook, in de mate van het mogelijke, aanpassen aan het gedefinieerd response_model. Op die manier kunnen we specifieke data uitfilteren en hebben we zekerheid over de inhoud van de respons.

In onderstaand voorbeeld zorgen we er bvb. voor dat het password veld zeker niet wordt meegestuurd in de respons.

main.py

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None

@app.post("/user/", response_model=UserOut)
def create_user(user: UserIn):
    return user

Extra: Een beter manier…

Er bestaat een beter manier om tot hetzelfde resultaat te komen. Een manier waarmee we wel nog type-validation in editors als VSCode of PyCharm behouden (nu is dat niet het geval want de return type van de functie is onbekend).

UserIn is eigenlijk een uitbreiding van UserOut. Het is UserOut + een extra password veld. Dit kan natuurlijk uitgedrukt wordt via een subclass.

Nu kan de return type van de functie als UserOut worden gedefinieerd. Ook al wordt er eigenlijk een UserIn gereturned. Immers, een UserIn is een UserOut!

main.py

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None

class UserIn(UserOut):
    password: str

@app.post("/user/", response_model=UserOut)
def create_user(user: UserIn) -> UserOut:
    return user

Zie ook https://fastapi.tiangolo.com/tutorial/extra-models/#reduce-duplication

Response_model_exclude

Er zijn een aantal response_model_exclude* parameters (met overeenkomstige Pydantic BaseModel.dict parameters) waarmee we de mogelijkheid krijgen bepaalde velden uit het response model uit te sluiten in de uiteindelijke response.

• response_model_exclude_unset zorgt dat enkel velden die gespecificeerd werden bij het aanmaken van het object (m.a.w. die in de input zaten) worden teruggestuurd
• response_model_exclude_defaults sluit alle default waarden uit
• response_model_exclude_none sluit alle velden met waarde None (null in JSON) uit
• response_model_exclude is een set met veldnamen (als str) die uitgesloten moeten worden

main.py

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float = 10.5
    tags: list[str] = []

items = {
    "foo": Item(name="Foo", price=50.2),
    "bar": Item(name="Bar", price=62, description="The bartenders", tax=20.2),
    "baz": Item(name="Baz", price=50.2, description=None, tax=10.5, tags=[]),
}

@app.get("/read_item_exclude_unset/{item_id}", response_model_exclude_unset=True)
def read_item_exclude_unset(item_id: str) -> Item:
    return items[item_id]

@app.get("/read_item_exclude_defaults/{item_id}", response_model_exclude_defaults=True)
def read_item_exclude_defaults(item_id: str) -> Item:
    return items[item_id]

@app.get("/read_item_exclude_none/{item_id}", response_model_exclude_none=True)
def read_item_exclude_none(item_id: str) -> Item:
    return items[item_id]

@app.get("/read_item_exclude_tax/{item_id}", response_model_exclude={"tax"})
def read_item_exclude_tax(item_id: str) -> Item:
    return items[item_id]

Geen Model

Het is voor FastAPI zeker niet verplicht om een Pydantic model als response type te hebben. Als de structuur van een respons niet vooraf gedefinieerd kan worden kan bvb. een gewone dict door de path operation functie gereturned worden.

main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/keyword-weights/", response_model=dict[str, float])
def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

Respons Status Code

https://fastapi.tiangolo.com/tutorial/response-status-code/#changing-the-default

Naast response_model kan ook status_code worden meegegeven in de path operation decorator. Hiermee wordt aangegeven wat de normale HTTP status code (default 200) is bij een succesvolle respons.

I.p.v. een int kan ook gebruik worden gemaakt van de handige fastapi.status module.

main.py

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(name: str):
    return {"name": name}

Ook hier wordt de toegevoegde informatie doorgevoerd in de OAS.

Forms

Als Parameters

Data uit Form velden wordt beschikbaar via functie argumenten geannoteerd met Form.

main.py

from typing import Annotated
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
def login(username: Annotated[str, Form()], password: Annotated[str, Form()]):
    return {"username": username}

Form Models

Net als bij een JSON request Body kan opnieuw gebruik worden gemaakt van een Pydantic model om verschillende Form velden eenmalig samen te definiëren.

main.py

from typing import Annotated
from fastapi import FastAPI, Form
from pydantic import BaseModel

app = FastAPI()

class FormData(BaseModel):
    username: str
    password: str

@app.post("/login/")
def login(data: Annotated[FormData, Form()]):
    return data

Error Responses

Om een fout (bvb. HTTP 404) terug te sturen naar de client kan een HTTPException exception geraised worden.

Deze heeft een verplicht status_code argument en optionele argumenten als detail. Omdat het een exception is kan dit rechtstreeks gebruikt worden in path operation functie, maar ook in onderliggende helper functies. We zagen hiervan al een voorbeeld in Custom Validatie.

main.py

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Item not found"
        )
    return {"item": items[item_id]}

Merk op dat FastAPI (nog) geen automatisch oplossing heeft om de extra 404 respons op te nemen in het OAS schema. Als we dit willen moet het handmatig gedefinieerd worden via de responses parameter in de path decorator.

main.py

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

class ErrorMessage(BaseModel):
    detail: str

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get(
    "/items/{item_id}",
    responses={
        status.HTTP_404_NOT_FOUND: {
            "description": "Item not found",
            "model": ErrorMessage,
        }
    },
)
def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Item not found"
        )
    return {"item": items[item_id]}

Extra Decorator Parameters

We zagen hierboven reeds the responses parameters. Er zijn nog een aantal andere handige parameters die in de decorator gebruikt kunnen worden.

• status_code voor de status code van een normale succesvolle respons.
• tags om operaties te groeperen.
• summary om een korte beschrijving van de operatie te geven.
• description om een uitgebreidere beschrijving te geven - maar dit kan beter via de functie doc-string.
• response_description voor de normale respons te beschrijven (i.p.v. default Successful response”).
• deprecated om aan te geven dat de operatie niet meer gebruikt moet worden.

main.py

from fastapi import FastAPI, status

app = FastAPI()

@app.get("/users/", summary="Get all users", tags=["users"])
async def read_users():
    """Get the list of all users:

    By default returns only the username field.
    """
    return [{"username": "johndoe"}]

@app.get("/items/", tags=["items"], response_description="A list of items")
async def read_items():
    return [{"name": "Foo", "price": 42}]

@app.get(
    "/elements/",
    tags=["items"],
    deprecated=True,
    status_code=status.HTTP_203_NON_AUTHORITATIVE_INFORMATION,
)
async def read_elements():
    return [{"item_id": "Foo"}]

JSON Encoder

Wanneer een path operation functie een Pydantic model returnt zorgt FastAPI automatisch voor een correcte omzetting naar JSON in de HTTP respons.

Diezelfde omzetting maakt FastAPI ook beschikbaar via de fastapi.encoders.jsonable_encoder functie. Dit kan erg handig zijn om een model naar een JSON-encodeerbare dict om te zetten. Bijvoorbeeld voor opslag in een JSON database zoals MongoDB.

In het volgende voorbeeld bevat het Pydantic model een datetime veld. Dit kan niet rechtstreeks worden omgezet naar JSON. De jsonable_encoder functie zal hier een ISO 8601 string van maken.

main.py

from datetime import datetime
from fastapi import FastAPI, HTTPException, status
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

fake_db = {}

class Item(BaseModel):
    name: str
    timestamp: datetime
    description: str | None = None

app = FastAPI()

@app.post("/items/", response_model=Item)
def create_item(item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[item.name.lower()] = json_compatible_item_data
    print(f"DEBUG: fake_db = {fake_db}")
    return item

@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: str):
    if item := fake_db.get(item_id):
        return item
    else:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Item not found"
        )

Updates en Partial Updates

Met de HTTP POST methode worden nieuwe API resources (zoals in het vorige voorbeeld: Items) aangemaakt. Een bestaande resource te updaten kan op twee verschillende manieren, met twee verschillende HTTP methods: PUT en PATCH.

We bekijken eerst PUT door het vorige voorbeeld iets uit te breiden. Met PUT wordt het volledige Item vervangen door nieuwe data. Probeer aan de hand van volgende code te bepalen welk probleem kan voorkomen bij het gebruik van PUT.

main.py

from datetime import datetime
from fastapi import FastAPI, HTTPException, status
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

fake_db = {
    "foo": {"name": "Foo", "timestamp": "2026-01-23T07:30:39.105Z", "price": 50.2},
    "bar": {
        "name": "Bar",
        "timestamp": "2026-01-23T07:32:29.421Z",
        "description": "The bartenders",
        "price": 62,
        "tax": 20.2,
    },
    "baz": {
        "name": "Baz",
        "timestamp": "2026-01-23T07:33:57.632Z",
        "description": None,
        "price": 50.2,
        "tax": 10.5,
        "tags": [],
    },
}

class Item(BaseModel):
    name: str
    timestamp: datetime = datetime.fromtimestamp(0)
    description: str | None = None
    price: float | None = None
    tax: float = 10.5
    tags: list[str] = []

app = FastAPI()

@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: str):
    if item := fake_db.get(item_id):
        return Item(**item)
    else:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Item not found"
        )

@app.put("/items/{item_id}", response_model=Item)
def update_item(item_id: str, item: Item):
    update_item_encoded = jsonable_encoder(item)
    fake_db[item_id] = update_item_encoded
    return update_item_encoded

Het probleem met PUT zoals hier geïmplementeerd is dat default waarden (bvb tax) ongewild de bestaande waarde van een Item kunnen overschrijven.

In een PATCH request worden in principe enkel de gewijzigde velden meegegeven. Pydantic voorziet twee handige functies waar we gebruik van kunnen maken om dit gedrag met FastAPI te implementeren.

• Met model_dump(exclude_unset=True) krijgen we van een bestaand model object een dict die enkel de velden bevat die expliciet werden meegegeven bij het aanmaken van het object. Op die manier krijgen kunnen we dus enkel de velden uit de PATCH request verkrijgen.
• Met model_copy(update=update_data) krijgen we een kopie van een model object maar worden de velden uit de update_data (dict) gebruikt om de bestaande waarden te overschrijven.

main.py

@app.patch("/items/{item_id}", response_model=Item)
async def update_item_partial(item_id: str, item: Item):
    stored_item_data = fake_db[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.model_dump(exclude_unset=True)
    updated_item = stored_item_model.model_copy(update=update_data)
    fake_db[item_id] = jsonable_encoder(updated_item)
    return updated_item

Oefening

Als oefening maken we een eenvoudige JSON API voor een ToDo/Task-management applicatie. Elke stap bouwt voort op de vorige. Probeer enkel gebruik te maken van de uitstekende FastAPI en Pydantic documentatie.

Een Task stelt een stuk werk voor, zoals:

• “Documentatie schrijven”
• “Bug fixen”
• “Presentatie voorbereiden”

Elke task heeft:

• Een ID
• Een titel met minimumlengte van 3 karakters
• Een optionele beschrijving
• Een prioriteit van 1 tot 5
• Een interne timestamp (created_at) die automatisch wordt ingesteld
  • Is enkel voor intern gebruik en wordt nooit teruggestuurd naar de client.

Stap 1 – Task aanmaken & opvragen

Maak een eerste werkende API met volgende mogelijkheden:

• Tasks aanmaken (POST /tasks)
  • De aangemaakte task wordt in de respons body teruggestuurd met status code 201.
• Één task opvragen (GET /tasks/{task_id})

Gebruik voor de ‘opslag’ van de tasks een globale dictionary als pseudo-database.

tasks = {
    1: {...},
    2: {...},
}

Tip

Om een task voor te stellen zul je 3 modellen nodig hebben en gebruik maken van overerving.

Bvb:

• TaskCreate: title, description, priority
• TaskExternal: TaskCreate + task_id
• TaskInternal: TaskExternal + created_at

Oplossing

main.py

from datetime import datetime, timezone
from fastapi import FastAPI, HTTPException, Path, status
from pydantic import BaseModel, Field
from typing_extensions import Annotated

app = FastAPI()

TASKS_DB: dict[int, "TaskInternal"] = {}

class TaskCreate(BaseModel):
    title: Annotated[str, Field(min_length=3)]
    description: str | None = None
    priority: Annotated[int, Field(ge=1, le=5)]

class TaskExternal(TaskCreate):
    task_id: Annotated[int, Field(serialization_alias="id")]

class TaskInternal(TaskExternal):
    created_at: Annotated[
        datetime, Field(default_factory=lambda: datetime.now(tz=timezone.utc))
    ]

@app.post("/tasks", response_model=TaskExternal, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate):
    next_task_id = max(TASKS_DB.keys(), default=0) + 1
    new_task = TaskInternal(task_id=next_task_id, **task.model_dump())
    TASKS_DB[next_task_id] = new_task

    return new_task


@app.get("/tasks/{task_id}", response_model=TaskExternal)
def get_task(task_id: Annotated[int, Path(ge=1)]):
    try:
        return TASKS_DB[task_id]
    except KeyError:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")

Stap 2 - Tasks filteren en updaten

Breid de code uit de vorige stap uit met twee extra API paths:

• Tasks oplijsten en filteren (GET /tasks)
  • Stuurt by default alle tasks terug in een lijst
  • Query parameter min_priority (int) beperkt de lijst tot tasks met ten minste deze priority
  • Query parameter q (str) beperkt de lijst tot tasks waarbij deze string in de titel voorkomt (substring match)
• Task aanpassen (PATCH /tasks/{task_id})
  • Past een of meerdere van de volgende velden aan: title, description, priority (partiële update)
  • Stuurt aangepaste task terug

Tip

Om een task aan de passen met PATCH zal een extra Pydantic model nodig zijn waarbij alle velden optioneel zijn.

Oplossing

main.py

from datetime import datetime, timezone
from fastapi import FastAPI, HTTPException, Path, Query, status
from pydantic import BaseModel, Field
from typing_extensions import Annotated

app = FastAPI()

TASKS_DB: dict[int, "TaskInternal"] = {}

class TaskCreate(BaseModel):
    title: Annotated[str, Field(min_length=3)]
    description: str | None = None
    priority: Annotated[int, Field(ge=1, le=5)]

class TaskExternal(TaskCreate):
    task_id: Annotated[int, Field(serialization_alias="id")]

class TaskInternal(TaskExternal):
    created_at: Annotated[
        datetime, Field(default_factory=lambda: datetime.now(tz=timezone.utc))
    ]

class TaskUpdate(TaskCreate):
    title: Annotated[str | None, Field(min_length=3)] = None
    priority: Annotated[int | None, Field(ge=1, le=5)] = None

TaskId = Annotated[int, Path(ge=1)]

@app.post("/tasks", response_model=TaskExternal, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate):
    next_task_id = max(TASKS_DB.keys(), default=0) + 1
    new_task = TaskInternal(task_id=next_task_id, **task.model_dump())

    TASKS_DB[next_task_id] = new_task

    return new_task

@app.get("/tasks/{task_id}", response_model=TaskExternal)
def get_task(task_id: TaskId):
    try:
        return TASKS_DB[task_id]
    except KeyError:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")

@app.get("/tasks", response_model=list[TaskExternal])
def list_tasks(
    min_priority: Annotated[int | None, Query(ge=1)] = None,
    query: Annotated[str | None, Query(alias="q", min_length=3)] = None,
):
    results = TASKS_DB.values()

    if min_priority is not None:
        results = [t for t in results if t.priority >= min_priority]

    if query:
        results = [t for t in results if query.lower() in t.title.lower()]

    return results

@app.patch("/tasks/{task_id}", response_model=TaskExternal)
def update_task(task_id: TaskId, task_data: TaskUpdate):
    stored_task = TASKS_DB.get(task_id)
    if not stored_task:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")

    update_data = task_data.model_dump(exclude_unset=True)
    updated_task = stored_task.model_copy(update=update_data)

    TASKS_DB[task_id] = updated_task
    return updated_task

Stap 3 - Voorkeur Cookie en afwerking

We willen via een nieuw POST /preferences API path de mogelijkheid bieden gebruikersvoorkeuren op de slaan. Initieel is er maar één voorkeur: een default minimum prioriteit die gebruikt moet worden bij het GET /tasks path.

In de POST /preferences request wordt een minimum prioriteit meegegeven. In de respons moet diezelfde prioriteit in een cookie worden opgeslagen.

Opmerking

Deze oefening is louter didactisch om de gebruik van cookies en request/response objecten te oefenen. Een echte API zal zelden cookies gebruiken om gebruikersvoorkeuren in de browser op de slaan.

In GET /tasks moet de minimum prioriteit uit de request cookie (indien aanwezig) worden gebruikt enkel wanneer de min_priority query parameter ontbreekt.

We gebruiken deze laatste stap ook om de API verder af te werken, bvb: - Voeg summary en tags toe in de path decorators - Voeg, waar nuttig, realistische voorbeelden (examples) toe in de modellen - Geef de API een titel (bvb Task Management API) en een 1.0.0 versie

Oplossing

main.py

from datetime import datetime, timezone
from fastapi import Cookie, FastAPI, HTTPException, Path, Query, Response, status
from pydantic import BaseModel, Field
from typing_extensions import Annotated

app = FastAPI(title="Task Management API", version="1.0.0")

TASKS_DB: dict[int, "TaskInternal"] = {}

class TaskCreate(BaseModel):
    title: Annotated[str, Field(min_length=3, examples=["Learn FastAPI"])]
    description: Annotated[str | None, Field(examples=["Use the FastAPI tutorial"])] = None
    priority: Annotated[int, Field(ge=1, le=5, examples=[4])]

class TaskExternal(TaskCreate):
    task_id: Annotated[int, Field(serialization_alias="id", examples=[1])]

class TaskInternal(TaskExternal):
    created_at: Annotated[datetime, Field(default_factory=lambda: datetime.now(tz=timezone.utc))]

class TaskUpdate(TaskCreate):
    title: Annotated[str | None, Field(min_length=3)] = None
    priority: Annotated[int | None, Field(ge=1, le=5)] = None

class Preference(BaseModel):
    min_priority: Annotated[int, Field(ge=1, le=5)]

TaskId = Annotated[int, Path(ge=1)]

@app.post(
    "/tasks",
    response_model=TaskExternal,
    status_code=status.HTTP_201_CREATED,
    tags=["Tasks"],
    summary="Create a new task",
)
def create_task(task: TaskCreate):
    next_task_id = max(TASKS_DB.keys(), default=0) + 1
    new_task = TaskInternal(task_id=next_task_id, **task.model_dump())

    TASKS_DB[next_task_id] = new_task

    return new_task

@app.get(
    "/tasks/{task_id}",
    response_model=TaskExternal,
    tags=["Tasks"],
    summary="Get a single task",
)
def get_task(task_id: TaskId):
    try:
        return TASKS_DB[task_id]
    except KeyError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
        )

@app.get(
    "/tasks", response_model=list[TaskExternal], tags=["Tasks"], summary="List tasks"
)
def list_tasks(
    min_priority: Annotated[int | None, Query(ge=1)] = None,
    query: Annotated[str | None, Query(alias="q", min_length=3)] = None,
    cookie_min_priority: Annotated[int | None, Cookie(alias="min_priority", include_in_schema=False)] = None,
):
    results = TASKS_DB.values()

    if min_priority is None and cookie_min_priority is not None:
        min_priority = cookie_min_priority

    if min_priority is not None:
        results = [t for t in results if t.priority >= min_priority]

    if query:
        results = [t for t in results if query.lower() in t.title.lower()]

    return results

@app.patch(
    "/tasks/{task_id}",
    response_model=TaskExternal,
    tags=["Tasks"],
    summary="Update a task",
)
def update_task(task_id: TaskId, task_data: TaskUpdate):
    stored_task = TASKS_DB.get(task_id)
    if not stored_task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
        )

    update_data = task_data.model_dump(exclude_unset=True)
    updated_task = stored_task.model_copy(update=update_data)

    TASKS_DB[task_id] = updated_task
    return updated_task

@app.post(
    "/preferences",
    status_code=status.HTTP_204_NO_CONTENT,
    tags=["Preferences"],
    summary="Set user preferences",
)
def set_preferences(pref: Preference, response: Response):
    response.set_cookie(
        key="min_priority",
        value=str(pref.min_priority),
    )

Testen

Met onderstaande code kan je zelf je API code testen. Daarvoor moet wel eerst pytest geïnstalleerd worden:

uv add pytest

Om te testen maken we gebruiken we hier de FastAPI TestClient. Het gebruik is heel gelijkaardig aan de Flask TestClient.

Er wordt van uitgegaan dat in dezelfde directory een main module (main.py) bestaat met daarin een FastAPI app object. Om de testen uit te voeren:

uv run pytest test.py

De volledige code voor de oefening en testen staan ook hier: https://github.com/dvx76/fastapi-tutorial/tree/main/les2/oefening1.

test.py

test.py

import pytest
from fastapi.testclient import TestClient

@pytest.fixture
def client():
    """Recreate app for every test to start with clean database."""
    import importlib
    import main
    importlib.reload(main)
    from main import app

    return TestClient(app)

@pytest.mark.parametrize("priority", [1, 2, 3, 4, 5])
def test_post_task_all_priorities(client: TestClient, priority: int):
    response = client.post(
        "/tasks",
        json={
            "title": "abc",
            "description": "test description",
            "priority": priority,
        },
    )
    assert response.status_code == 201
    assert response.json() == {
        "title": "abc",
        "description": "test description",
        "priority": priority,
        "id": 1,
    }

def test_post_task_no_description_and_id_increments(client: TestClient):
    response_1 = client.post("/tasks", json={"title": "title 1", "priority": 1})
    response_2 = client.post("/tasks", json={"title": "title 2", "priority": 1})
    assert response_2.status_code == 201
    assert response_1.json()["id"] == 1
    assert response_2.json()["id"] == 2

@pytest.mark.parametrize(
    "title, priority",
    [("", 1), ("te", 1), ("test", -1), ("test", 0), ("test", 6), ("test", "a")],
)
def test_post_task_validation_errors(client: TestClient, title: str, priority: int):
    response = client.post(
        "/tasks",
        json={
            "title": title,
            "priority": priority,
        },
    )
    assert response.status_code == 422

def test_get_task_succeess(client: TestClient):
    post_input = {
        "title": "test title",
        "description": "test description",
        "priority": 1,
    }
    client.post("/tasks", json=post_input)
    response = client.get("/tasks/1")
    assert response.status_code == 200
    expected_response = post_input | {"id": 1}
    assert response.json() == expected_response

def test_get_task_not_found(client: TestClient):
    response = client.get("/tasks/1")
    assert response.status_code == 404
    assert response.json()

def test_get_all_tasks_filtered(client: TestClient):
    for post in range(1, 4):
        client.post(
            "/tasks",
            json={
                "title": f"test title {post}",
                "description": f"test description {post}",
                "priority": post,
            },
        )

    # All tasks
    response = client.get("/tasks")
    assert response.status_code == 200
    assert len(response.json()) == 3

    # Only 2 tasks with priority >= 2
    response = client.get("/tasks", params={"min_priority": 2})
    assert response.status_code == 200
    assert len(response.json()) == 2

    # Only 1 task with priority >= 2 and 'title 2' in title
    response = client.get("/tasks", params={"min_priority": 2, "q": "title 2"})
    assert response.status_code == 200
    assert len(response.json()) == 1

    # Test min_priority read from cookies
    response = client.get("/tasks", cookies={"min_priority": "3"})
    assert response.status_code == 200
    assert len(response.json()) == 1

    # Test query param overrides preference from cookies
    response = client.get(
        "/tasks", params={"min_priority": 2}, cookies={"min_priority": "3"}
    )
    assert response.status_code == 200
    assert len(response.json()) == 2

def test_patch_task(client: TestClient):
    client.post(
        "/tasks",
        json={
            "title": "test title",
            "description": "test description",
            "priority": 3,
        },
    )
    response = client.patch("/tasks/1", json={"description": "updated"})
    assert response.status_code == 200
    assert response.json() == {
        "title": "test title",
        "description": "updated",
        "priority": 3,
        "id": 1,
    }

def test_patch_task_found(client: TestClient):
    response = client.patch("/tasks/1", json={"description": "updated"})
    assert response.status_code == 404


def test_preferences_set(client: TestClient):
    response = client.post("/preferences", json={"min_priority": 3})
    assert response.status_code == 204
    assert response.cookies == {"min_priority": "3"}

Links

Wil je verder leren over FastAPI en Pydantic, bekijk dan zeker eens volgende links:

• FastAPI Docs
• FastAPI Tutorial
• Pydantic Docs
• Corey Schafer’s “FastAPI Tutorial” (Playlist)
• Corey Schafer’s “Pydantic Tutorial”
• ArjanCodes’s “Anatomy of a Scalable FastAPI Project”