← Terug naar tutorials

REST API bouwen met Python en Flask (voor gevorderde beginners)

pythonflaskrest apibackend developmentjsonapi designauthenticatietesting

REST API bouwen met Python en Flask (voor gevorderde beginners)

Dit is een complete, praktijkgerichte tutorial om een REST API te bouwen met Python en Flask. Je leert niet alleen wat je moet doen, maar vooral waarom je het zo doet: projectstructuur, configuratie, validatie, foutafhandeling, versies, authenticatie, testen, logging en deployment-achtige aandachtspunten. Alle voorbeelden zijn uitvoerbaar en bevatten echte commando’s.

Inhoud

Doel en uitgangspunten

Je gaat een REST API bouwen die “taken” beheert. De API ondersteunt:

Belangrijke ontwerpkeuzes:

We gebruiken Flask omdat het lichtgewicht is en je veel controle geeft. Dit is ideaal om REST-concepten goed te begrijpen.

Benodigdheden

Controleer je Python-versie:

python --version

Project opzetten (virtuele omgeving, dependencies)

Maak een map en initialiseer een virtuele omgeving:

mkdir flask-rest-api-tutorial
cd flask-rest-api-tutorial

python -m venv .venv

Activeer de omgeving:

source .venv/bin/activate
.venv\Scripts\Activate.ps1

Installeer dependencies:

pip install Flask python-dotenv pytest gunicorn

Leg je dependencies vast:

pip freeze > requirements.txt

Waarom dit belangrijk is:

Projectstructuur voor groei

Maak deze structuur:

mkdir -p app/api app/core app/services app/storage tests
touch app/__init__.py app/api/__init__.py app/core/__init__.py app/services/__init__.py app/storage/__init__.py
touch app/api/routes.py app/core/config.py app/core/errors.py app/services/tasks_service.py app/storage/memory.py
touch wsgi.py .env

Aanbevolen structuur:

Waarom deze structuur:

Eerste Flask-app en healthcheck

Configuratie via .env

Vul .env:

FLASK_ENV=development
APP_NAME=TakenAPI
API_KEY=supergeheime-sleutel

Let op: zet echte secrets nooit in versiebeheer. In een echt project gebruik je een secrets manager of CI-variabelen.

app/core/config.py

import os

class Config:
    APP_NAME = os.getenv("APP_NAME", "FlaskAPI")
    API_KEY = os.getenv("API_KEY", "")
    JSON_SORT_KEYS = False

JSON_SORT_KEYS = False zorgt dat Flask JSON sleutels niet automatisch sorteert; dit maakt responses vaak leesbaarder.

app/__init__.py (app-factory)

from flask import Flask
from dotenv import load_dotenv

from app.core.config import Config
from app.core.errors import register_error_handlers
from app.api.routes import api_bp

def create_app():
    load_dotenv()

    app = Flask(__name__)
    app.config.from_object(Config)

    app.register_blueprint(api_bp, url_prefix="/api/v1")
    register_error_handlers(app)

    @app.get("/health")
    def health():
        return {"status": "ok", "app": app.config["APP_NAME"]}, 200

    return app

Waarom app-factory:

wsgi.py

from app import create_app

app = create_app()

Starten in development

Je kunt Flask direct draaien via een klein script, maar met deze structuur is het handig om Flask te wijzen op wsgi.py.

macOS/Linux:

export FLASK_APP=wsgi.py
flask run --port 5000

Windows (PowerShell):

$env:FLASK_APP="wsgi.py"
flask run --port 5000

Test de healthcheck:

curl -i http://127.0.0.1:5000/health

Je verwacht iets als:

REST-principes en HTTP-statuscodes

REST is geen “magische library”, maar een set afspraken:

Een goede API is voorspelbaar: dezelfde foutstructuur, duidelijke codes, en zo min mogelijk “verrassingen”.

Een voorbeeld-API: takenlijst

Een taak heeft:

We beginnen met in-memory opslag om de API te begrijpen. Daarna kun je eenvoudig overstappen op een database.

In-memory opslag en later uitbreiden

app/storage/memory.py

from __future__ import annotations
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Dict, List, Optional

def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()

@dataclass
class Task:
    id: int
    title: str
    done: bool
    created_at: str
    updated_at: str

class MemoryTaskRepository:
    def __init__(self):
        self._tasks: Dict[int, Task] = {}
        self._next_id: int = 1

    def list_tasks(self) -> List[dict]:
        return [asdict(t) for t in self._tasks.values()]

    def get_task(self, task_id: int) -> Optional[dict]:
        task = self._tasks.get(task_id)
        return asdict(task) if task else None

    def create_task(self, title: str) -> dict:
        task_id = self._next_id
        self._next_id += 1

        ts = now_iso()
        task = Task(
            id=task_id,
            title=title,
            done=False,
            created_at=ts,
            updated_at=ts,
        )
        self._tasks[task_id] = task
        return asdict(task)

    def update_task(self, task_id: int, *, title: Optional[str] = None, done: Optional[bool] = None) -> Optional[dict]:
        task = self._tasks.get(task_id)
        if not task:
            return None

        if title is not None:
            task.title = title
        if done is not None:
            task.done = done

        task.updated_at = now_iso()
        return asdict(task)

    def delete_task(self, task_id: int) -> bool:
        return self._tasks.pop(task_id, None) is not None

Waarom repository:

Routes bouwen: CRUD voor taken

We gebruiken een Flask Blueprint voor versie v1.

app/api/routes.py

from flask import Blueprint, current_app, request

from app.core.errors import ApiError
from app.services.tasks_service import TasksService
from app.storage.memory import MemoryTaskRepository

api_bp = Blueprint("api", __name__)

_repo = MemoryTaskRepository()
_service = TasksService(_repo)

def require_api_key():
    expected = current_app.config.get("API_KEY", "")
    provided = request.headers.get("X-API-Key", "")
    if not expected:
        # Als er geen sleutel is geconfigureerd, staan we alles toe (handig lokaal).
        return
    if provided != expected:
        raise ApiError(401, "unauthorized", "Ongeldige of ontbrekende API-sleutel")

@api_bp.get("/tasks")
def list_tasks():
    require_api_key()

    limit = request.args.get("limit", default="20")
    offset = request.args.get("offset", default="0")
    done = request.args.get("done")  # optioneel: "true"/"false"

    return _service.list_tasks(limit=limit, offset=offset, done=done), 200

@api_bp.post("/tasks")
def create_task():
    require_api_key()

    data = request.get_json(silent=True)
    if data is None:
        raise ApiError(400, "bad_request", "Request-body moet geldige JSON zijn")

    task = _service.create_task(data)
    return task, 201

@api_bp.get("/tasks/<int:task_id>")
def get_task(task_id: int):
    require_api_key()

    task = _service.get_task(task_id)
    return task, 200

@api_bp.patch("/tasks/<int:task_id>")
def patch_task(task_id: int):
    require_api_key()

    data = request.get_json(silent=True)
    if data is None:
        raise ApiError(400, "bad_request", "Request-body moet geldige JSON zijn")

    task = _service.patch_task(task_id, data)
    return task, 200

@api_bp.delete("/tasks/<int:task_id>")
def delete_task(task_id: int):
    require_api_key()

    _service.delete_task(task_id)
    return "", 204

Belangrijk:

Validatie en consistente fouten (JSON)

app/core/errors.py

from dataclasses import dataclass
from flask import Flask

@dataclass
class ApiError(Exception):
    status_code: int
    code: str
    message: str
    details: dict | None = None

def register_error_handlers(app: Flask) -> None:
    @app.errorhandler(ApiError)
    def handle_api_error(err: ApiError):
        payload = {
            "error": {
                "code": err.code,
                "message": err.message,
            }
        }
        if err.details:
            payload["error"]["details"] = err.details
        return payload, err.status_code

    @app.errorhandler(404)
    def handle_404(_):
        return {"error": {"code": "not_found", "message": "Resource niet gevonden"}}, 404

    @app.errorhandler(405)
    def handle_405(_):
        return {"error": {"code": "method_not_allowed", "message": "Methode niet toegestaan"}}, 405

    @app.errorhandler(500)
    def handle_500(_):
        # In productie wil je hier geen interne details lekken.
        return {"error": {"code": "internal_error", "message": "Interne serverfout"}}, 500

Waarom dit patroon:

Service-laag met validatie: app/services/tasks_service.py

from typing import Any, Optional

from app.core.errors import ApiError

class TasksService:
    def __init__(self, repo):
        self.repo = repo

    def _parse_int(self, value: str, field: str) -> int:
        try:
            return int(value)
        except (TypeError, ValueError):
            raise ApiError(400, "bad_request", f"'{field}' moet een geheel getal zijn")

    def _parse_bool_str(self, value: Optional[str], field: str) -> Optional[bool]:
        if value is None:
            return None
        v = value.strip().lower()
        if v in ("true", "1", "yes"):
            return True
        if v in ("false", "0", "no"):
            return False
        raise ApiError(400, "bad_request", f"'{field}' moet true of false zijn")

    def list_tasks(self, *, limit: str, offset: str, done: Optional[str]):
        limit_i = self._parse_int(limit, "limit")
        offset_i = self._parse_int(offset, "offset")
        if limit_i < 1 or limit_i > 200:
            raise ApiError(400, "bad_request", "'limit' moet tussen 1 en 200 liggen")
        if offset_i < 0:
            raise ApiError(400, "bad_request", "'offset' mag niet negatief zijn")

        done_b = self._parse_bool_str(done, "done")

        tasks = self.repo.list_tasks()

        if done_b is not None:
            tasks = [t for t in tasks if t["done"] is done_b]

        total = len(tasks)
        page = tasks[offset_i: offset_i + limit_i]

        return {
            "items": page,
            "pagination": {
                "limit": limit_i,
                "offset": offset_i,
                "total": total,
            },
        }

    def create_task(self, data: dict[str, Any]):
        title = data.get("title")
        if not isinstance(title, str) or not title.strip():
            raise ApiError(
                400,
                "validation_error",
                "Validatiefout",
                details={"title": "Titel is verplicht en moet een niet-lege tekst zijn"},
            )

        task = self.repo.create_task(title=title.strip())
        return task

    def get_task(self, task_id: int):
        task = self.repo.get_task(task_id)
        if not task:
            raise ApiError(404, "not_found", f"Taak met id={task_id} bestaat niet")
        return task

    def patch_task(self, task_id: int, data: dict[str, Any]):
        allowed = {"title", "done"}
        unknown = set(data.keys()) - allowed
        if unknown:
            raise ApiError(
                400,
                "validation_error",
                "Onbekende velden in request",
                details={"unknown_fields": sorted(list(unknown))},
            )

        title = data.get("title", None)
        done = data.get("done", None)

        if title is not None and (not isinstance(title, str) or not title.strip()):
            raise ApiError(
                400,
                "validation_error",
                "Validatiefout",
                details={"title": "Als je 'title' meestuurt, moet het een niet-lege tekst zijn"},
            )

        if done is not None and not isinstance(done, bool):
            raise ApiError(
                400,
                "validation_error",
                "Validatiefout",
                details={"done": "Als je 'done' meestuurt, moet het een boolean zijn"},
            )

        updated = self.repo.update_task(task_id, title=title.strip() if isinstance(title, str) else None, done=done)
        if not updated:
            raise ApiError(404, "not_found", f"Taak met id={task_id} bestaat niet")
        return updated

    def delete_task(self, task_id: int):
        ok = self.repo.delete_task(task_id)
        if not ok:
            raise ApiError(404, "not_found", f"Taak met id={task_id} bestaat niet")

Wat hier “gevorderd beginners”-waardig is:

Paginering, filtering en sortering

We hebben paginering en filtering al toegevoegd:

Voorbeeld:

curl -s -H "X-API-Key: supergeheime-sleutel" "http://127.0.0.1:5000/api/v1/tasks?limit=10&offset=0&done=false" | python -m json.tool

Sortering kun je toevoegen door bijvoorbeeld sort=created_at en order=asc|desc te accepteren. Een simpele aanpak is sorteren in Python op sleutel, maar bij grotere datasets wil je dit in de database doen. Het ontwerpprincipe blijft: de API biedt parameters, de implementatie kan later veranderen.

Versiebeheer van je API

We gebruiken al /api/v1. Dit is een pragmatische keuze:

Alternatieven:

Voor veel teams is een padversie (/v1) het meest duidelijk.

Authenticatie met API-sleutels (eenvoudig maar nuttig)

We hebben X-API-Key toegevoegd. Dit is geen volledige beveiliging (zeker niet voor eindgebruikers), maar wel nuttig voor:

Test zonder sleutel:

curl -i http://127.0.0.1:5000/api/v1/tasks

Je verwacht 401 met JSON-fout.

Test met sleutel:

curl -i -H "X-API-Key: supergeheime-sleutel" http://127.0.0.1:5000/api/v1/tasks

Belangrijke nuance: API-sleutels moeten over HTTPS worden verstuurd in echte omgevingen, anders kan iemand op het netwerk de sleutel meelezen.

Logging en observatie

Flask logt standaard requests, maar je wilt vaak extra context. Een eenvoudige stap is loggen wanneer taken worden aangemaakt of aangepast. Je kunt dit in de service-laag doen.

Pas app/services/tasks_service.py aan door logging toe te voegen:

import logging
logger = logging.getLogger(__name__)

En bijvoorbeeld in create_task:

logger.info("Taak aangemaakt: title=%s", title.strip())

Om loggingniveau te zetten kun je in create_app() configureren:

import logging
logging.basicConfig(level=logging.INFO)

Waarom logging belangrijk is:

Testen met pytest

Testen is waar veel beginners afhaken, maar het is juist bij API’s enorm waardevol. Je test:

Testconfiguratie

Maak tests/test_api.py:

import os
import pytest

from app import create_app

@pytest.fixture()
def client(monkeypatch):
    monkeypatch.setenv("API_KEY", "testkey")
    app = create_app()
    app.config["TESTING"] = True
    return app.test_client()

def test_health(client):
    res = client.get("/health")
    assert res.status_code == 200
    data = res.get_json()
    assert data["status"] == "ok"

def test_unauthorized_without_key(client):
    res = client.get("/api/v1/tasks")
    assert res.status_code == 401
    data = res.get_json()
    assert data["error"]["code"] == "unauthorized"

def test_create_and_get_task(client):
    res = client.post(
        "/api/v1/tasks",
        headers={"X-API-Key": "testkey"},
        json={"title": "Boodschappen doen"},
    )
    assert res.status_code == 201
    created = res.get_json()
    assert created["title"] == "Boodschappen doen"
    assert created["done"] is False
    task_id = created["id"]

    res2 = client.get(f"/api/v1/tasks/{task_id}", headers={"X-API-Key": "testkey"})
    assert res2.status_code == 200
    fetched = res2.get_json()
    assert fetched["id"] == task_id

def test_validation_error_on_empty_title(client):
    res = client.post(
        "/api/v1/tasks",
        headers={"X-API-Key": "testkey"},
        json={"title": "   "},
    )
    assert res.status_code == 400
    data = res.get_json()
    assert data["error"]["code"] == "validation_error"
    assert "title" in data["error"]["details"]

Draai tests:

pytest -q

Waarom dit werkt:

CORS en clients

Als je browser-gebaseerde clients hebt (bijvoorbeeld een webapp op een andere domeinnaam/poort), dan krijg je te maken met CORS. In Flask kun je dit oplossen met een extra dependency, maar omdat we het simpel houden, volstaat het conceptueel:

Als je dit echt nodig hebt, installeer dan een CORS-extensie en configureer strikt welke origins zijn toegestaan. Laat het niet “open” in productie.

Productie-achtige run met gunicorn

Flask’s ingebouwde server is bedoeld voor ontwikkeling. Voor een productie-achtige run gebruik je gunicorn.

Start gunicorn:

gunicorn -w 2 -b 0.0.0.0:8000 wsgi:app

Uitleg:

Test:

curl -i http://127.0.0.1:8000/health

Let op: in echte productie zet je een reverse proxy ervoor (bijvoorbeeld nginx) en gebruik je HTTPS.

Handmatige API-tests met curl (praktijk)

Taak aanmaken

curl -i \
  -H "Content-Type: application/json" \
  -H "X-API-Key: supergeheime-sleutel" \
  -d '{"title":"API tutorial afmaken"}' \
  http://127.0.0.1:5000/api/v1/tasks

Verwacht 201 Created en een JSON-body met id.

Takenlijst ophalen

curl -s \
  -H "X-API-Key: supergeheime-sleutel" \
  "http://127.0.0.1:5000/api/v1/tasks?limit=5&offset=0" | python -m json.tool

Taak markeren als gedaan

Vervang <id> door een echt id:

curl -i \
  -H "Content-Type: application/json" \
  -H "X-API-Key: supergeheime-sleutel" \
  -d '{"done": true}' \
  http://127.0.0.1:5000/api/v1/tasks/<id>

Taak verwijderen

curl -i \
  -H "X-API-Key: supergeheime-sleutel" \
  -X DELETE \
  http://127.0.0.1:5000/api/v1/tasks/<id>

Verwacht 204 No Content.

Veelgemaakte fouten en hoe je ze voorkomt

  1. HTML-foutpagina’s teruggeven

    • Oplossing: eigen error handlers en get_json(silent=True).

  2. Validatie verspreid over routes

    • Oplossing: service-laag met duidelijke regels.

  3. Geen versie in je API

    • Oplossing: /api/v1 vanaf het begin.

  4. Geen tests

    • Oplossing: begin met 3–5 kernscenario’s (health, unauthorized, create/get, validation).

  5. Geen consistente response-structuur

    • Oplossing: standaardiseer succes- en foutresponses.

Volgende stappen

Als je deze tutorial af hebt, kun je uitbreiden met:

Complete code-overzicht (controlelijst)

Als alles klopt, heb je minimaal deze bestanden:

Met deze basis heb je een REST API die niet alleen “werkt”, maar ook al rekening houdt met onderhoudbaarheid, uitbreidbaarheid en betrouwbaarheid.

Gerelateerde tutorials

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