Alembic Cheatsheet 07 — Multi-Database and Multi-Tenant

alembic init -t multidb migrations

[alembic]
databases = users, analytics

[users]
sqlalchemy.url = postgresql://...
script_location = migrations/users

[analytics]
sqlalchemy.url = postgresql://...
script_location = migrations/analytics

alembic --name users upgrade head
alembic --name analytics upgrade head

# env.py
def run_migrations_online():
    schemas = get_all_tenant_schemas()
    
    connectable = create_engine(...)
    for schema in schemas:
        with connectable.connect() as conn:
            conn.execute(text(f"SET search_path TO {schema}"))
            context.configure(
                connection=conn,
                target_metadata=target_metadata,
                version_table_schema=schema,        # per-schema version table
            )
            with context.begin_transaction():
                context.run_migrations()
    connectable.dispose()

def run_migrations_online():
    for tenant in get_all_tenants():
        connectable = create_engine(tenant.dsn)
        with connectable.connect() as conn:
            context.configure(connection=conn, target_metadata=target_metadata)
            with context.begin_transaction():
                context.run_migrations()
        connectable.dispose()

import concurrent.futures
from alembic.config import Config
from alembic import command

def migrate_one(tenant):
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", tenant.dsn)
    command.upgrade(cfg, "head")

with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
    list(ex.map(migrate_one, tenants))

async def onboard(tenant_slug: str):
    schema = f"tenant_{tenant_slug}"
    
    # 1. Create schema
    async with engine.connect() as conn:
        await conn.execute(text(f"CREATE SCHEMA {schema}"))
    
    # 2. Apply current head
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", DATABASE_URL)
    # ... configure for this schema ...
    command.upgrade(cfg, "head")
    
    # 3. Register
    await register_tenant(tenant_slug, schema)

async with engine.begin() as conn:
    await conn.execute(text(f"SET search_path TO {schema}"))
    await conn.run_sync(Base.metadata.create_all)

command.stamp(cfg, "head")    # mark as up-to-date

async def onboard_db_tenant(slug):
    # 1. Provision DB
    dsn = f"postgresql+asyncpg://app:pass@host/{slug}"
    # ... call infra to create DB ...
    
    # 2. Run migrations
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", dsn)
    await asyncio.to_thread(command.upgrade, cfg, "head")
    
    # 3. Register
    await register_tenant(slug, dsn)

- run: alembic --name users upgrade head
- run: alembic --name analytics upgrade head
- run: pytest tests/

def test_no_drift_users(users_engine):
    from alembic.autogenerate import compare_metadata
    from alembic.migration import MigrationContext
    
    with users_engine.connect() as conn:
        ctx = MigrationContext.configure(conn, opts={"version_table": "alembic_version"})
        diff = compare_metadata(ctx, UsersBase.metadata)
        assert diff == [], f"users drift: {diff}"