Naki-404/naki-postgres

PostgreSQL und SQLAlchemy 2

Wann verwenden

• Entwerfen von Tabellen, Constraints und Indizes.
• Schreiben von SQLAlchemy-2-Modellen und -Abfragen, synchron oder asynchron.
• Eine Abfrage ist langsam, Locks häufen sich an oder Verbindungen gehen aus.
• Schreiben einer Alembic-Migration, insbesondere für Tabellen mit Live-Traffic.

Für destruktive Datenänderungen zusätzlich naki-safety anwenden. Falls das Projekt eine Regel-Fähigkeit hat (z. B. naki-myapp), dieser folgen.

Schema-Regeln

1. Primärschlüssel: bigint generated always as identity oder UUID (wenn verfügbar v7, wegen Index-Lokalität). Keine fortlaufenden IDs öffentlich preisgeben, wenn Aufzählbarkeit relevant ist.
2. Typen: timestamptz (niemals timestamp ohne Zeitzone), text mit einem CHECK auf Länge statt varchar(n), wenn Grenzen sich ändern könnten, numeric für Geldbeträge, jsonb (nicht json), boolean not null default false.
3. Constraints sind die letzte Verteidigungslinie: NOT NULL standardmäßig, UNIQUE, CHECK, Fremdschlüssel mit expliziter ON DELETE-Wahl.
4. Jeden Fremdschlüssel indexieren, der in Joins oder Kaskaden verwendet wird; PostgreSQL erledigt das nicht automatisch.
5. Constraints und Indizes benennen über eine Namenskonvention in SQLAlchemy MetaData, damit Alembic-Diffs stabil sind.
6. jsonb für wirklich variable Daten, nicht um Modellierung zu vermeiden. Abfrage-häufige Schlüssel werden zu echten Spalten.
7. Soft Delete nur, wenn nötig: falls doch, partielle Indizes WHERE deleted_at IS NULL hinzufügen.

Index-Regeln

Abfragemuster	Index
Gleichheit und Bereich auf einer Spalte	B-tree
WHERE a = ? AND b > ? ORDER BY b	zusammengesetzt (a, b), Gleichheitsspalten zuerst
Teilmenge von Zeilen (offene Elemente)	partiell WHERE status = 'open'
lower(email) = ?	Ausdrucksindex auf lower(email)
jsonb-Enthaltensein @>	GIN
Volltext, Trigramm ILIKE '%x%'	GIN mit tsvector oder pg_trgm
riesige Append-Only-Zeitreihen	BRIN auf dem Zeitstempel

• Jeder Index kostet Schreiboperationen und Speicherplatz. Nicht verwendete Indizes entfernen (pg_stat_user_indexes.idx_scan = 0 über einen langen Zeitraum).
• Abdeckender Index mit INCLUDE (...) für Index-Only-Scans bei heißen Leseoperationen.

Query-Tuning-Workflow

1. Die echte Abfrage besorgen (echo oder log_min_duration_statement aktivieren) und die echten Parameter.
2. EXPLAIN (ANALYZE, BUFFERS) auf einem produktionsähnlichen Datensatz ausführen. Plans von einer 100-Zeilen-Entwicklungstabelle nie vertrauen.
3. Suchen nach: Seq Scan auf großen Tabellen mit selektiven Filtern, Zeilenschätzung weit weg vom tatsächlichen Wert (veraltete Statistiken: ANALYZE), verschachtelten Schleifen über viele Zeilen, Sortierungen, die auf Festplatte auslagern, hohen shared read-Werten.
4. Die günstigste Ursache beheben: Index hinzufügen oder anpassen, Prädikat sargbar umschreiben, N+1 entfernen, per Keyset paginieren.
5. EXPLAIN ANALYZE erneut ausführen und Zeiten vergleichen. pg_stat_statements verwenden, um die Abfragen mit der höchsten Gesamtzeit zu finden.

SQLAlchemy-2-Regeln

1. Den 2.0-Stil verwenden: select(), session.execute(), session.scalars(), Mapped[...] und mapped_column.
2. Eine Session pro Arbeitseinheit (Request, Job). Niemals eine Session über Threads oder Tasks hinweg teilen.
3. N+1 vermeiden: Beziehungen explizit mit selectinload (Sammlungen) oder joinedload (viele-zu-eins) laden. In asynchronem Code löst lazy loading einen Fehler aus; setzen Sie lazy="raise" auf Beziehungen, um es auch in synchronem Code zu erwischen.
4. Nur gebundene Parameter verwenden. Niemals Benutzereingaben in SQL-Strings einfügen, auch nicht in text().
5. Pool-Größe und pool_pre_ping=True setzen; die Gesamtzahl der Verbindungen über alle App-Instanzen hinweg muss unter max_connections bleiben (PgBouncer verwenden, wenn viele Worker).
6. Transaktionen kurz halten; keine Netzwerkaufrufe an andere Dienste, während eine Transaktion gehalten wird.

from datetime import datetime
from sqlalchemy import ForeignKey, Index, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload

class Base(DeclarativeBase):
    pass

class Incident(Base):
    __tablename__ = "incidents"
    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[str] = mapped_column(String(16), index=True)
    opened_at: Mapped[datetime]
    alerts: Mapped[list["Alert"]] = relationship(back_populates="incident", lazy="raise")
    __table_args__ = (Index("ix_incidents_open", "opened_at", postgresql_where="status = 'open'"),)

async def open_incidents(session: AsyncSession, limit: int) -> list[Incident]:
    stmt = (
        select(Incident)
        .where(Incident.status == "open")
        .order_by(Incident.opened_at.desc(), Incident.id.desc())
        .limit(limit)
        .options(selectinload(Incident.alerts))
    )
    return list((await session.scalars(stmt)).all())

Transaktionen und Locking

• Standard READ COMMITTED ist für die meisten Arbeiten ausreichend. SELECT ... FOR UPDATE verwenden, um Zeilen zu sperren, die basierend auf dem Gelesenen geändert werden.
• Für Warteschlangen: FOR UPDATE SKIP LOCKED ermöglicht mehreren Workern, Jobs zu beanspruchen, ohne sich gegenseitig zu blockieren.
• Upserts: INSERT ... ON CONFLICT (key) DO UPDATE anstelle von erst lesen, dann einfügen (Rennbedingungen).
• lock_timeout und statement_timeout für App-Rollen setzen, damit eine schlechte Abfrage nicht alles blockieren kann.
• Deadlocks entstehen durch Sperren von Zeilen in unterschiedlicher Reihenfolge; in einer konsistenten Reihenfolge sperren (nach ID sortieren).

Alembic und Änderungen ohne Ausfallzeit

Automatisch generierte Migrationen immer überprüfen; Autogenerate übersieht Umbenennungen (es generiert Drop und Add) sowie einige Typänderungen.

Expand and Contract für breaking changes:

1. Expand: neue Spalte nullable hinzufügen (oder mit konstantem Default, was auf PG 11+ sofort geht), neue Tabelle, neuen Index CONCURRENTLY.
2. Code deployen, der sowohl alt als auch neu schreibt.
3. Backfill in Batches (z. B. 5k Zeilen pro Transaktion) außerhalb der Migration oder in einer separaten Datenmigration mit Fortschrittsprotokollierung.
4. Code deployen, der neu liest.
5. Contract: alte Spalte in einem späteren Release löschen.

Gefährliche Operationen auf großen Live-Tabellen:

Operation	Sicherer Ansatz
CREATE INDEX	CREATE INDEX CONCURRENTLY (Alembic: postgresql_concurrently=True innerhalb von with op.get_context().autocommit_block():)
NOT NULL hinzufügen	CHECK (col IS NOT NULL) NOT VALID hinzufügen, VALIDATE CONSTRAINT, dann NOT NULL setzen
Fremdschlüssel hinzufügen	mit NOT VALID hinzufügen, dann VALIDATE CONSTRAINT
Spalte umbenennen	Expand and Contract, niemals direkt, während alter Code läuft
Spaltentyp ändern	Neue Spalte plus Backfill, nicht ALTER TYPE (schreibt Tabelle neu)

• SET lock_timeout = '5s' in Migrationen setzen, damit sie schnell fehlschlagen, anstatt sich hinter langen Transaktionen anzustellen.
• Jede Migration hat ein funktionierendes downgrade() oder einen schriftlichen Grund, warum es nicht möglich ist.
• Migrationen auf einer Kopie realistischer Daten testen (up, down und wieder up).

Checkliste

• timestamptz, NOT NULL-Defaults, benannte Constraints, indizierte Fremdschlüssel.
• Abfragen parametrisiert; keine f-Strings in SQL.
• Kein N+1; Beziehungen explizit geladen.
• Langsame Abfragen mit EXPLAIN (ANALYZE, BUFFERS) auf realistischen Daten überprüft.
• Transaktionen kurz; Timeouts gesetzt.
• Migration überprüft, rückgängig machbar, wo nötig CONCURRENTLY / NOT VALID verwendet.
• Backfills gebatched und fortsetzbar.

Häufige Fehler

• OFFSET 100000-Paginierung; Keyset auf einem indizierten Sortierschlüssel verwenden.
• WHERE date(created_at) = ? zerstört den Index; Bereich auf der rohen Spalte verwenden.
• LIKE '%term%' ohne Trigram-Index.
• await session.commit() vergessen oder in einer Schleife pro Zeile committen.
• alembic revision --autogenerate ausführen und ungelesen committen.
• Eine Transaktion offen halten, während eine externe API aufgerufen wird.
• Zeitstempel in lokaler Zeit speichern.