Alembic Cheatsheet 01 — Setup and env.py

uv add alembic
alembic init -t async migrations         # async template
# or
alembic init migrations                  # sync

migrations/
├── env.py
├── README
├── script.py.mako
└── versions/
alembic.ini

[alembic]
script_location = migrations
sqlalchemy.url =                # set from env.py
file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
transaction_per_migration = false  # CONCURRENTLY ops won't be in transaction

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

# Import models so Base.metadata is populated
import src.app.models  # noqa

from src.app.db import Base
from src.app.settings import settings

config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)

target_metadata = Base.metadata

def do_run_migrations(connection):
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
        compare_server_default=True,
        include_schemas=False,
    )
    with context.begin_transaction():
        context.run_migrations()

async def run_migrations_online():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
    )
    async with connectable.connect() as conn:
        await conn.run_sync(do_run_migrations)
    await connectable.dispose()

if context.is_offline_mode():
    raise NotImplementedError("offline mode not supported")
else:
    asyncio.run(run_migrations_online())

from alembic import context
from sqlalchemy import engine_from_config, pool

from src.app.db import Base
import src.app.models  # noqa

config = context.config
target_metadata = Base.metadata

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as conn:
        context.configure(connection=conn, target_metadata=target_metadata, compare_type=True)
        with context.begin_transaction():
            context.run_migrations()

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

NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)

config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"])

alembic current        # current revision
alembic heads          # head revisions
alembic history        # all revisions

alembic revision --autogenerate -m "initial schema"
# review the file
alembic upgrade head