obsidian/wiki/shared-patterns/alembic-migrations.md

title	aliases	tags	sources	created	updated
Alembic migrations pattern	alembic db migrations	python alembic postgresql shared-pattern	cc-dashboard/alembic/env.py cc-dashboard/alembic/alembic.ini cc-dashboard/alembic/versions/0001_initial.py cc-dashboard/alembic/versions/0002_project_metadata.py cc-dashboard/alembic/versions/0003_corporate.py	2026-05-15	2026-05-15

Key Takeaways

• All Oliver projects use async SQLAlchemy (asyncpg) — env.py must use create_async_engine + asyncio.run(), not the synchronous default template.
• Migration filenames use NNNN_<description>.py numbering (e.g. 0001_initial.py) instead of UUID hashes — easier to read in git log and order of application is obvious.
• sqlalchemy.url in alembic.ini is a dummy placeholder; actual URL is injected from settings.DATABASE_URL in env.py. Never commit real credentials into alembic.ini.
• Always write a downgrade() that is the exact mirror of upgrade() — drop indexes before tables, drop columns before constraints that depend on them.
• Import all models via a single import src.models # noqa line in env.py to ensure Base.metadata is fully populated before autogenerate.

Setup Commands

# 1. Init (run once per project, then commit the generated alembic/ dir)
alembic init alembic

# 2. Generate a new migration (autogenerate from model diff)
alembic revision --autogenerate -m "add user table"

# 3. Apply all pending migrations
alembic upgrade head

# 4. Apply exactly one step forward
alembic upgrade +1

# 5. Rollback one step
alembic downgrade -1

# 6. Rollback to specific revision
alembic downgrade 0002

# 7. Rollback everything
alembic downgrade base

# 8. Show current revision applied to DB
alembic current

# 9. Show full migration history
alembic history --verbose

# 10. Show pending (not yet applied) migrations
alembic history -r current:head

Inside Docker (Oliver standard):

docker compose exec api alembic upgrade head
docker compose exec api alembic downgrade -1

alembic.ini Key Settings

Real file from cc-dashboard (alembic/alembic.ini):

[alembic]
script_location = alembic
prepend_sys_path = .
# Real URL is set in env.py via settings.DATABASE_URL — this is a dummy
sqlalchemy.url = driver://user:pass@localhost/dbname

[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

Key points:

• script_location = alembic — the alembic/ directory is a sibling of src/; run alembic from the project root.
• prepend_sys_path = . — allows from src.config import settings to resolve inside env.py.
• sqlalchemy.url left as dummy — overridden programmatically in env.py.

env.py Pattern

Full async pattern from cc-dashboard (alembic/env.py):

import asyncio
import os
import sys
from logging.config import fileConfig

from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from src.config import settings
from src.database import Base
import src.models  # noqa — registers all models with 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():
    context.configure(
        url=settings.DATABASE_URL,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations():
    engine = create_async_engine(settings.DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(
            lambda conn: context.configure(
                connection=conn,
                target_metadata=target_metadata,
                compare_type=True,   # detect column type changes in autogenerate
            )
        )
        await conn.run_sync(lambda conn: context.run_migrations())
    await engine.dispose()


def run_migrations_online():
    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

compare_type=True — critical: without it, autogenerate ignores column type changes (e.g. String(100) → String(255)).

Migration Patterns

Create table (from 0001_initial.py — cc-dashboard):

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import JSONB, UUID

def upgrade():
    op.create_table(
        "users",
        sa.Column("id", UUID(as_uuid=False), primary_key=True),
        sa.Column("email", sa.String(255), nullable=False, unique=True),
        sa.Column("role", sa.Enum("admin", "user", name="user_role"),
                  nullable=False, server_default="user"),
        sa.Column("is_active", sa.Boolean, nullable=False, server_default="true"),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
    )
    op.create_index("ix_users_email", "users", ["email"])

def downgrade():
    op.drop_table("users")
    op.execute("DROP TYPE user_role")  # custom Enum type must be dropped separately

Add column (from 0002_project_metadata.py — cc-dashboard):

def upgrade():
    op.add_column("projects", sa.Column("client",     sa.String(255), nullable=False, server_default=""))
    op.add_column("projects", sa.Column("job_number", sa.String(100), nullable=False, server_default=""))
    op.add_column("projects", sa.Column("repo_url",   sa.String(500), nullable=False, server_default=""))

def downgrade():
    op.drop_column("projects", "repo_url")
    op.drop_column("projects", "job_number")
    op.drop_column("projects", "client")

Add column with FK + index (from 0003_corporate.py):

def upgrade():
    op.add_column(
        "sessions",
        sa.Column("task_id", postgresql.UUID(as_uuid=False),
                  sa.ForeignKey("tasks.id", ondelete="SET NULL"), nullable=True),
    )
    op.create_index("ix_sessions_task_id", "sessions", ["task_id"])

def downgrade():
    op.drop_index("ix_sessions_task_id", "sessions")
    op.drop_column("sessions", "task_id")

Drop constraint / rename:

def upgrade():
    op.drop_constraint("uq_project_user_slug", "projects", type_="unique")
    op.create_unique_constraint("uq_project_slug", "projects", ["slug"])

def downgrade():
    op.drop_constraint("uq_project_slug", "projects", type_="unique")
    op.create_unique_constraint("uq_project_user_slug", "projects", ["user_id", "slug"])

JSONB column with default:

sa.Column("commits", JSONB, server_default="[]")
sa.Column("tools_used", JSONB, server_default="{}")

Composite unique constraint (named — required for drop_constraint in downgrade):

sa.UniqueConstraint("user_id", "session_id", "date", name="uq_session_user_date")

Gotchas

1. --autogenerate misses nullable-only changes without compare_type=True. Changing nullable=True → nullable=False on an existing column is not detected by autogenerate unless compare_type=True is set in context.configure(). Always include it. Even then, check generated migration by hand — autogenerate is a starting point, not a final answer.

2. Custom Enum types (sa.Enum(...)) must be dropped manually in downgrade(). SQLAlchemy creates a named PostgreSQL TYPE for sa.Enum. op.drop_table() does NOT drop the type — doing so leaves orphaned types that block recreating the table. Always add op.execute("DROP TYPE type_name") after drop_table.

3. server_default vs Python default. server_default="true" is a SQL-level default (string literal for Postgres). default=True is an SQLAlchemy ORM-level default applied only when inserting via ORM, not via raw SQL or future migrations. For migrations, always use server_default so the DB itself handles it.

4. Async env.py and alembic current / alembic history commands. These commands do not actually run migrations — they work offline. But alembic upgrade head requires a running database. In CI, ensure the DB container is healthy before running migrations: docker compose up -d db && sleep 3 && alembic upgrade head or use a proper healthcheck.

5. Revision ordering with numeric prefixes. Using 0001_, 0002_ prefixes makes alembic history human-readable but the actual ordering is determined by down_revision chain, not the filename. A file named 0005_ with down_revision = "0003" skips 0004 in the chain — this is valid but confusing. Keep filenames in sync with chain order.

Related

• wiki/tech-patterns/fastapi-python-docker
• wiki/shared-patterns/pydantic-settings-env