"""Alembic-Umgebung. Nutzt die Engine und die Metadaten der Anwendung (SQLite), damit Migrationen und Laufzeit garantiert dieselbe Datenbank/dasselbe Schema verwenden. ``render_as_batch`` ist für SQLite Pflicht, damit ALTER-TABLE über Tabellen-Kopien läuft. """ from logging.config import fileConfig from alembic import context from app.database import Base, engine from app import models # noqa: F401 – registriert alle Tabellen an Base.metadata config = context.config if config.config_file_name is not None: fileConfig(config.config_file_name) target_metadata = Base.metadata def run_migrations_offline() -> None: context.configure( url=str(engine.url), target_metadata=target_metadata, literal_binds=True, dialect_opts={"paramstyle": "named"}, render_as_batch=True, compare_type=True, ) with context.begin_transaction(): context.run_migrations() def run_migrations_online() -> None: with engine.connect() as connection: # Fremdschlüssel für die Dauer der Migration abschalten. # # ``render_as_batch`` baut eine Tabelle um, indem es eine Kopie anlegt und # das Original löscht. SQLite führt bei DROP TABLE ein implizites DELETE # aus – mit aktiven Fremdschlüsseln schlägt das per ON DELETE CASCADE auf # abhängige Tabellen durch. Ein zusätzliches Feld an ``categories`` löschte # so sämtliche Schlagwörter und Budget-Vorlagen und setzte die Kategorie # aller Buchungen auf NULL. # # Der Befehl muss VOR der Transaktion laufen, sonst ignoriert SQLite ihn. roh = connection.connection roh.rollback() roh.execute("PRAGMA foreign_keys=OFF") aktiv = roh.execute("PRAGMA foreign_keys").fetchone()[0] if aktiv: raise RuntimeError( "Fremdschlüssel liessen sich nicht abschalten – Migration abgebrochen, " "sonst droht Datenverlust beim Tabellen-Umbau." ) try: context.configure( connection=connection, target_metadata=target_metadata, render_as_batch=True, compare_type=True, ) with context.begin_transaction(): context.run_migrations() finally: roh.rollback() roh.execute("PRAGMA foreign_keys=ON") if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()