FastAPI Dependency Injection

6 Kernkonzepte

Die wichtigsten Konzepte für Dependency Injection in FastAPI: Depends · Sub-Dependencies · yield · Path/Query/Header · Class Dependencies · Global Dependencies Dependency Injection (DI) ist ein zentrales Feature von FastAPI. Es ermöglicht die Wiederverwendung von Logik, die saubere Trennung von Verantwortlichkeiten und das einfache Testen von Anwendungen.

Basis – Depends() Grundlagen

Depends(callable) · async def · parameter

from fastapi import Depends async def common_params(q: str = None, skip: int = 0): return {"q": q, "skip": skip} @app.get("/items") async def get_items(params: dict = Depends(common_params)): return params

Depends() ist die zentrale Funktion für Dependency Injection in FastAPI. Es injiziert den Rückgabewert einer Callable als Parameter in Ihre Endpunkt-Funktion.

Beispiele

# Einfache Dependency ohne Parameter

async def get_db():

return {"connection": "db_conn"}

@app.get("/users")

async def get_users(db: dict = Depends(get_db)):

return {"db": db, "users": []}

# Dependency mit Parametern

async def pagination(page: int = 1, size: int = 10):

return {"page": page, "size": size}

@app.get("/items")

async def get_items(pagination: dict = Depends(pagination)):

return pagination

# Sync Dependency (wird im Thread-Pool ausgeführt)

def get_config():

return {"debug": True}

@app.get("/config")

async def get_config(config: dict = Depends(get_config)):

return config

Tipp: Dependencies können sowohl async def als auch def sein. FastAPI entscheidet automatisch, ob sie im Event-Loop oder im Thread-Pool ausgeführt werden.

Sub-Dependencies – Abhängigkeiten verschachteln

Depends(another_dependency) · chain

async def get_user(user_id: int): return {"id": user_id, "name": "Anna"} async def get_current_user(user: dict = Depends(get_user)): return user @app.get("/profile") async def profile(user: dict = Depends(get_current_user)): return user

Sub-Dependencies sind Dependencies, die selbst wieder andere Dependencies verwenden. FastAPI löst die gesamte Kette automatisch auf.

Beispiele

# Token aus Header extrahieren

async def get_token(authorization: str = Header(...)):

if not authorization.startswith("Bearer "):

raise HTTPException(401, "Ungültiger Token")

return authorization.split(" ")[1]

# Token validieren und User laden

async def get_current_user(token: str = Depends(get_token)):

# Token validieren (z.B. JWT)

user = await validate_token(token)

return user

# Drei Ebenen tiefe Sub-Dependency

@app.get("/protected")

async def protected_endpoint(user: dict = Depends(get_current_user)):

return {"user": user, "message": "Zugriff gewährt"}

# Mehrere Sub-Dependencies kombinieren

async def get_db_connection():

return "db_connection"

async def get_user_repo(db: str = Depends(get_db_connection)):

return f"UserRepository({db})"

@app.get("/users/repo")

async def get_users_repo(repo: str = Depends(get_user_repo)):

return {"repo": repo}

Tipp: FastAPI optimiert Sub-Dependencies – eine Dependency wird nur einmal pro Request aufgerufen, auch wenn sie mehrfach verwendet wird.

yield – Resource-Management

yield · try/finally · context manager

async def get_db(): db = Database() try: yield db finally: db.close()

yield in Dependencies ermöglicht die Verwaltung von Ressourcen (Datenbankverbindungen, Datei-Handles). Der Code nach yield wird als Cleanup ausgeführt.

Beispiele

# Datenbankverbindung mit Cleanup

async def get_db_connection():

conn = await create_connection()

try:

yield conn

finally:

await conn.close()

# SQLAlchemy-Session

async def get_session():

async with async_session() as session:

yield session

@app.get("/users")

async def get_users(session: AsyncSession = Depends(get_session)):

result = await session.execute(select(User))

return result.scalars().all()

# File-Handle mit yield

async def get_file(path: str):

file = open(path, "r")

try:

yield file

finally:

file.close()

# Exception-Handling im yield

async def get_session():

try:

session = Session()

yield session

except Exception:

session.rollback()

raise

finally:

session.close()

Tipp: yield-Dependencies werden automatisch als Context Manager behandelt. Der Cleanup-Code wird auch bei Exceptions ausgeführt – perfekt für Datenbankverbindungen.

Path/Query/Header – Dependencies mit Request-Daten

Path() · Query() · Header() · Cookie()

from fastapi import Query, Header, Path async def get_pagination( page: int = Query(1, ge=1), size: int = Query(10, le=100) ): return {"page": page, "size": size}

Path, Query, Header sind Hilfsfunktionen, um Dependencies mit Request-Daten zu definieren – mit automatischer Validierung und Typüberprüfung.

Beispiele

# Query-Parameter als Dependency

async def get_search_params(

q: str = Query(..., min_length=2),

sort: str = Query("asc", regex="^(asc|desc)$")

return {"q": q, "sort": sort}

# Header-Dependency (z.B. für Authentifizierung)

async def get_authorization(

authorization: str = Header(..., description="Bearer Token")

return authorization

# Path-Parameter als Dependency

async def get_user_id(

user_id: int = Path(..., ge=1)

return user_id

# Cookie-Dependency

async def get_session_id(

session_id: str = Cookie(None)

return session_id

# Kombination verschiedener Request-Daten

async def get_request_metadata(

user_id: int = Path(...),

lang: str = Query("de"),

user_agent: str = Header(None)

return {"user_id": user_id, "lang": lang, "user_agent": user_agent}

Tipp: Nutzen Sie die Validierungsparameter (ge, le, regex, etc.) direkt in Query(), Path() und Header() – das spart zusätzliche Validierungslogik.

Class Dependencies – Dependencies als Klassen

__call__ · __init__ · class-based

class Pagination: def __init__(self, page: int = 1, size: int = 10): self.page = page self.size = size def __call__(self): return {"page": self.page, "size": self.size}

Class Dependencies bieten eine objektorientierte Möglichkeit, Dependencies zu definieren – besonders nützlich für wiederverwendbare Komponenten mit Zustand.

Beispiele

# Class Dependency für Authentifizierung

class AuthDependency:

def __init__(self, required: bool = True):

self.required = required

async def __call__(self, token: str = Header(None)):

if self.required and not token:

raise HTTPException(401, "Token erforderlich")

return token

# Verwendung

@app.get("/protected")

async def protected(token: str = Depends(AuthDependency(required=True))):

return {"token": token}

# Class Dependency für Datenbank-Repository

class UserRepository:

def __init__(self, session: AsyncSession):

self.session = session

async def get_all(self):

result = await self.session.execute(select(User))

return result.scalars().all()

# Factory für Repository

async def get_user_repo(session: AsyncSession = Depends(get_session)):

return UserRepository(session)

@app.get("/users")

async def get_users(repo: UserRepository = Depends(get_user_repo)):

return await repo.get_all()

Tipp: Class Dependencies sind ideal für Repository-Patterns oder Services, die während der gesamten Request-Lebensdauer bestehen bleiben.

Globale Dependencies – Für alle Endpunkte

app = FastAPI(dependencies=[Depends(...)])

from fastapi import FastAPI, Depends async def global_dep(): return {"global": "data"} app = FastAPI(dependencies=[Depends(global_dep)])

Globale Dependencies werden für jeden Endpunkt der Anwendung ausgeführt – ideal für Logging, Authentifizierung oder Request-Tracking.

Beispiele

# Globales Request-Logging

import time

async def log_request():

print(f"Request gestartet: {time.time()}")

return "logged"

app = FastAPI(dependencies=[Depends(log_request)])

# Globale Authentifizierung (alle Endpunkte schützen)

async def auth_global(token: str = Header(...)):

if token != "secret":

raise HTTPException(401, "Ungültiger Token")

return token

app = FastAPI(dependencies=[Depends(auth_global)])

# Router-spezifische Dependencies

from fastapi import APIRouter

router = APIRouter(dependencies=[Depends(auth_global)])

@router.get("/protected")

async def protected():

return {"message": "Geschützt"}

app.include_router(router)

# Globale Dependency mit Sub-Dependencies

async def get_db():

return "db_connection"

async def global_with_db(db: str = Depends(get_db)):

return {"db": db}

app = FastAPI(dependencies=[Depends(global_with_db)])

Tipp: Verwenden Sie globale Dependencies für Cross-Cutting-Concerns wie Authentifizierung, Logging oder Request-Tracking. Für ressourcenintensive Dependencies (Datenbank) ist yield die bessere Wahl.

Dependency Injection im Überblick

Depends Basis-Dependency
Funktion/Class aufrufen

Sub-Dep Verschachtelte Dependencies
Automatische Auflösung

yield Resource-Management
Context Manager

Query/Path Request-Daten
Validierung

Class OOP-Dependencies
Zustand und Methoden

Global Globale Dependencies
Für alle Endpunkte

Quick Summary

Depends

Basis

Sub-Dep

Verschachtelt

yield

Resource-Management

Query

Request-Daten

Class

OOP

Global

Global

Depends(get_db) · async def get_db(): yield db; db.close() · FastAPI(dependencies=[Depends(auth)])