Claude Code SQLAlchemy: Models, Queries und Migrations effizient umsetzen

SQLAlchemy ist das mächtigste Python-ORM auf dem Markt — und gleichzeitig eines, bei dem die Lernkurve steil ist. DeclarativeBase, Relationship-Konfiguration, Session-Management, Async-Support mit asyncpg, Alembic-Migrations: Wer SQLAlchemy professionell einsetzt, hat ein Dutzend Konzepte gleichzeitig im Kopf. Claude Code SQLAlchemy verändert diesen Workflow grundlegend: nicht weil es das Nachdenken abnimmt, sondern weil es Models generiert, Queries korrekt formuliert und N+1-Probleme erkennt, bevor sie in Produktion landen.

Dieser Artikel zeigt den kompletten SQLAlchemy-Werkzeugkasten — von der Model-Definition über Async-Sessions bis zu Alembic-Migrations — und wie Claude Code an jedem Punkt konkret hilft.

1. Was ist SQLAlchemy?

SQLAlchemy ist eine Python-Bibliothek mit zwei unabhängigen Schichten: dem Core (SQL-Ausdrucks-Toolkit) und dem ORM (Object Relational Mapper). Man kann beide kombinieren oder einzeln nutzen.

• SQLAlchemy Core erlaubt es, SQL-Ausdrücke als Python-Objekte zu bauen — typsicher, datenbankagnostisch, ohne ein vollständiges ORM-Modell zu benötigen.
• SQLAlchemy ORM mapped Python-Klassen auf Datenbanktabellen. Relations, Lazy-Loading, Identity-Map, Unit-of-Work — alles inklusive.
• Async-Support wurde mit SQLAlchemy 1.4 eingeführt und in 2.0 vollständig stabilisiert. Mit AsyncSession und asyncpg (für PostgreSQL) oder aiosqlite (für SQLite) laufen Datenbankoperationen vollständig nicht-blockierend.

SQLAlchemy 2.0 (aktuelle Hauptversion) brachte einen vollständigen API-Umbau: select() statt session.query(), DeclarativeBase statt declarative_base(), typisierte Mapped-Felder statt Column-Deklarationen ohne Typinfo. Wer noch Legacy-Code von 1.x betreibt, findet in Claude Code einen zuverlässigen Migrations-Partner für den Umbau.

2. Models definieren: DeclarativeBase, Column-Typen, Relationships

Das Herzstück jedes SQLAlchemy-Projekts ist die Model-Definition. Mit SQLAlchemy 2.0 und DeclarativeBase sieht das so aus:

from sqlalchemy import String, Integer, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    posts: Mapped[list["Post"]] = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    body: Mapped[str] = mapped_column(String)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    author: Mapped["User"] = relationship("User", back_populates="posts")

Das ist eine One-to-Many-Beziehung: ein User hat viele Posts. back_populates sorgt dafür, dass beide Seiten synchron bleiben. Für Many-to-Many braucht man eine Zwischentabelle:

from sqlalchemy import Table, Column

tag_post = Table(
    "tag_post",
    Base.metadata,
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    posts: Mapped[list["Post"]] = relationship("Post", secondary=tag_post, back_populates="tags")

# Post bekommt zusätzlich:
# tags: Mapped[list["Tag"]] = relationship("Tag", secondary=tag_post, back_populates="posts")

Claude Code SQLAlchemy: Models generieren lassen

Statt Models von Hand zu tippen, lässt sich Claude Code die Arbeit abnehmen. Ein kurzes Prompt reicht:

claude "Erstelle SQLAlchemy 2.0 Models für ein Blog-System mit Users,
Posts, Comments und Tags. One-to-many für User→Posts und User→Comments,
many-to-many für Posts↔Tags. Alle Felder mit Mapped[] typisiert."

Claude Code liest bestehende Models im Projekt, erkennt Konventionen und generiert konsistente neue Models — inklusive korrekter back_populates-Konfiguration, die bei manueller Arbeit häufig fehlt oder falsch gesetzt wird.

3. Sessions: Session, sessionmaker, AsyncSession

Die Session ist das Herzstück des SQLAlchemy ORM — sie verwaltet den Lebenszyklus von Objekten, puffert Änderungen (Unit of Work) und kommuniziert mit der Datenbank. Falsch eingesetzt ist sie die häufigste Quelle von Bugs:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql://user:pass@localhost/mydb")
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

# Verwendung
with SessionLocal() as session:
    user = User(email="test@example.com", name="Test User")
    session.add(user)
    session.commit()
    session.refresh(user)  # ID nach INSERT laden

Für FastAPI und andere Async-Frameworks ist AsyncSession mit asyncpg die richtige Wahl:

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

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_session():
    async with AsyncSessionLocal() as session:
        yield session

# In einer FastAPI-Route:
async def create_user(session: AsyncSession = Depends(get_session)):
    user = User(email="async@example.com", name="Async User")
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

4. Queries: select(), filter(), join(), order_by(), limit()

SQLAlchemy 2.0 verwendet die select()-Funktion für alle Abfragen — konsistent für sync und async:

from sqlalchemy import select, and_, or_

# Einfache Abfrage
stmt = select(User).where(User.email == "test@example.com")
result = await session.execute(stmt)
user = result.scalar_one_or_none()

# Filter kombinieren
stmt = select(Post).where(
    and_(Post.author_id == user.id, Post.title.ilike("%python%"))
).order_by(Post.created_at.desc()).limit(10)

# JOIN
stmt = (
    select(Post, User)
    .join(User, Post.author_id == User.id)
    .where(User.name == "Daniel")
    .order_by(Post.created_at.desc())
)
rows = (await session.execute(stmt)).all()

# Eager Loading (N+1-Problem vermeiden)
from sqlalchemy.orm import selectinload
stmt = select(User).options(selectinload(User.posts))
users = (await session.execute(stmt)).scalars().all()

5. Migrations mit Alembic

Alembic ist das offizielle Migrations-Tool für SQLAlchemy. Schema-Änderungen werden als versionierte Python-Skripte gespeichert, die vor- und rückwärts ausgeführt werden können.

# Einrichtung
pip install alembic
alembic init alembic

# alembic/env.py — Models importieren und Metadata setzen
from myapp.models import Base
target_metadata = Base.metadata

# Neue Migration erstellen (Autogenerate)
alembic revision --autogenerate -m "add_posts_table"

# Migration ausführen
alembic upgrade head

# Rückgängig machen
alembic downgrade -1

Eine generierte Migration sieht so aus:

def upgrade() -> None:
    op.create_table(
        "posts",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("title", sa.String(255), nullable=False),
        sa.Column("author_id", sa.Integer(), nullable=True),
        sa.ForeignKeyConstraint(["author_id"], ["users.id"]),
        sa.PrimaryKeyConstraint("id"),
    )

def downgrade() -> None:
    op.drop_table("posts")

Claude Code hilft dabei, env.py korrekt zu konfigurieren (vor allem für Async-Setups, bei denen run_async_migrations() nötig ist) und überprüft generierte Migrationen auf häufige Fehler wie fehlende Index-Definitionen oder falsch gesetzte Nullable-Flags.

6. Raw SQL vs. ORM: Wann welches?

SQLAlchemy ORM ist nicht für jede Abfrage die beste Wahl. Eine klare Entscheidungsregel:

• ORM verwenden für CRUD-Operationen, wenn Objekte mit Relationships verarbeitet werden, und überall wo das Unit-of-Work-Pattern hilft (mehrere verbundene Inserts/Updates in einer Transaktion).
• Core (SQL-Expressions) verwenden für komplexe Aggregations-Abfragen, Bulk-Inserts/-Updates, Reporting-Queries mit vielen Joins — überall wo Performance kritisch ist und kein Objektgraph benötigt wird.
• Raw SQL (text()) nur wenn die ORM-Ausdrucks-API die benötigte Query nicht ausdrücken kann — zum Beispiel für datenbankspezifische Funktionen oder sehr komplexe CTEs.

# Bulk-Insert mit Core statt ORM (deutlich schneller)
from sqlalchemy import insert

await session.execute(
    insert(Post),
    [{"title": f"Post {i}", "author_id": 1} for i in range(1000)]
)
await session.commit()

7. Claude Code SQLAlchemy: Konkrete Tipps

Aus dem täglichen Einsatz drei Muster, die konstant Zeit sparen:

N+1-Problem erkennen und lösen

Das N+1-Problem ist die häufigste Performance-Falle in ORM-Code: für jeden Datensatz wird eine zusätzliche Query ausgeführt. Claude Code erkennt das Muster zuverlässig:

claude "Analysiere diese Datenbankabfrage auf N+1-Probleme und
zeige die optimierte Version mit selectinload oder joinedload:

users = session.execute(select(User)).scalars().all()
for user in users:
    print(user.posts)  # N+1!"

Claude Code liefert nicht nur den Fix (selectinload oder joinedload je nach Usecase), sondern erklärt auch wann welches sinnvoll ist: joinedload für Single-Objekte, selectinload für Collections, weil es ein separates IN-Query erzeugt statt eines kartesischen Produkts.

Komplexe Queries schreiben lassen

SQLAlchemy-Queries mit mehreren Joins, Subqueries und Window-Functions sind auch für erfahrene Entwickler zeitaufwendig. Ein konkreter Prompt spart hier am meisten:

claude "Schreibe eine SQLAlchemy 2.0 Query die für jeden User die Anzahl
seiner Posts im letzten Monat zählt, nur Users mit mindestens 3 Posts
zurückgibt, und nach Post-Anzahl absteigend sortiert."

Legacy-Code auf SQLAlchemy 2.0 umstellen

claude "Migriere diese SQLAlchemy 1.x Query auf 2.0 API:
result = session.query(User).filter(User.active == True).all()"

Claude Code wandelt session.query() in select(), aktualisiert Filter-Syntax und ergänzt fehlende scalars()-Aufrufe — und macht dabei das Typing konsistent, falls Mapped[] noch fehlt.

"SQLAlchemy hat viele richtige Wege und noch mehr falsche. Claude Code kennt die Unterschiede — nicht als Gedächtnissatz, sondern weil es deinen konkreten Code sieht und den richtigen Weg für dein Setup wählt."

Verwandte Artikel, die auf diesem Thema aufbauen:

• Claude Code für Python-Projekte — der komplette Python-Workflow mit Claude Code
• Claude Code Debugging — Datenbankfehler und ORM-Bugs systematisch aufspüren