SQLAlchemy 2 + Alembic programmatic migrations: working versions (tested 2026-10-15)

sqlalchemy==2.1.4
alembic==1.20.0
pytest==9.1.1

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = "account"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)

from alembic import context
from sqlalchemy import engine_from_config, pool
from models import Base

config = context.config
connectable = engine_from_config(config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=pool.NullPool)
with connectable.connect() as connection:
    context.configure(connection=connection, target_metadata=Base.metadata, render_as_batch=True)
    with context.begin_transaction():
        context.run_migrations()

"""${message}"""
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
from alembic import op
import sqlalchemy as sa
def upgrade():
    ${upgrades if upgrades else "pass"}
def downgrade():
    ${downgrades if downgrades else "pass"}

"""create account"""
revision = "0001"
down_revision = None
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table("account", sa.Column("id", sa.Integer, primary_key=True), sa.Column("email", sa.String(255), nullable=False, unique=True))

def downgrade():
    op.drop_table("account")

from alembic import command
from alembic.config import Config
from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from sqlalchemy import create_engine, inspect
from models import Base

def cfg(url):
    c = Config()
    c.set_main_option("script_location", "migrations")
    c.set_main_option("sqlalchemy.url", url)
    return c

def test_upgrade_downgrade(tmp_path):
    url = f"sqlite:///{tmp_path}/db.sqlite"
    command.upgrade(cfg(url), "head")
    assert "account" in inspect(create_engine(url)).get_table_names()
    command.downgrade(cfg(url), "base")
    assert "account" not in inspect(create_engine(url)).get_table_names()

def test_models_match_migrations(tmp_path):
    url = f"sqlite:///{tmp_path}/db.sqlite"
    command.upgrade(cfg(url), "head")
    with create_engine(url).connect() as conn:
        diff = compare_metadata(MigrationContext.configure(conn), Base.metadata)
    assert diff == []

[pytest]
pythonpath = .