""" Alembic environment script for database migrations """ import sys from logging.config import fileConfig from pathlib import Path from alembic import context # Custom module to help with finding modules in containerized environments from alembic.find_modules import setup_python_path from sqlalchemy import engine_from_config, pool # Setup Python path to find app modules in various environments setup_python_path() # 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) # Try different import strategies to handle various deployment environments try: # Standard import approach from app.models import Todo # noqa from app.core.database import Base # noqa except ImportError: try: # Path-based import for some container setups import importlib.util import sys # Try to find the module file for base_path in sys.path: model_path = Path(base_path) / "app" / "models" / "todo.py" db_path = Path(base_path) / "app" / "core" / "database.py" if model_path.exists() and db_path.exists(): # Load the todo module spec = importlib.util.spec_from_file_location("todo", model_path) todo_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(todo_module) # Load the database module spec = importlib.util.spec_from_file_location("database", db_path) db_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(db_module) # Get the required objects Todo = getattr(todo_module, "Todo") Base = getattr(db_module, "Base") break else: raise ImportError("Could not find app modules in any path") except Exception as e: print(f"Error importing models: {e}") print(f"Current sys.path: {sys.path}") raise target_metadata = Base.metadata def run_migrations_offline(): """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(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ connectable = engine_from_config( config.get_section(config.config_ini_section), prefix="sqlalchemy.", poolclass=pool.NullPool, ) with connectable.connect() as connection: is_sqlite = connection.dialect.name == 'sqlite' context.configure( connection=connection, target_metadata=target_metadata, render_as_batch=is_sqlite, # Key configuration for SQLite ) with context.begin_transaction(): context.run_migrations() if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()