Setting Up Alembic env.py for asyncpg

Setting up Alembic's env.py for asyncpg requires replacing the default synchronous engine with an AsyncEngine and bridging Alembic's synchronous migration context to the async runtime via conn.run_sync(). This page is a focused deep-dive within the broader Configuring Alembic with Async SQLAlchemy Engines guide — start there for the end-to-end setup, then use this page for the exact env.py wiring.

Quick Answer

The default env.py generated by alembic init is built for synchronous drivers. Below is a direct comparison of the legacy pattern and the correct async version for asyncpg.

Legacy sync env.py (do not use with asyncpg)

# env.py — synchronous, not compatible with asyncpg
from sqlalchemy import engine_from_config, pool
from alembic import context

config = context.config

def run_migrations_online() -> None:
    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=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

Modern async env.py for asyncpg

# env.py — async-compatible with asyncpg
import asyncio
import os
from logging.config import fileConfig

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import pool

from alembic import context

# Import your declarative Base so autogenerate can inspect metadata
from app.models import Base  # User, Order, Product, Invoice, Tenant models live here

config = context.config
target_metadata = Base.metadata

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://user:pass@localhost:5432/appdb")


def run_migrations_offline() -> None:
    """Generate SQL script without a live database connection."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection) -> None:
    """Synchronous callback executed inside run_sync()."""
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """Create an async engine and delegate to the sync callback via run_sync."""
    connectable = create_async_engine(DATABASE_URL, poolclass=pool.NullPool)

    async with connectable.connect() as conn:
        await conn.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    """Entry point for online migration mode."""
    asyncio.run(run_async_migrations())


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

Execution Context & Async Workflow Integration

Alembic's migration context was designed around synchronous DBAPI connections. Every internal operation — schema comparison, DDL emission, transaction management — calls blocking methods on a raw connection object. asyncpg, on the other hand, exposes only coroutine-based methods and refuses to be driven from a synchronous call stack without an event loop.

The bridge is AsyncConnection.run_sync(). When you call await conn.run_sync(do_run_migrations), SQLAlchemy creates a temporary synchronous shim that passes a wrapped connection object to your callback. Inside do_run_migrations, all operations appear synchronous to Alembic, but the underlying I/O is dispatched through the running event loop. The key rule is that run_sync must be called inside an active async with connectable.connect() block — the synchronous connection shim is only valid for the lifetime of that async context manager.

The asyncio.run() call in run_migrations_online() is the correct entry point. It creates a fresh event loop, runs the coroutine to completion, and tears the loop down. This matches how the async engine and connection pool lifecycle is designed to work: each migration run is a discrete, self-contained process, not a long-lived application server.

Using pool.NullPool is intentional. Migrations run as a CLI process, not a persistent service. Connection pooling would leave dangling connections when the process exits, potentially blocking other operations on short-lived database server configurations.

Reading the DATABASE_URL from the environment with os.getenv() rather than hardcoding it in alembic.ini is a production requirement. Secrets stored in INI files end up in version control; environment variables are injected by your deployment platform. The online migration path uses the env var exclusively. Note that the offline path still reads config.get_main_option("sqlalchemy.url") from alembic.ini — the implications of that asymmetry are covered in the error table below.

Resolving Warnings, Errors & Common Mistakes

Advanced Async env.py Optimization

Sharing configuration between the application engine and migrations

A common pattern is to centralize engine configuration in a single module, then import it in both the application and env.py. The naive approach — importing the application's AsyncEngine instance directly — causes the "different loop" error described above because the application engine is often created at module import time, before any event loop exists.

The correct approach is to share the configuration rather than the engine object. Extract a get_engine_kwargs() factory:

# app/database.py
import os
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import pool

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://user:pass@localhost:5432/appdb")

def get_engine_kwargs() -> dict:
    """Return engine configuration shared by the app and Alembic migrations."""
    return {
        "url": DATABASE_URL,
        "pool_size": 10,
        "max_overflow": 5,
        "pool_pre_ping": True,
        "connect_args": {
            "server_settings": {"application_name": "myapp"},
            "statement_cache_size": 0,  # required for pgbouncer compatibility
        },
    }

# Application uses a long-lived pooled engine
engine = create_async_engine(**get_engine_kwargs())
SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Then in env.py, build a migration-specific engine from the same configuration dict:

# env.py — advanced version sharing app configuration
import asyncio
import os
from logging.config import fileConfig
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import pool
from alembic import context

from app.models import Base
from app.database import get_engine_kwargs  # shared config, not the engine object

config = context.config
target_metadata = Base.metadata

if config.config_file_name is not None:
    fileConfig(config.config_file_name)


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


async def run_async_migrations() -> None:
    # Override pool settings for migration context
    kwargs = get_engine_kwargs()
    kwargs["poolclass"] = pool.NullPool
    kwargs.pop("pool_size", None)
    kwargs.pop("max_overflow", None)

    connectable = create_async_engine(**kwargs)
    async with connectable.connect() as conn:
        await conn.run_sync(do_run_migrations)
    await connectable.dispose()


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())


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

This pattern ensures that SSL certificates, connect_args, and custom server settings (critical for PgBouncer environments serving User, Order, and Invoice tables with statement caching disabled) remain consistent between the running application and the migration process, without ever sharing an event-loop-bound engine object across boundaries.

When autogenerating migration scripts for schemas that span multiple models — for example when adding a tenant_id column to the products table — this configuration consistency ensures Alembic compares the live schema against the same dialect options the application uses. Discrepancies between connection settings at migration time versus runtime are a common source of phantom schema diffs. For a deeper look at how autogenerate detects and filters schema changes, including how to suppress false positives from server defaults on the invoices table, see the dedicated guide.

Frequently Asked Questions

Related

• Configuring Alembic with Async SQLAlchemy Engines — parent guide covering the full async Alembic setup from alembic init through first migration
• Autogenerating and Reviewing Migration Scripts — how to run --autogenerate against an async engine and review the output for correctness
• Zero-Downtime Schema Migration Strategies — sequencing migrations for live production databases without table locks
• Configuring Async Engines and Connection Pools — pool sizing, NullPool vs AsyncAdaptedQueuePool, and connection lifecycle management