FastAPI: Een framework voor High-Performance API’s in Python

by

Wat is FastAPI?

FastAPI is een modern en goed presterend framework voor het bouwen van API’s in Python. Het is flexibel en gemakkelijk in gebruik. Qua performance is het gelijk aan Node.js en Go terwijl het het gebruiksgemak heeft van Python.

Het is gebouwd rondom Starlette (routering en async functionaliteit) en Pydantic (data validatie). Dit maakt het bouwen van robuuste oplossingen gemakkelijker.

FastAPI groeit daarom flink in populariteit en wordt bij veel bedrijven ingezet.

Klinkt goed, maar wat kan ik er mee?

Met behulp van dit framework kunnen koppelingen worden gebouwd tussen verschillende systemen. Het kan worden gebruikt te koppelen tussen systemen voor data uitwisseling en aansturing.

Voorbeelden van uiteindelijke oplossingen zijn:

  • Een interface voor team overstijgende automatisering.
  • Een interface om te voorzien in self service.
  • Als portaal om data te leveren wat wordt gebruikt voor rapportage / compliancy.

Voorbeelden van implementaties

Ik heb voor diverse bedrijven FastAPI applicaties mogen ontwikkelen, voorbeelden zijn:

  • Monitoring as Code orchestrator, single source of truth, die een monitoring platform cross tooling inricht en onderhoudt (desired state).
  • Diverse workers voor het configureren en bewaking op configuration drift voor Grafana en Zabbix implementaties.
  • Middleware tussen API’s en reporting tooling, zo blijft de reporting “snappy” en worden backend API’s niet teveel belast.

Automatische API documentatie

FastAPI voorziet in automatische documentatie van de API’s die je gebouwd hebt met behulp twee documentatie interfaces:

Swagger UI (/docs) – Hier vind je interactieve documentatie van de API. Je kunt de API hier ook testen om te zien wat de resultaten zijn voor een bepaalde query.

ReDoc (/redoc) – Vergelijkbaar met de Swagger UI alleen dan met een modernere UI met 3 panels.

Validatie van input

Door data te valideren wanneer het wordt ingegeven voorkom je problemen verderop in een process. Als je bevoorbeeld verwacht dat een integer wordt ingeven, dan kun je dit valideren.

  • De waarde wordt gecontroleerd, is dit echt een integer.
  • Zal een ingegeven numeriek string, als het een geheel getal is, omzetten naar een integer.
  • Geeft een duidelijke error melding wanneer de validatie mislukt.
  • Geeft de validatie argumenten weer in de API documentatie

Ondersteuning voor Asynchroniteit

Als de FastAPI is gebouwd op een ASGI server (Asychronous Server Gateway Interface) zoals uvicorn, dan geeft dit het ondersteuning voor async/await. Dit maakt het mogelijk om duizenden connecties tegelijkertijd te behandelen, in plaats van een voor een.

Performance

Bovenstaande en meer zorgt ervoor dat FastAPI erg goed presteert, en maakt het mogelijk om vrij gemakkelijk schaalbare applicaties te maken.

Bouw een eigen FastAPI applicatie

De minimale stappen voor het starten van een FastAPI applicatie zijn als volgt:

Installeer FastAPI en uvicorn:

pip install fastapi uvicorn[standard]

Een minimale FastAPI applicatie:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "query_param": q}

Uitleg bij de code

  • Pad-parameters: De syntax {item_id} in het URL-pad @app.get("/items/{item_id}") correspondeert direct met het functieargument item_id.
  • Typevalidatie: Door item_id: int te definiëren, gebruikt FastAPI Python type hints om data automatisch te valideren. Als een gebruiker /items/foo bezoekt, retourneert FastAPI een foutmelding omdat “foo” geen geheel getal (integer) is.
  • Query-parameters: Functieargumenten die niet expliciet in het pad zijn gedeclareerd (zoals q), worden automatisch geïnterpreteerd als query-parameters (bijvoorbeeld /items/5?q=zoekopdracht).
  • Optionele waarden: Door de standaardwaarde op None in te stellen, wordt de parameter optioneel.

Uitvoeren

Start de applicatie daarna met:

uvicorn main:app --reload

Open nu de volgende URL om de API documentatie te zien: http://localhost:8000/docs

Bouw een API voor productie

Een duidelijk datamodel is erg belangrijk voor het bouwen van een API voor productie. Dit zorgt ervoor dat de applicatie beheersbaar een efficient blijft. Dit voorkomt ook veel rework achteraf. Neem dus de tijd voor het ontwerp, dit verdien je later terug.

De modellen die je moet uitwerken zijn die voor verzoek (request) en antwoord (response).

from pydantic import BaseModel, EmailStr, Field, ConfigDict
from datetime import datetime

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    created_at: datetime
    
    # De nieuwe syntax voor V2
    model_config = ConfigDict(from_attributes=True)

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    # Logica...
    return new_user

Dit response model definieert responses met alleen specifieke velden, het geeft bijvoorbeeld niet het wachtwoord terug.

Authenticatie

Authenticatie, of inlogfunctionaliteit, is uiteraard erg belangrijk. Dit beschermd je applicatie tegen ongeoorloofde toegang.

In plaats van in elke functie te checken of een gebruiker is ingelogd, gebruik je Depends. FastAPI regelt dan dat de functie get_current_user eerst wordt uitgevoerd. Als die faalt (geen token), wordt de hoofd-functie nooit aangeroepen.

Dit model scheidt beveiliging van logica.

Dit is een voorbeeld van een authenticatieflow in FastAPI:

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

# 1. Define the Security Scheme (The 'Lock')
# This tells FastAPI that the client must send a Bearer token in the 'Authorization' header.
# The 'tokenUrl' parameter points to the endpoint where the client can obtain a token.
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 2. The User Model
class User(BaseModel):
    username: str
    email: str | None = None
    disabled: bool | None = None

# 3. The Dependency (The 'Key Check')
# This function is automatically called by any endpoint that requires security.
# It receives the token from the request header and validates it.
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    # Simulate token decoding/validation (in production, use a JWT library like python-jose)
    if token != "geheim-token-123":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # If the token is valid, return the user object (this becomes the dependency value)
    return User(username="johndoe", email="john@example.com", disabled=False)

# 4. Login Endpoint (Issues the token)
# This endpoint receives a standard OAuth2 form (username & password).
@app.post("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    # Here you would verify the username/password against your database (hashing, etc.)
    if form_data.username == "johndoe" and form_data.password == "secret":
        return {"access_token": "geheim-token-123", "token_type": "bearer"}
    
    # If credentials are wrong, return a 400 error
    raise HTTPException(status_code=400, detail="Incorrect username or password")

# 5. Protected Endpoint (Uses the dependency)
# This route requires the 'get_current_user' dependency to succeed.
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    # This code runs ONLY if the token was valid.
    # You have direct access to the validated 'current_user' object here.
    return current_user

Foutafhandeling

In een professionele API wil je niet overal try/except blokken in je endpoints hebben. Het is schoner om Custom Exceptions te definiëren en deze op één plek (globaal) af te vangen.

Dit houdt je endpoints leesbaar: ze bevatten alleen de “Happy Path”.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 1. Define a Custom Exception
# This is a custom error class inheriting from Python's standard Exception.
# We can raise this specific error anywhere in our business logic.
class ThingException(Exception):
    def __init__(self, name: str):
        self.name = name

# 2. Register the Global Exception Handler
# This tells FastAPI: "Whenever 'ThingException' is raised anywhere in the application,
# catch it and execute this function instead of crashing."
@app.exception_handler(ThingException)
async def thing_exception_handler(request: Request, exc: ThingException):
    return JSONResponse(
        status_code=418,  # I'm a teapot (using 418 as a custom example)
        content={
            "message": f"ERROR: The thing '{exc.name}' caused an issue.",
            "error_code": "THING_ESCAPE_ERROR"
        },
    )

# 3. The Endpoint (Business Logic)
@app.get("/thing/{name}")
async def read_thing(name: str):
    if name == "yolo":
        # We do not need to construct a JSONResponse here manually.
        # We simply raise the exception, and the global handler defined above intercepts it.
        # This keeps the main logic clean and focused on the "Happy Path".
        raise ThingException(name=name)
    
    return {"thing_name": name}

Achtergrondtaken

Het is mogelijk om achtergrondtaken te laten uitvoeren door de FastAPI applicatie. Dit is nuttig voor langlopende taken of onderhoud binnen de applicatie, bijvoorbeeld voor het versturen van een rapportage of het verwijderen van records ouder dan x dagen.

Taak op gestart n.a.v. een request (fire and forget)

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

# 1. The Task (a standard Python function)
# This function defines the work to be done. 
# It will execute *after* the HTTP response has been sent to the client.
def write_log_file(message: str):
    with open("log.txt", "a") as f:
        f.write(f"LOG: {message}\n")

# 2. The Endpoint
@app.post("/simple/")
async def run_task(background_tasks: BackgroundTasks):
    # 3. Schedule the task
    # We add the function 'write_log_file' to the queue with the specific argument.
    # This is non-blocking; the code continues immediately.
    background_tasks.add_task(write_log_file, "Button was pressed!")
    
    # 4. Return an immediate response to the user
    # The user gets a 200 OK instantly. The server starts processing the 
    # background task only after this response is delivered.
    return {"message": "Task has been added to the queue"}

Geplande taak

from fastapi import FastAPI
from contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOScheduler

# 1. Initialize the scheduler
scheduler = AsyncIOScheduler()

# 2. Define the task (a function)
def daily_cleanup():
    print("Time for cleanup.")

# 3. The Lifespan (Startup & Shutdown)
# This block manages what happens when the app starts and shuts down
@asynccontextmanager
async def lifespan(app: FastAPI):
    # A. Schedule the job (interval, cron, or date)
    # Here we schedule the task to run once every day
    scheduler.add_job(daily_cleanup, 'interval', days=1)
    
    # B. Start the scheduler
    scheduler.start()
    print("Scheduler started...")
    
    yield  # The application runs here...
    
    # C. Stop the scheduler on shutdown
    scheduler.shutdown()
    print("Scheduler stopped.")

# 4. Attach lifespan to the app
app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    return {"message": "The app is running, check the console"}

SQLAlchemy en ORM

Hier is een voorbeeld van SQLAlchemy in een FastAPI-applicatie.

Dit voorbeeld een volledige implementatie van een database-operatie:

  1. Engine: De verbinding met de database.
  2. Model: De Python-klasse die de database-tabel voorstelt.
  3. Schema: De Pydantic-klasse voor validatie (zoals eerder besproken).
  4. Session: Het verzoek.

Ik gebruik hier sqlite zodat je het lokaal direct kunt draaien zonder installatie van een server, in productie zal je een database als MySQL of PostgreSQL gebruiken.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, Session, declarative_base
from pydantic import BaseModel

# ==========================================
# 1. DATABASE CONFIGURATION
# ==========================================
# We use SQLite for this example (a file named 'sql_app.db')
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# The Engine is the entry point to the database
# connect_args is only needed for SQLite
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# SessionLocal is a factory to create new database sessions
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class for our ORM models
Base = declarative_base()

# ==========================================
# 2. SQLALCHEMY MODEL (The Database Table)
# ==========================================
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Integer, default=True)

# Create the tables in the database (usually done via migrations in prod)
Base.metadata.create_all(bind=engine)

# ==========================================
# 3. PYDANTIC SCHEMAS (Validation)
# ==========================================
class UserCreate(BaseModel):
    email: str
    password: str

class UserResponse(BaseModel):
    id: int
    email: str
    is_active: bool

    class Config:
        orm_mode = True  # Allows Pydantic to read data from the ORM object

# ==========================================
# 4. DEPENDENCY (Database Session Management)
# ==========================================
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

# ==========================================
# 5. CRUD OPERATIONS
# ==========================================
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # Check if user already exists
    # We query the User model directly
    db_user = db.query(User).filter(User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")

    # Create the ORM Instance
    fake_hashed_password = user.password + "notreallyhashed"
    new_user = User(email=user.email, hashed_password=fake_hashed_password)

    # Add to session, commit transaction, and refresh to get the ID
    db.add(new_user)
    db.commit()
    db.refresh(new_user)

    return new_user

@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    # Fetch user by ID
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

Hoe dit werkt (Conceptueel)

Het belangrijkste concept om te begrijpen is de Session.

  1. Request komt binnen: FastAPI ziet db: Session = Depends(get_db).
  2. De Dependency (get_db) start: Er wordt een nieuwe database-sessie geopend (db = SessionLocal()).
  3. De ORM doet zijn werk: In create_user vertaal je Python objecten (new_user) naar SQL queries (INSERT INTO...).
  4. Commit: db.commit() maakt de wijzigingen permanent.
  5. Refresh: db.refresh() haalt de net gemaakte data (zoals de auto-generated id) terug uit de database naar je Python object.
  6. Cleanup: Zodra de request klaar is, zorgt de finally in get_db ervoor dat de verbinding wordt gesloten (db.close()).

Database connection tuning

De connectie met de database backend kan worden geoptimaliseerd door het toepassen van asychroniteit en pooling.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base

# Import the async connection string (e.g., postgresql+asyncpg://user:pass@localhost/db)
from .config import DATABASE_URL_ASYNC

# ============================================================
# ASYNCHRONOUS ENGINE
# ============================================================
# The engine is the primary entry point to the database.
# It manages the connection pool and dialect (SQL dialect).
async_engine = create_async_engine(
    DATABASE_URL_ASYNC,
    pool_size=10,           # Keep 10 connections open by default (baseline)
    max_overflow=20,        # Allow up to 20 additional connections during peak load
    pool_pre_ping=True,     # Critical: Checks if the connection is alive before using it.
                            # Prevents "server has gone away" errors if the DB restarts.
    pool_recycle=300        # Recycle connections every 300 seconds (5 minutes)
                            # to prevent firewalls from dropping idle connections.
)

# ============================================================
# ASYNCHRONOUS SESSION FACTORY
# ============================================================
# This factory generates a new Session for every incoming request.
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False  # CRITICAL SETTING for Async:
                            # By default, SQLAlchemy expires objects after a commit.
                            # Accessing an expired attribute triggers a refresh (IO operation).
                            # Since attribute access (e.g., user.name) is synchronous, 
                            # this would cause a crash in an async context. 
                            # Setting this to False keeps data in memory after commit.
)

# ============================================================
# MODELS BASE
# ============================================================
# All database models (tables) must inherit from this class.
Base = declarative_base()

Om dit te gebruiken in API endpoints is het volgende nodig:

from typing import AsyncGenerator

# Dependency to get the database session
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            # Note: You should handle commits explicitly in your route logic
        except Exception:
            # If an error occurs, rollback the transaction
            await session.rollback()
            raise
        finally:
            # The context manager automatically closes the session, 
            # but explicit closing is good practice in some patterns.
            await session.close()

Uitrollen van de FastAPI applicatie

Hoewel je de applicatie kunt draaien op een server of werkstation is dit vaak niet de geprefereerde manier. Vaak worden FastAPI applicaties opgenomen in een Docker image, om daarna met behulp van Docker, Podman of Kubernetes te deployen. Eventueel met tussenkomst van een CI/CD pipeline.

Docker file

FROM python:3.11-slim

WORKDIR /app
COPY . .

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

EXPOSE 8000

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

Conclusie

De kracht van FastAPI zit in de balans tussen performance en ontwikkelsnelheid.

FastAPI geeft het beste van twee werelden: de snelheid van Go of NodeJS, maar met de leesbaarheid van Python. Met minimale lappen code zet je in no-time een asynchrone API neer die klaar is voor productie.

Minder tijd kwijt aan debuggen en documenteren betekent meer tijd voor het bouwen van features die er echt toe doen.

In volgende posts wil ik de volgende onderwerpen behandelen:

  • Observability van FastAPI applicaties
  • FastAPI ingezet als frontend voor AI oplossingen.
  • …

Leave a Comment