"""Alembic migration environment configuration. This module configures Alembic to work with the Flask application and SQLAlchemy models. """ import os from logging.config import fileConfig from pathlib import Path from alembic import context from flask import Flask from flask_sqlalchemy import SQLAlchemy # this is the Alembic Config object, which provides # access to the values within the .ini file in use. config = context.config # Interpret the config file for Python logging. # This line sets up loggers basically. if config.config_file_name is not None: fileConfig(config.config_file_name) # Create minimal Flask app for migrations (without session initialization) # This avoids Flask-Session trying to create tables before migrations run app = Flask(__name__) BASE_DIR = Path(__file__).parent.parent DATA_DIR = BASE_DIR / "data" DATA_DIR.mkdir(parents=True, exist_ok=True) app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get( "DATABASE_URL", f"sqlite:///{DATA_DIR / 'sneaky-klaus.db'}" ) app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False db = SQLAlchemy() db.init_app(app) # Import all models to register them with SQLAlchemy metadata with app.app_context(): from src.models import ( # noqa: F401 Admin, Exchange, MagicToken, Participant, RateLimit, ) config.set_main_option("sqlalchemy.url", app.config["SQLALCHEMY_DATABASE_URI"]) # Add your model's MetaData object here for 'autogenerate' support target_metadata = db.metadata def run_migrations_offline() -> None: """Run migrations in 'offline' mode. This configures the context with just a URL and not an Engine, though an Engine is acceptable here as well. By skipping the Engine creation we don't even need a DBAPI to be available. Calls to context.execute() here emit the given string to the script output. """ url = config.get_main_option("sqlalchemy.url") context.configure( url=url, target_metadata=target_metadata, literal_binds=True, dialect_opts={"paramstyle": "named"}, ) with context.begin_transaction(): context.run_migrations() def run_migrations_online() -> None: """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ # Use the Flask app's engine instead of creating a new one with app.app_context(): connectable = db.engine with connectable.connect() as connection: context.configure(connection=connection, target_metadata=target_metadata) with context.begin_transaction(): context.run_migrations() if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()