← Terug naar tutorials

REST API’s bouwen met Python: van ontwerp tot implementatie

pythonrest apifastapiflaskopenapiauthenticatievalidatietestingapi designbackend development

REST API’s bouwen met Python: van ontwerp tot implementatie

Een REST API bouwen is meer dan “een endpoint maken dat JSON teruggeeft”. Een goede API is voorspelbaar, veilig, goed te testen, makkelijk te onderhouden en duidelijk voor gebruikers. In deze tutorial doorlopen we het volledige traject: van ontwerpkeuzes en datamodellering tot implementatie met Python, validatie, authenticatie, versiebeheer, documentatie, testen, logging en deployment.

We gebruiken FastAPI omdat het modern is, snelle feedback geeft via automatische documentatie, en sterk is in type hints en validatie. De concepten (resources, HTTP-methodes, statuscodes, idempotentie, paginatie) zijn echter algemeen en gelden ook voor Flask, Django REST Framework of andere frameworks.

Inhoud

  1. Wat is REST (en wat niet)
  2. API-ontwerp: resources, routes en conventies
  3. HTTP-methodes, statuscodes en idempotentie
  4. Datamodellen en validatie met Pydantic
  5. Projectstructuur en afhankelijkheden
  6. Implementatie met FastAPI
  7. Database-laag met SQLAlchemy en migraties
  8. CRUD-endpoints bouwen
  9. Foutenafhandeling en consistente responses
  10. Authenticatie en autorisatie (JWT)
  11. Paginatie, filtering en sortering
  12. Versiebeheer van je API
  13. OpenAPI-documentatie en voorbeelden
  14. Testen met pytest
  15. Logging, configuratie en omgevingsvariabelen
  16. Deployment met Docker
  17. Checklist voor productie

Wat is REST (en wat niet)

REST is een architectuurstijl voor netwerkapplicaties. In de praktijk betekent dit meestal:

Wat REST niet automatisch betekent:

API-ontwerp: resources, routes en conventies

Resources kiezen

Begin met het identificeren van zelfstandige entiteiten in je domein. Stel: we bouwen een eenvoudige taken-app.

Mogelijke resources:

Een task heeft bijvoorbeeld:

URL-conventies

Gebruik zelfstandige naamwoorden (meervoud) en hiërarchie waar logisch:

Vermijd werkwoorden in paden zoals /createTask of /doLogin. Acties worden uitgedrukt via HTTP-methodes of via subresources als het echt nodig is.

Representaties

Bepaal hoe je JSON eruitziet. Houd het consistent:

Voorbeeld van een task response:

{
  "id": 42,
  "title": "Boodschappen doen",
  "description": "Melk, brood, groenten",
  "done": false,
  "created_at": "2026-02-14T10:15:00Z",
  "owner_id": 7
}

HTTP-methodes, statuscodes en idempotentie

Methodes

Statuscodes (veelgebruikte set)

Idempotentie in de praktijk

Idempotentie is belangrijk voor retries. Als een client een request opnieuw verstuurt door timeouts, wil je geen dubbele records.

Datamodellen en validatie met Pydantic

FastAPI gebruikt Pydantic voor:

Je definieert aparte modellen voor:

Belangrijk: scheid je API-schema’s van je database-modellen. Dit voorkomt dat databasewijzigingen je API breken en maakt security eenvoudiger (bijv. geen wachtwoordhash terugsturen).

Projectstructuur en afhankelijkheden

We bouwen een minimalistische maar schaalbare structuur.

Vereisten

Project aanmaken

mkdir tasks-api
cd tasks-api
python -m venv .venv
source .venv/bin/activate

Installeer dependencies:

pip install fastapi uvicorn sqlalchemy alembic pydantic python-jose passlib[bcrypt] python-multipart pytest httpx

Maak mappen:

mkdir -p app/api app/core app/db app/models app/schemas app/services tests
touch app/__init__.py

Een mogelijke structuur:

app/
  api/
    routes.py
    deps.py
  core/
    config.py
    security.py
    logging.py
  db/
    session.py
    base.py
  models/
    task.py
    user.py
  schemas/
    task.py
    user.py
    token.py
  services/
    task_service.py
    user_service.py
main.py
tests/
  test_tasks.py

Implementatie met FastAPI

main.py

from fastapi import FastAPI
from app.api.routes import router as api_router

app = FastAPI(
    title="Tasks API",
    version="1.0.0",
)

app.include_router(api_router, prefix="/api")

Router: app/api/routes.py

from fastapi import APIRouter
from app.api import tasks, auth

router = APIRouter()
router.include_router(auth.router, prefix="/auth", tags=["auth"])
router.include_router(tasks.router, prefix="/tasks", tags=["tasks"])

We moeten nu app/api/tasks.py en app/api/auth.py maken (zie verderop).

Start de server:

uvicorn main:app --reload --port 8000

Open:

Database-laag met SQLAlchemy en migraties

Database sessie: app/db/session.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},  # nodig voor SQLite + threads
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base declaratie: app/db/base.py

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

Modellen

app/models/user.py

from sqlalchemy import String, Integer
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String, nullable=False)

app/models/task.py

from datetime import datetime, timezone
from sqlalchemy import String, Integer, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base

class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String, nullable=False)
    description: Mapped[str] = mapped_column(String, nullable=True)
    done: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        default=lambda: datetime.now(timezone.utc),
    )
    owner_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False)

Tabellen aanmaken (snelle start)

Voor een snelle demo kun je tabellen aanmaken bij start. In productie gebruik je migraties.

Maak app/db/init_db.py:

from app.db.base import Base
from app.db.session import engine
from app.models.user import User
from app.models.task import Task

def init_db() -> None:
    Base.metadata.create_all(bind=engine)

Pas main.py aan:

from fastapi import FastAPI
from app.api.routes import router as api_router
from app.db.init_db import init_db

app = FastAPI(title="Tasks API", version="1.0.0")
app.include_router(api_router, prefix="/api")

@app.on_event("startup")
def on_startup() -> None:
    init_db()

CRUD-endpoints bouwen

Pydantic schema’s

app/schemas/task.py

from datetime import datetime
from pydantic import BaseModel, Field

class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=2000)

class TaskUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=2000)
    done: bool | None = None

class TaskRead(BaseModel):
    id: int
    title: str
    description: str | None
    done: bool
    created_at: datetime
    owner_id: int

    class Config:
        from_attributes = True

Dependency voor DB-sessie

app/api/deps.py:

from collections.abc import Generator
from app.db.session import SessionLocal

def get_db() -> Generator:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Service-laag (optioneel maar nuttig)

Door database-logica buiten je routes te houden, worden je endpoints dunner en beter testbaar.

app/services/task_service.py:

from sqlalchemy.orm import Session
from sqlalchemy import select
from app.models.task import Task
from app.schemas.task import TaskCreate, TaskUpdate

def create_task(db: Session, owner_id: int, data: TaskCreate) -> Task:
    task = Task(
        title=data.title,
        description=data.description,
        done=False,
        owner_id=owner_id,
    )
    db.add(task)
    db.commit()
    db.refresh(task)
    return task

def get_task(db: Session, task_id: int, owner_id: int) -> Task | None:
    stmt = select(Task).where(Task.id == task_id, Task.owner_id == owner_id)
    return db.scalar(stmt)

def list_tasks(db: Session, owner_id: int, limit: int, offset: int) -> list[Task]:
    stmt = select(Task).where(Task.owner_id == owner_id).limit(limit).offset(offset)
    return list(db.scalars(stmt).all())

def update_task(db: Session, task: Task, data: TaskUpdate) -> Task:
    if data.title is not None:
        task.title = data.title
    if data.description is not None:
        task.description = data.description
    if data.done is not None:
        task.done = data.done
    db.commit()
    db.refresh(task)
    return task

def delete_task(db: Session, task: Task) -> None:
    db.delete(task)
    db.commit()

Foutenafhandeling en consistente responses

Een veelgemaakte fout is willekeurige error-JSON terugsturen. Beter is:

FastAPI heeft HTTPException. Voorbeeld:

Authenticatie en autorisatie (JWT)

We implementeren:

Security helpers

app/core/security.py:

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

JWT_SECRET = "vervang-dit-door-een-sterk-geheim"
JWT_ALG = "HS256"
JWT_EXPIRE_MINUTES = 60

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(password: str, password_hash: str) -> bool:
    return pwd_context.verify(password, password_hash)

def create_access_token(subject: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": subject,
        "iat": int(now.timestamp()),
        "exp": int((now + timedelta(minutes=JWT_EXPIRE_MINUTES)).timestamp()),
    }
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALG)

def decode_token(token: str) -> str:
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG])
        sub = payload.get("sub")
        if not sub:
            raise ValueError("Token mist subject")
        return str(sub)
    except (JWTError, ValueError) as exc:
        raise ValueError("Ongeldig token") from exc

User schema’s

app/schemas/user.py:

from pydantic import BaseModel, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8, max_length=200)

class UserRead(BaseModel):
    id: int
    email: EmailStr

    class Config:
        from_attributes = True

app/schemas/token.py:

from pydantic import BaseModel

class TokenRead(BaseModel):
    access_token: str
    token_type: str = "bearer"

User service

app/services/user_service.py:

from sqlalchemy.orm import Session
from sqlalchemy import select
from app.models.user import User
from app.core.security import hash_password, verify_password

def get_user_by_email(db: Session, email: str) -> User | None:
    stmt = select(User).where(User.email == email)
    return db.scalar(stmt)

def create_user(db: Session, email: str, password: str) -> User:
    user = User(email=email, password_hash=hash_password(password))
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

def authenticate(db: Session, email: str, password: str) -> User | None:
    user = get_user_by_email(db, email)
    if not user:
        return None
    if not verify_password(password, user.password_hash):
        return None
    return user

Auth routes

app/api/auth.py:

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from app.api.deps import get_db
from app.schemas.user import UserCreate, UserRead
from app.schemas.token import TokenRead
from app.services.user_service import create_user, get_user_by_email, authenticate
from app.core.security import create_access_token

router = APIRouter()

@router.post("/register", response_model=UserRead, status_code=status.HTTP_201_CREATED)
def register(data: UserCreate, db: Session = Depends(get_db)) -> UserRead:
    existing = get_user_by_email(db, data.email)
    if existing:
        raise HTTPException(status_code=409, detail="E-mail is al geregistreerd")
    user = create_user(db, data.email, data.password)
    return user

@router.post("/token", response_model=TokenRead)
def login(form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)) -> TokenRead:
    user = authenticate(db, form.username, form.password)
    if not user:
        raise HTTPException(status_code=401, detail="Onjuiste inloggegevens")
    token = create_access_token(subject=str(user.id))
    return TokenRead(access_token=token)

Let op: OAuth2PasswordRequestForm verwacht application/x-www-form-urlencoded met velden username en password. Dat is standaard voor token endpoints.

Huidige gebruiker dependency

app/api/deps_auth.py:

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from sqlalchemy import select

from app.api.deps import get_db
from app.core.security import decode_token
from app.models.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token")

def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)) -> User:
    try:
        user_id = int(decode_token(token))
    except ValueError:
        raise HTTPException(status_code=401, detail="Ongeldig token")

    stmt = select(User).where(User.id == user_id)
    user = db.scalar(stmt)
    if not user:
        raise HTTPException(status_code=401, detail="Gebruiker niet gevonden")
    return user

Tasks routes (beveiligd)

app/api/tasks.py:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.api.deps import get_db
from app.api.deps_auth import get_current_user
from app.models.user import User
from app.schemas.task import TaskCreate, TaskRead, TaskUpdate
from app.services.task_service import create_task, get_task, list_tasks, update_task, delete_task

router = APIRouter()

@router.get("", response_model=list[TaskRead])
def get_tasks(
    limit: int = 50,
    offset: int = 0,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user),
) -> list[TaskRead]:
    limit = min(max(limit, 1), 200)
    offset = max(offset, 0)
    return list_tasks(db, owner_id=user.id, limit=limit, offset=offset)

@router.post("", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_new_task(
    data: TaskCreate,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user),
) -> TaskRead:
    return create_task(db, owner_id=user.id, data=data)

@router.get("/{task_id}", response_model=TaskRead)
def get_one_task(
    task_id: int,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user),
) -> TaskRead:
    task = get_task(db, task_id=task_id, owner_id=user.id)
    if not task:
        raise HTTPException(status_code=404, detail="Taak niet gevonden")
    return task

@router.patch("/{task_id}", response_model=TaskRead)
def patch_task(
    task_id: int,
    data: TaskUpdate,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user),
) -> TaskRead:
    task = get_task(db, task_id=task_id, owner_id=user.id)
    if not task:
        raise HTTPException(status_code=404, detail="Taak niet gevonden")
    return update_task(db, task, data)

@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def remove_task(
    task_id: int,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user),
) -> None:
    task = get_task(db, task_id=task_id, owner_id=user.id)
    if not task:
        raise HTTPException(status_code=404, detail="Taak niet gevonden")
    delete_task(db, task)
    return None

Handmatige tests met echte commando’s

Registreren

curl -i -X POST "http://127.0.0.1:8000/api/auth/register" \
  -H "Content-Type: application/json" \
  -d '{"email":"alice@example.com","password":"supergeheim123"}'

Token ophalen

curl -i -X POST "http://127.0.0.1:8000/api/auth/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "username=alice@example.com&password=supergeheim123"

Je krijgt JSON met access_token. Zet die in een shell-variabele:

TOKEN="plak-hier-je-token"

Taak aanmaken

curl -i -X POST "http://127.0.0.1:8000/api/tasks" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"title":"Boodschappen doen","description":"Melk en brood"}'

Takenlijst ophalen

curl -i "http://127.0.0.1:8000/api/tasks?limit=20&offset=0" \
  -H "Authorization: Bearer $TOKEN"

Taak bijwerken

curl -i -X PATCH "http://127.0.0.1:8000/api/tasks/1" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"done":true}'

Taak verwijderen

curl -i -X DELETE "http://127.0.0.1:8000/api/tasks/1" \
  -H "Authorization: Bearer $TOKEN"

Paginatie, filtering en sortering

Zodra je dataset groeit, wil je niet “alles” teruggeven. We gebruikten al limit en offset. Voor grote tabellen is offset soms traag; dan is keyset pagination beter (bijv. “geef items met id > X”). Voor nu is limit/offset prima.

Filtering

Je kunt queryparameters toevoegen zoals:

Je service-laag kan conditioneel filters toevoegen. Conceptueel:

Belangrijk: valideer en whitelist sorteer-velden om SQL-injectie via veldnamen te voorkomen.

Versiebeheer van je API

Er zijn meerdere strategieën:

  1. URL-versies: /api/v1/tasks
  2. Header-versies: Accept: application/vnd...
  3. Query-versies: ?version=1 (minder gebruikelijk)

URL-versies zijn het meest praktisch. Je kunt routers onderbrengen in app/api/v1/... en later v2 toevoegen.

Voorbeeld:

Versies zijn vooral nuttig als je breaking changes verwacht. Kleine uitbreidingen (extra velden) kunnen vaak binnen dezelfde versie zolang je backward compatible blijft.

OpenAPI-documentatie en voorbeelden

FastAPI genereert OpenAPI automatisch, maar je kunt het verbeteren met:

Voorbeeld van extra metadata:

@router.post(
    "",
    response_model=TaskRead,
    status_code=201,
    summary="Maak een nieuwe taak aan",
    description="Maakt een taak aan voor de ingelogde gebruiker.",
)
def create_new_task(...):
    ...

Zorg dat je documentatie klopt met je echte gedrag (statuscodes, foutmeldingen). Automatische docs zijn geen excuus om niet na te denken over API-contracten.

Testen met pytest

Testen van API’s is het effectiefst op HTTP-niveau: je test routes, dependencies, validatie en auth in één.

Maak tests/test_tasks.py (vereenvoudigd voorbeeld). Let op: voor nette tests wil je een aparte testdatabase en dependency overrides. Hieronder een basispatroon met TestClient.

Installeer testclient dependency (FastAPI gebruikt Starlette):

pip install pytest

Voorbeeldtest (conceptueel; voor echte isolatie moet je DB override toevoegen):

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_register_and_create_task():
    r = client.post("/api/auth/register", json={"email": "bob@example.com", "password": "wachtwoord123"})
    assert r.status_code in (201, 409)

    token_resp = client.post(
        "/api/auth/token",
        data={"username": "bob@example.com", "password": "wachtwoord123"},
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    assert token_resp.status_code == 200
    token = token_resp.json()["access_token"]

    task_resp = client.post(
        "/api/tasks",
        json={"title": "Test taak", "description": "Via pytest"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert task_resp.status_code == 201
    body = task_resp.json()
    assert body["title"] == "Test taak"
    assert body["done"] is False

Voor productiekwaliteit tests:

Logging, configuratie en omgevingsvariabelen

Hardcoded secrets zoals JWT_SECRET zijn onveilig. Gebruik omgevingsvariabelen.

Een eenvoudige configuratie: app/core/config.py

import os

def get_env(name: str, default: str | None = None) -> str:
    value = os.getenv(name, default)
    if value is None:
        raise RuntimeError(f"Ontbrekende omgevingsvariabele: {name}")
    return value

JWT_SECRET = get_env("JWT_SECRET", "dev-only-vervang-dit")
DATABASE_URL = get_env("DATABASE_URL", "sqlite:///./app.db")

Pas dan session.py en security.py aan om deze waarden te gebruiken.

Logging: begin simpel en structureer later. Uvicorn logt al requests, maar applicatielogs zijn ook nuttig:

Deployment met Docker

Docker maakt je runtime reproduceerbaar.

Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt /app/requirements.txt
RUN pip install --no-cache-dir -r /app/requirements.txt

COPY . /app

ENV PYTHONUNBUFFERED=1

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Maak requirements.txt:

pip freeze > requirements.txt

Build en run:

docker build -t tasks-api:latest .
docker run --rm -p 8000:8000 -e JWT_SECRET="een-sterk-geheim" tasks-api:latest

Voor productie:

Checklist voor productie

API-contract en ontwerp

Security

Database

Observability

Tests

Volgende stappen

Als je deze basis hebt staan, zijn logische uitbreidingen:

Als je wil, kan ik je project omzetten naar een volledige v1-structuur met Alembic migraties en PostgreSQL, inclusief dependency overrides voor nette tests.

Gerelateerde tutorials

Aan de slag met Docker: beginnersgids Aan De Slag Met Docker Applicaties deployen met Docker Compose: geavanceerde best practices