Alembic env.py Structure (Example)

This page provides a generic alembic/env.py example and explains the key responsibilities of the file.

Why env.py matters

env.py is the entry point Alembic uses to:

• pick the correct database URL for the environment you are migrating
• load your SQLAlchemy metadata (Base.metadata) so --autogenerate can diff models vs the live DB
• configure offline vs online migration execution

Key rules

• Import all models for autogenerate: target_metadata must include every table/enum/index you want Alembic to manage.
• Prefer “import modules” over “import classes”: importing modules (e.g. from database.users import models) ensures tables are registered without turning env.py into a long list of class imports.
• Avoid async engines in Alembic: use a normal synchronous SQLAlchemy engine in env.py.
• Keep DB URL selection predictable: standardize env var naming and selection logic (dev/staging/prod).
• No secrets in code: URLs/credentials come from environment variables.

Recommended model import pattern

Create a single function in your database/ package that imports all model modules, then call it from env.py.

Example:

• database/imports.py contains import_all_models()
• alembic/env.py calls import_all_models() once, then sets target_metadata = Base.metadata

Example alembic/env.py

from __future__ import annotations

import os
from logging.config import fileConfig

from alembic import context
from sqlalchemy import create_engine, pool

# Optional: load dotenv for local usage (keep override=False to avoid clobbering CI/env vars)
try:
    from dotenv import load_dotenv

    load_dotenv(override=False)
except Exception:
    pass

# Alembic Config object (read from alembic.ini)
config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)


def get_environment() -> str:
    """
    Decide which environment you are running migrations against.
    Prefer OS env vars so wrapper scripts can set this explicitly.
    """
    return os.environ.get("ENVIRONMENT", os.getenv("ENVIRONMENT", "development"))


def get_database_url(env: str) -> str:
    """
    Select the correct database URL for the environment.
    Standardize these names across projects.
    """
    key_by_env = {
        "local": "DATABASE_URL_LOCAL",
        "development": "DATABASE_URL_DEV",
        "staging": "DATABASE_URL_STAGING",
        "production": "DATABASE_URL_PROD",
    }

    key = key_by_env.get(env, "DATABASE_URL_DEV")
    url = os.environ.get(key) or os.getenv(key)
    if not url:
        raise ValueError(f"{key} is not set for ENVIRONMENT={env}")
    return url


# Import Base + ensure all model modules are imported so Base.metadata is complete.
from database.database import Base  # noqa: E402

try:
    # Prefer a single “import all models” helper.
    from database.imports import import_all_models  # noqa: E402

    import_all_models()
except Exception:
    # Fallback: import model modules directly (project-specific list).
    # Example (do not copy this list blindly):
    # from database.users import models as users_models
    # from database.items import models as items_models
    pass

target_metadata = Base.metadata

ENVIRONMENT = get_environment()
DATABASE_URL = get_database_url(ENVIRONMENT)


def run_migrations_offline() -> None:
    """
    Offline mode: configures Alembic with a URL only (no DB connection).
    Generates SQL scripts instead of applying them.
    """
    context.configure(
        url=DATABASE_URL,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """
    Online mode: connects to the DB and runs migrations.
    Use a synchronous engine for Alembic.
    """
    engine = create_engine(DATABASE_URL, poolclass=pool.NullPool)

    with engine.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()