· Patryk Komow · 8 min read

Alembic migrations with async SQLAlchemy in FastAPI

Alembic itself is synchronous, your app's engine is async, and the default template doesn't know your settings or your models. Here is a setup that runs migrations through the same async engine and settings as the app, generates predictable files, and gets tested on every test run.

Contents

  1. The engine and the declarative base
  2. Name your constraints before the first migration
  3. An async env.py
  4. Make sure Alembic sees every model
  5. alembic.ini: dated file names and auto-formatting
  6. The workflow: generate, read, apply
  7. What autogenerate gets wrong
  8. Running migrations in production
  9. Testing the migrations

The engine and the declarative base

Everything starts in one module that the app, Alembic and the tests all import:

# app/db.py
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.pool import NullPool

# Stable constraint names, so Alembic autogenerate produces predictable migrations.
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)


settings = get_settings()

# Tests run each test in its own event loop, and pooled asyncpg connections
# can't move between loops, so pooling is switched off there.
engine = create_async_engine(
    settings.database_url,
    poolclass=NullPool if settings.app_env == "test" else None,
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

expire_on_commit=False matters in async code. By default SQLAlchemy expires every object after a commit, and the next attribute access triggers a lazy reload, which in async mode raises MissingGreenlet instead of quietly running a query. Keeping the loaded values avoids that whole class of errors after await session.commit().

Name your constraints before the first migration

Without a naming convention, PostgreSQL invents names for unique constraints, foreign keys and indexes, and Alembic writes None for the name in the migration. That works until the day you need to drop or change one of them: the migration has to know a name that only exists in the database, and it may differ between your laptop and production.

With the convention above, every constraint gets a name derived from the table and column, and autogenerate writes it into the migration with op.f():

sa.ForeignKeyConstraint(
    ["user_id"], ["users.id"], name=op.f("fk_plan_grants_user_id_users"), ondelete="CASCADE"
),
sa.PrimaryKeyConstraint("user_id", name=op.f("pk_plan_grants")),

Set it before your first migration. Adding it later is possible, but then you're renaming every existing constraint by hand.

An async env.py

Alembic's migration runner is synchronous. An async engine can still drive it: open an async connection, then hand Alembic a synchronous view of it with run_sync. (alembic init -t async migrations generates a similar skeleton.) The important change from the template is where the URL and the metadata come from:

# migrations/env.py
import asyncio
from logging.config import fileConfig

from alembic import context
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import create_async_engine

import app.models  # noqa: F401  (registers every model on Base.metadata)
from app.config import get_settings
from app.db import Base

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

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Print the SQL instead of running it: `alembic upgrade head --sql`."""
    context.configure(
        url=get_settings().database_url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


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


async def run_migrations_online() -> None:
    engine = create_async_engine(get_settings().database_url)
    async with engine.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await engine.dispose()


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

Three choices in there:

Make sure Alembic sees every model

Autogenerate compares the database with Base.metadata, and a model is only in the metadata if its module has been imported. Forget one, and autogenerate concludes the table should not exist and writes a drop_table for it. Keep one module whose only job is to import every model:

# app/models.py
"""Imports every model, so Alembic and the tests see all tables.

When you add a models.py to a new module, import it here.
"""

from app.accounts.models import User
from app.billing.models import (
    BillingCustomer,
    BillingProfile,
    Invoice,
    Payment,
    PlanGrant,
    StripeEvent,
    Subscription,
)

env.py imports app.models, and so does the test setup. When a generated migration contains a drop_table you didn't ask for, a missing import here is the first thing to check.

alembic.ini: dated file names and auto-formatting

[alembic]
script_location = migrations
prepend_sys_path = .
file_template = %%(year)d%%(month).2d%%(day).2d_%%(rev)s_%%(slug)s

[post_write_hooks]
hooks = ruff_fix, ruff
ruff_fix.type = exec
ruff_fix.executable = ruff
ruff_fix.options = check --fix --quiet REVISION_SCRIPT_FILENAME
ruff.type = exec
ruff.executable = ruff
ruff.options = format REVISION_SCRIPT_FILENAME

The workflow: generate, read, apply

alembic revision --autogenerate -m "add projects"
# open migrations/versions/2026..._add_projects.py and read it
alembic upgrade head

The middle step is not optional. Autogenerate is a diff tool, not a mind reader, and it's right most of the time, which is exactly why the time it's wrong slips through. Read every generated file before applying it, and check that downgrade() really undoes upgrade().

What autogenerate gets wrong

A model detail that makes migrations cleaner: use server_default=func.now() for timestamps rather than a Python-side default. The default then lives in the database, and autogenerate writes it as server_default=sa.text("now()"), so rows inserted by hand or by another tool get it too.

Running migrations in production

The simplest reliable option for a single app is to migrate when the container starts:

CMD ["sh", "-c", "alembic upgrade head && exec uvicorn app.main:app --host 0.0.0.0 --port ${PORT:-8000} --proxy-headers"]

If a migration fails, the app never starts, the health check fails, and a platform with health-checked deploys keeps the old version serving. Take a backup before any deploy that contains a migration; rolling back code is easy, rolling back a dropped column isn't. And once you run several instances that start at the same time, move alembic upgrade head into a separate release step so they don't race.

Locally, the development container runs the same command before uvicorn --reload, so pulling a branch with a new migration and restarting is all a teammate has to do.

Testing the migrations

The test suite needs a database schema anyway. Build it with the real migrations instead of Base.metadata.create_all(), and you test the migrations on every run for free:

# tests/conftest.py
@pytest.fixture(scope="session", autouse=True)
def migrated_database():
    """Builds the test database with the real migrations (down, then up), so the
    migrations are tested too."""
    for command in (["alembic", "downgrade", "base"], ["alembic", "upgrade", "head"]):
        subprocess.run(command, check=True, capture_output=True)

Running downgrade base first means every downgrade() is exercised too: a broken downgrade fails the test run, not your 2 a.m. rollback. Running it as a subprocess keeps Alembic's asyncio.run out of pytest's event loop. The tests point DATABASE_URL at a separate test database before anything imports the settings, so this never touches development data.

create_all() would hide exactly the bugs you want to find: a column in the model that no migration creates passes every test and then fails in production.

Checklist

This database setup is part of getpylaunch, a FastAPI SaaS starter I sell, where the migrations for accounts and billing are already written and tested.