Running Alembic migrations programmatically from async code

Open a connection from your async engine, call await conn.run_sync(run_upgrade, cfg) where run_upgrade stores the connection in cfg.attributes["connection"] and calls command.upgrade(cfg, "head"), and teach env.py to use that connection instead of calling asyncio.run() — which cannot start inside an already-running loop. This guide belongs to configuring Alembic with async SQLAlchemy engines.

Quick Answer

The async env.py template ends with asyncio.run(run_async_migrations()). That works from the alembic command line and fails from inside an application that is already running an event loop.

Before — calling the command API from a FastAPI lifespan:

from contextlib import asynccontextmanager

from alembic import command
from alembic.config import Config
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    command.upgrade(Config("alembic.ini"), "head")
    yield

app = FastAPI(lifespan=lifespan)
# RuntimeError: asyncio.run() cannot be called from a running event loop

After — share an async connection with Alembic:

# shop/migrate.py
from contextlib import asynccontextmanager
from pathlib import Path

from alembic import command
from alembic.config import Config
from fastapi import FastAPI

from shop.db import engine

ROOT = Path(__file__).resolve().parent.parent


def alembic_config() -> Config:
    cfg = Config(str(ROOT / "alembic.ini"))
    cfg.set_main_option("script_location", str(ROOT / "alembic"))
    return cfg


def _upgrade(connection, cfg: Config) -> None:
    cfg.attributes["connection"] = connection
    command.upgrade(cfg, "head")


async def upgrade_to_head() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(_upgrade, alembic_config())


@asynccontextmanager
async def lifespan(app: FastAPI):
    await upgrade_to_head()
    yield
    await engine.dispose()

# alembic/env.py (the end of the file)
import asyncio

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

from shop.models import Base

config = context.config


def do_run_migrations(connection) -> None:
    context.configure(connection=connection, target_metadata=Base.metadata,
                      transaction_per_migration=True)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.", poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()


def run_migrations_online() -> None:
    connection = config.attributes.get("connection")
    if connection is None:
        asyncio.run(run_async_migrations())      # CLI: no loop is running
    else:
        do_run_migrations(connection)            # application: connection provided


if context.is_offline_mode():
    raise SystemExit("offline mode not configured")
run_migrations_online()

Execution Context & Async Workflow Integration

Alembic's command API — command.upgrade(), command.downgrade(), command.current() — is synchronous. It loads env.py as a script and lets it decide how to connect. The async template makes that decision by creating an async engine and driving it with asyncio.run(), which creates a new event loop, runs the coroutine, and closes the loop. From the CLI there is no loop yet, so this is fine. Inside FastAPI's lifespan, a pytest-asyncio test, or any async def, a loop is already running in the thread, and asyncio.run() refuses to nest one.

config.attributes is Alembic's supported channel for passing objects from the caller into env.py. It is a plain dictionary on the Config object, and anything placed there before command.upgrade() is visible to env.py as context.config.attributes. Placing a connection there lets env.py skip engine creation entirely.

The connection must be synchronous-facing, because Alembic's operations are synchronous. AsyncConnection.run_sync() supplies exactly that: it runs a plain function inside SQLAlchemy's greenlet bridge and passes it a Connection facade whose calls are carried out by the async driver underneath. Everything Alembic does inside _upgrade() — reading alembic_version, running DDL, autocommit_block() — travels through asyncpg on the application's own engine.

Using engine.begin() rather than engine.connect() matters for how transactions nest. engine.begin() opens an outer transaction and commits it when the block exits. Inside, context.begin_transaction() sees that the connection is already in a transaction and does not start one of its own, so every pending migration runs inside the transaction engine.begin() opened. The net effect is that migrations commit together when upgrade_to_head() returns, and a failure rolls all of them back — unless a revision uses an autocommit block, which commits what came before it, as described in creating indexes concurrently in Alembic migrations.

The other way around the error is a thread: await asyncio.to_thread(command.upgrade, cfg, "head") runs the unchanged env.py in a worker thread, where asyncio.run() can create its own loop. It works, and it needs no env.py changes, but it creates a second engine and a second event loop for the duration, and anything that expects to share the application's connection or configuration will not. The shared-connection pattern is the one Alembic documents for asyncio applications, and it is the one to prefer. The base env.py it extends is explained in setting up Alembic env.py for asyncpg.

Resolving Warnings, Errors & Common Mistakes

The logging row is easy to miss because nothing fails — logs simply change format, or stop, once the lifespan runs. The default env.py configures logging from alembic.ini at import time, which is right for the CLI and wrong for an application that has already configured logging:

from logging.config import fileConfig

from alembic import context

config = context.config
if config.config_file_name is not None and "connection" not in config.attributes:
    fileConfig(config.config_file_name, disable_existing_loggers=False)

The percent-sign row matters for anyone building sqlalchemy.url from environment variables. Config is backed by configparser, which treats % as interpolation syntax, and URL-encoded passwords are full of them. Setting the URL with cfg.set_main_option("sqlalchemy.url", url.replace("%", "%%")) works; passing an engine or connection through cfg.attributes, as above, avoids the URL entirely.

Advanced: Replicas, Advisory Locks and Verifying the Head

Running migrations at application startup is convenient for a single instance and dangerous for several. Three replicas starting together each read alembic_version, each see the same pending revision, and each start applying it. Depending on the revision, the losers fail on a duplicate key in alembic_version, on relation already exists, or worse, succeed at a non-idempotent data migration twice.

A PostgreSQL advisory lock serialises them. Take it on the same connection that runs the migrations, so it is released automatically if the process dies:

from sqlalchemy import text

from shop.db import engine
from shop.migrate import _upgrade, alembic_config

MIGRATION_LOCK_ID = 7_214_553_901  # any stable 64-bit integer unique to this application


async def upgrade_to_head_once() -> None:
    async with engine.connect() as conn:
        await conn.execute(text("SELECT pg_advisory_lock(:id)"), {"id": MIGRATION_LOCK_ID})
        # End the transaction that execute() began; the session-level lock stays held.
        await conn.commit()
        try:
            async with conn.begin():
                await conn.run_sync(_upgrade, alembic_config())
        finally:
            await conn.execute(text("SELECT pg_advisory_unlock(:id)"), {"id": MIGRATION_LOCK_ID})
            await conn.commit()

The first replica migrates; the others wait on the lock, then find nothing to do. Their startup is delayed by the migration's duration, which is the argument for the alternative: run migrations once, as a separate step in the deployment — a Kubernetes Job or init container, or a platform's release phase — and have the application only verify that the schema is current.

Verification is cheap and needs no DDL privileges, so the application can run with a role that cannot alter the schema at all:

import logging

from alembic.runtime.migration import MigrationContext
from alembic.script import ScriptDirectory

from shop.db import engine
from shop.migrate import alembic_config

log = logging.getLogger("startup")


def _current_heads(connection) -> set[str]:
    return set(MigrationContext.configure(connection).get_current_heads())


async def database_is_at_head() -> bool:
    expected = set(ScriptDirectory.from_config(alembic_config()).get_heads())
    async with engine.connect() as conn:
        current = await conn.run_sync(_current_heads)
    if current != expected:
        log.error("schema not at head", extra={"expected": sorted(expected),
                                               "current": sorted(current)})
        return False
    return True

Call it from the readiness probe or the lifespan, and refuse to serve if it returns False. A deploy that forgot to migrate then stalls visibly at readiness instead of throwing UndefinedColumnError at users. The broader ordering of migration and application releases is covered in running Alembic migrations in CI/CD pipelines.

Running Migrations From Tests

Test suites are the other common caller of the command API, and they hit the same event-loop error as applications: a session-scoped async fixture that wants to create the schema is already inside pytest-asyncio's loop.

The shared-connection pattern works unchanged. A session-scoped fixture creates an engine pointing at the test database, upgrades it to head through run_sync, and yields the engine to the rest of the suite:

import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine

from shop.migrate import _upgrade, alembic_config


@pytest_asyncio.fixture(scope="session")
async def migrated_engine(postgres_url: str):
    engine = create_async_engine(postgres_url)
    async with engine.begin() as conn:
        await conn.run_sync(_upgrade, alembic_config())
    yield engine
    await engine.dispose()

Using migrations rather than Base.metadata.create_all() to build the test schema is a deliberate choice. create_all() builds the schema the models describe; migrations build the schema production will have. The two drift apart exactly when something has gone wrong — a missing migration, an enum label added only in Python, an index defined in a migration but not in the model — and a suite built with create_all() passes straight through that drift. A suite built from migrations catches it on the first test that touches the difference.

The same fixture is the natural place for a downgrade round trip, which proves that each revision's downgrade() actually works:

from alembic import command


def _downgrade_base_then_upgrade(connection, cfg) -> None:
    cfg.attributes["connection"] = connection
    command.downgrade(cfg, "base")
    command.upgrade(cfg, "head")

Run that in a separate, slower CI job rather than every test session. Downgrades that drop data will, by design, drop the data other tests created, and a round trip through every revision grows with the migration history. The test-database fixtures it builds on — containers, per-test rollback — are covered in running tests against a Postgres testcontainer.

Frequently Asked Questions

Related

• Configuring Alembic with Async SQLAlchemy Engines — The parent guide: async env.py structure and configuration.
• Setting up Alembic env.py for asyncpg — The env.py this guide extends.
• Running Alembic migrations in CI/CD pipelines — Migrating as a release step instead of at startup.
• Setting up an async engine from scratch — The lifespan that owns the engine migrations borrow.