FastAPI Patterns
Moderne, produktionsreife FastAPI-Entwicklung: Projektstruktur, Pydantic v2-Schemas, Dependency Injection, asynchrone Muster, Authentifizierung, transaktionale Service-Methoden und Tests.
Projektstruktur
my_app/
|-- app/
| |-- main.py # App Factory, Lifespan, Middleware
| |-- config.py # Einstellungen via pydantic-settings
| |-- dependencies.py # Gemeinsame FastAPI-Dependencies
| |-- database.py # SQLAlchemy Engine + Session
| |-- routers/
| | `-- users.py
| |-- models/ # SQLAlchemy ORM Modelle
| | `-- user.py
| |-- schemas/ # Pydantic Request/Response Schemas
| | `-- user.py
| `-- services/ # Geschäftslogikschicht
| `-- user_service.py
|-- tests/
| |-- conftest.py
| `-- test_users.py
|-- pyproject.toml
`-- .env
App Factory und Lifespan
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.database import engine, Base
from app.routers import users
@asynccontextmanager
async def lifespan(app: FastAPI):
# Automatisches Erstellen von Tabellen beim Start für einfache Nutzung in Entwicklungs-/Demo-Umgebungen.
# Für strenge Produktionsanwendungen verwalten Sie Schemas stattdessen über Alembic-Migrationen.
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# Herunterfahren: Gepoolte Ressourcen schließen.
await engine.dispose()
def create_app() -> FastAPI:
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins,
allow_credentials=settings.allow_credentials,
allow_methods=settings.allowed_methods,
allow_headers=settings.allowed_headers,
)
app.include_router(users.router, prefix="/users", tags=["users"])
return app
app = create_app()
Konfiguration mit pydantic-settings
# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
app_name: str = "My App"
app_version: str = "0.1.0"
debug: bool = False
database_url: str
secret_key: str
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
# Pydantic-settings v2 wertet veränderbare List-Literale sicher direkt aus
allowed_origins: list[str] = ["http://localhost:3000"]
allowed_methods: list[str] = ["GET", "POST", "PATCH", "DELETE", "OPTIONS"]
allowed_headers: list[str] = ["Authorization", "Content-Type"]
allow_credentials: bool = True
settings = Settings()
Pydantic Schemas (v2)
# app/schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field, model_validator
class UserBase(BaseModel):
email: EmailStr
username: str = Field(min_length=3, max_length=50)
class UserCreate(UserBase):
password: str = Field(min_length=8)
password_confirm: str
@model_validator(mode="after")
def passwords_match(self) -> "UserCreate":
if self.password != self.password_confirm:
raise ValueError("Passwörter stimmen nicht überein")
return self
class UserUpdate(BaseModel):
username: str | None = Field(default=None, min_length=3, max_length=50)
email: EmailStr | None = None
class UserResponse(UserBase):
id: int
is_active: bool
created_at: datetime
model_config = {"from_attributes": True}
class UserListResponse(BaseModel):
total: int
items: list[UserResponse]
Dependency Injection
# app/dependencies.py
from typing import Annotated, AsyncGenerator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database import AsyncSessionLocal
from app.models.user import User
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/users/token")
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
try:
yield session
except Exception:
await session.rollback()
raise
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Anmeldeinformationen konnten nicht validiert werden",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
subject = payload.get("sub")
if subject is None:
raise credentials_exception
user_id = int(subject)
except (JWTError, TypeError, ValueError):
raise credentials_exception
user = await db.get(User, user_id)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(
current_user: Annotated[User, Depends(get_current_user)],
) -> User:
if not current_user.is_active:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Inaktiver Benutzer")
return current_user
DbDep = Annotated[AsyncSession, Depends(get_db)]
CurrentUserDep = Annotated[User, Depends(get_current_user)]
ActiveUserDep = Annotated[User, Depends(get_current_active_user)]
Router und Endpunkt-Design
# app/routers/users.py
from typing import Annotated
from fastapi import APIRouter, HTTPException, Query, status
from fastapi.security import OAuth2PasswordRequestForm
from app.dependencies import ActiveUserDep, DbDep
from app.schemas.user import UserCreate, UserResponse, UserUpdate, UserListResponse
from app.services.user_service import DuplicateUserError, UserService
router = APIRouter()
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserCreate, db: DbDep) -> UserResponse:
service = UserService(db)
try:
return await service.create(payload)
except DuplicateUserError:
raise HTTPException(status_code=400, detail="E-Mail bereits registriert")
@router.get("/me", response_model=UserResponse)
async def get_me(current_user: ActiveUserDep) -> UserResponse:
return current_user
@router.get("/", response_model=UserListResponse)
async def list_users(
db: DbDep,
current_user: ActiveUserDep,
skip: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
) -> UserListResponse:
service = UserService(db)
users, total = await service.list(skip=skip, limit=limit)
return UserListResponse(total=total, items=users)
@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
user_id: int,
payload: UserUpdate,
db: DbDep,
current_user: ActiveUserDep,
) -> UserResponse:
if current_user.id != user_id:
raise HTTPException(status_code=403, detail="Nicht autorisiert")
service = UserService(db)
try:
user = await service.update(user_id, payload)
except DuplicateUserError:
raise HTTPException(status_code=400, detail="E-Mail bereits registriert")
if user is None:
raise HTTPException(status_code=404, detail="Benutzer nicht gefunden")
return user
@router.post("/token")
async def login(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
db: DbDep,
) -> dict[str, str]:
service = UserService(db)
token = await service.authenticate(form_data.username, form_data.password)
if token is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Falscher Benutzername oder falsches Passwort",
headers={"WWW-Authenticate": "Bearer"},
)
return {"access_token": token, "token_type": "bearer"}
Service-Schicht
# app/services/user_service.py
from datetime import datetime, timedelta, timezone
from jose import jwt
from passlib.context import CryptContext
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class DuplicateUserError(Exception):
"""Wird ausgelöst, wenn ein eindeutiges Benutzerfeld mit einer vorhandenen Zeile kollidiert."""
class UserService:
def __init__(self, db: AsyncSession) -> None:
self.db = db
async def get_by_email(self, email: str) -> User | None:
result = await self.db.execute(select(User).where(User.email == email))
return result.scalar_one_or_none()
async def create(self, payload: UserCreate) -> User:
user = User(
email=payload.email,
username=payload.username,
hashed_password=pwd_context.hash(payload.password),
)
self.db.add(user)
try:
# Verlassen Sie sich auf atomare DB-Constraints anstelle von fehleranfälligen Pre-Checks auf Anwendungsebene
await self.db.commit()
except IntegrityError as exc:
await self.db.rollback()
raise DuplicateUserError from exc
await self.db.refresh(user)
return user
async def list(self, skip: int = 0, limit: int = 20) -> tuple[list[User], int]:
total_result = await self.db.execute(select(func.count(User.id)))
total = total_result.scalar_one()
# Erzwingen Sie eine explizite deterministische Sortierung für eine zuverlässige Paginierung
result = await self.db.execute(
select(User).order_by(User.id).offset(skip).limit(limit)
)
return list(result.scalars()), total
async def update(self, user_id: int, payload: UserUpdate) -> User | None:
user = await self.db.get(User, user_id)
if user is None:
return None
for field, value in payload.model_dump(exclude_unset=True).items():
setattr(user, field, value)
try:
await self.db.commit()
except IntegrityError as exc:
await self.db.rollback()
raise DuplicateUserError from exc
await self.db.refresh(user)
return user
async def authenticate(self, email: str, password: str) -> str | None:
user = await self.get_by_email(email)
if user is None or not pwd_context.verify(password, user.hashed_password):
return None
expire = datetime.now(timezone.utc) + timedelta(
minutes=settings.access_token_expire_minutes
)
return jwt.encode(
{"sub": str(user.id), "exp": expire},
settings.secret_key,
algorithm=settings.algorithm,
)
Hinweis zum Datenbankdesign: Die Behandlung von Eindeutigkeit auf Anwendungsebene erfordert einen zugrunde liegenden eindeutigen Datenbankindex (z. B.
unique=Truein Ihren SQLAlchemy-Mapping-Attributen). Ohne zugrunde liegende Constraints kann die Fehlerbehandlung auf Anwendungsebene gleichzeitige Race-Conditions nicht sicher verhindern.
Testen mit httpx und pytest
# tests/conftest.py
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.database import Base
from app.dependencies import get_db
from app.main import create_app
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
engine = create_async_engine(TEST_DATABASE_URL)
TestingSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
@pytest_asyncio.fixture(autouse=True)
async def setup_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture
async def db_session():
async with TestingSessionLocal() as session:
yield session
await session.rollback()
@pytest_asyncio.fixture
async def client(db_session: AsyncSession):
app = create_app()
async def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
yield ac
@pytest_asyncio.fixture
async def registered_user(client: AsyncClient) -> dict:
resp = await client.post("/users/", json={
"email": "[email protected]",
"username": "testuser",
"password": "securepass1",
"password_confirm": "securepass1",
})
assert resp.status_code == 201
return resp.json()
@pytest_asyncio.fixture
async def auth_token(client: AsyncClient, registered_user: dict) -> str:
resp = await client.post("/users/token", data={
"username": "[email protected]",
"password": "securepass1",
})
assert resp.status_code == 200
return resp.json()["access_token"]
@pytest_asyncio.fixture
async def auth_client(client: AsyncClient, auth_token: str) -> AsyncClient:
client.headers.update({"Authorization": f"Bearer {auth_token}"})
return client
Anti-Patterns
# Schlecht: Geschäftslogik in Route-Handlern.
@router.post("/users/")
async def create_user(payload: UserCreate, db: DbDep):
hashed = bcrypt.hash(payload.password)
user = User(email=payload.email, hashed_password=hashed)
db.add(user)
await db.commit()
return user
# Gut: Dünner Route, transaktionale Service-Abwicklung.
@router.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(payload: UserCreate, db: DbDep):
try:
return await UserServ
... (Fortsetzung des Markdown-Dokuments folgt)