FastAPI Database

6 Kernkonzepte

Die wichtigsten Konzepte für Datenbankintegration mit FastAPI: SQLAlchemy ORM · Async Datenbank · CRUD-Operationen · Alembic · Pydantic-Modelle · Beziehungen FastAPI lässt sich hervorragend mit SQLAlchemy und anderen Datenbank-Tools kombinieren. Diese Cheatsheet fasst die wichtigsten Techniken für die Datenbankanbindung zusammen – von ORM-Modellen über asynchrone Queries bis zu Migrationen.

SQLAlchemy ORM – Datenbankmodelle

DeclarativeBase · Mapped · mapped_column

# models.py from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] email: Mapped[str] = mapped_column(unique=True)

SQLAlchemy ORM ist die Standard-ORM für Python. Mit DeclarativeBase und Type-Hints definieren Sie Datenbanktabellen als Python-Klassen.

Beispiele

# database.py (Setup)

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})

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

# models.py – Vollständiges Modell

from datetime import datetime

from sqlalchemy import func

class User(Base):

__tablename__ = "users"

id: Mapped[int] = mapped_column(primary_key=True, index=True)

email: Mapped[str] = mapped_column(unique=True, index=True)

age: Mapped[int] = mapped_column(nullable=True)

created_at: Mapped[datetime] = mapped_column(server_default=func.now())

is_active: Mapped[bool] = mapped_column(default=True)

# Tabellen erstellen

from models import Base

Base.metadata.create_all(bind=engine)

Tipp: Für Produktionsumgebungen verwenden Sie Alembic für Migrationen (statt create_all()). Das ermöglicht kontrollierte Schema-Änderungen.

Asynchrone Datenbank – async SQLAlchemy & sqlalchemy.ext.asyncio

AsyncSession · async_engine · async_scoped_session

# database.py (async) from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker async_engine = create_async_engine("sqlite+aiosqlite:///./app.db") AsyncSessionLocal = async_sessionmaker(bind=async_engine, expire_on_commit=False)

Asynchrone Datenbank ermöglicht nicht-blockierende Datenbankoperationen in FastAPI. Verwenden Sie sqlalchemy.ext.asyncio für async/await-Unterstützung.

Beispiele

# Dependency für async Session

async def get_db() -> AsyncGenerator[AsyncSession, None]:

async with AsyncSessionLocal() as session:

yield session

# Async Query

@app.get("/users")

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

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

users = result.scalars().all()

return users

# Async Insert

@app.post("/users")

async def create_user(user_data: UserCreate, db: AsyncSession = Depends(get_db)):

user = User(name=user_data.name, email=user_data.email)

db.add(user)

await db.commit()

await db.refresh(user)

return user

# Filter Query

@app.get("/users/{user_id}")

async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):

result = await db.execute(

select(User).where(User.id == user_id)

)

user = result.scalar_one_or_none()

if user is None:

raise HTTPException(404, "Benutzer nicht gefunden")

return user

Tipp: Für asynchrone Datenbanken benötigen Sie asynchrone Treiber – für SQLite aiosqlite, für PostgreSQL asyncpg. Verwenden Sie await session.refresh(user) nach commit() für aktualisierte Daten.

CRUD-Operationen – Create, Read, Update, Delete

SELECT · INSERT · UPDATE · DELETE

# CRUD-Funktionen async def get_user(db, user_id): return await db.get(User, user_id) async def create_user(db, user_data): user = User(**user_data) db.add(user); await db.commit() return user

CRUD-Operationen sind die Grundlage jeder Datenbankanwendung – sie ermöglichen das Erstellen, Lesen, Aktualisieren und Löschen von Datensätzen.

Beispiele

# CRUD-Funktionen für User

from sqlalchemy import select

# Create

async def create_user(db: AsyncSession, user_data: UserCreate) -> User:

user = User(name=user_data.name, email=user_data.email, age=user_data.age)

db.add(user)

await db.commit()

await db.refresh(user)

return user

# Read (Alle)

async def get_users(db: AsyncSession, skip: int = 0, limit: int = 100) -> list[User]:

result = await db.execute(select(User).offset(skip).limit(limit))

return result.scalars().all()

# Read (Einzel)

async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:

return await db.get(User, user_id)

# Read (nach Email)

async def get_user_by_email(db: AsyncSession, email: str) -> User | None:

result = await db.execute(select(User).where(User.email == email))

return result.scalar_one_or_none()

# Update

async def update_user(db: AsyncSession, user_id: int, user_data: UserUpdate) -> User:

user = await get_user_by_id(db, user_id)

if user is None:

raise HTTPException(404, "User not found")

update_data = user_data.dict(exclude_unset=True)

for key, value in update_data.items():

setattr(user, key, value)

await db.commit()

await db.refresh(user)

return user

# Delete

async def delete_user(db: AsyncSession, user_id: int) -> User:

user = await get_user_by_id(db, user_id)

if user is None:

raise HTTPException(404, "User not found")

await db.delete(user)

await db.commit()

return user

Tipp: Verwenden Sie exclude_unset=True bei Pydantic-Modellen für Updates – so werden nur die tatsächlich gesendeten Felder aktualisiert.

Alembic – Datenbank-Migrationen

alembic init · alembic revision · alembic upgrade

# Initialisierung alembic init alembic # Migration erstellen alembic revision --autogenerate -m "Add users table" # Migration anwenden alembic upgrade head

Alembic ist das offizielle Migrations-Tool für SQLAlchemy. Es ermöglicht kontrollierte Schema-Änderungen und Rollbacks.

Beispiele

# alembic.ini (Konfiguration)

# SQLAlchemy-URL für Migration

sqlalchemy.url = sqlite:///./app.db

# env.py (Alembic Environment)

from models import Base

target_metadata = Base.metadata

# Migration Datei (autogeneriert)

def upgrade():

op.create_table('users',

sa.Column('id', sa.Integer(), nullable=False),

sa.Column('name', sa.String(), nullable=False),

sa.Column('email', sa.String(), nullable=False),

sa.PrimaryKeyConstraint('id')

)

# Befehle

# Migration erstellen (automatisch)

alembic revision --autogenerate -m "Add users table"

# Migration anwenden

alembic upgrade head

# Status anzeigen

alembic current

# Rollback (eine Migration)

alembic downgrade -1

# Geschichte anzeigen

alembic history

Tipp: Verwenden Sie alembic revision --autogenerate, um Migrationen automatisch aus Ihren Modellen zu generieren. Prüfen Sie die generierten Dateien immer manuell, bevor Sie sie anwenden.

Pydantic-Modelle – Validation & Serialisierung

BaseModel · from_attributes · model_dump

# schemas.py from pydantic import BaseModel, EmailStr, Field class UserBase(BaseModel): name: str = Field(..., min_length=2) email: EmailStr age: int | None = None

Pydantic-Modelle dienen als Schnittstelle zwischen API und Datenbank. Sie validieren eingehende Daten und serialisieren ORM-Objekte für Antworten.

Beispiele

# schemas.py – Vollständige Modelle

from datetime import datetime

from pydantic import BaseModel, EmailStr, Field, ConfigDict

# Basis-Modell

class UserBase(BaseModel):

email: EmailStr

age: int | None = Field(None, ge=0, le=120)

# Create-Modell

class UserCreate(UserBase):

password: str = Field(..., min_length=8)

# Update-Modell (alle Felder optional)

class UserUpdate(BaseModel):

email: EmailStr | None = None

age: int | None = None

# Response-Modell (mit ID und Timestamp)

class UserResponse(UserBase):

id: int

is_active: bool

created_at: datetime

model_config = ConfigDict(from_attributes=True)

# Verwendung im Endpunkt

@app.post("/users", response_model=UserResponse)

async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):

db_user = await crud.create_user(db, user)

return db_user # Wird automatisch in UserResponse konvertiert

Tipp: Verwenden Sie model_config = ConfigDict(from_attributes=True) in Response-Modellen, um ORM-Objekte automatisch zu serialisieren – das ersetzt manuelle from_orm-Aufrufe.

Beziehungen & Joins – One-to-Many, Many-to-Many

relationship · ForeignKey · joinedload

# One-to-Many Beispiel class User(Base): posts = relationship("Post", back_populates="author") class Post(Base): user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) author = relationship("User", back_populates="posts")

Beziehungen ermöglichen das Verknüpfen von Tabellen – One-to-Many (User → Posts) und Many-to-Many (User → Roles) sind häufig.

Beispiele

# models.py – One-to-Many

from sqlalchemy.orm import relationship, back_populates

class User(Base):

__tablename__ = "users"

id: Mapped[int] = mapped_column(primary_key=True)

posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")

class Post(Base):

__tablename__ = "posts"

id: Mapped[int] = mapped_column(primary_key=True)

title: Mapped[str]

user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)

author = relationship("User", back_populates="posts")

# Query mit Join (eager loading)

@app.get("/users/{user_id}/posts")

async def get_user_posts(user_id: int, db: AsyncSession = Depends(get_db)):

result = await db.execute(

select(User)

.where(User.id == user_id)

.options(joinedload(User.posts))

)

user = result.scalar_one_or_none()

return user.posts if user else []

# Many-to-Many (User ↔ Role)

user_roles = Table(

"user_roles", Base.metadata,

Column("user_id", ForeignKey("users.id"), primary_key=True),

Column("role_id", ForeignKey("roles.id"), primary_key=True)

)

# models.py – Many-to-Many

class User(Base):

roles = relationship("Role", secondary="user_roles", back_populates="users")

class Role(Base):

users = relationship("User", secondary="user_roles", back_populates="roles")

Tipp: Verwenden Sie joinedload() für eager loading (N+1-Problem vermeiden). Für große Datensätze ist selectinload() oft effizienter.

FastAPI Database im Überblick

ORM SQLAlchemy Modelle
DeclarativeBase, Mapped

Async Asynchrone Datenbank
AsyncSession, asyncpg

CRUD Create, Read, Update, Delete
Grundoperationen

Alembic Migrationen
Schema-Änderungen

Pydantic Validation & Serialisierung
from_attributes

Relations Beziehungen & Joins
One-to-Many, Many-to-Many

Quick Summary

Base

ORM-Modelle

async

Async Datenbank

CRUD

Operationen

Alembic

Migrationen

Pydantic

Validation

relationship

Beziehungen

class User(Base): __tablename__ = "users" · async def get_db() -> AsyncGenerator[AsyncSession, None] · alembic upgrade head