Skip to content
Contact →

6. Migrations

Updated
17 Sept 2026
3 min read
6 sections
Page 8 of 12
FastAPI: Production-Ready Backend Guide

Why Migrations Exist

Your application code and your database schema have to stay in sync. When you add a column or rename a table, that change needs to propagate consistently to every environment — local, staging, production. Without a migration system, you end up doing this manually, which means inconsistencies, forgotten steps, and no way to roll back safely.

This template includes a lightweight async migration runner that doesn't need Alembic. It's simple enough to understand completely and powerful enough to manage a real schema.

The Migration File Structure

Each migration is a Python module in server/migration/src/. The naming convention encodes the date and a description:

text
m20260408_000001_create_todos_table.py

Every migration file exports two async functions: up to apply the change and down to reverse it. Here is the one that creates the todos table:

python
# server/migration/src/m20260408_000001_create_todos_table.py
async def up(conn: AsyncConnection) -> None:
    await conn.execute(
        text("""
        CREATE TABLE IF NOT EXISTS todos (
            id VARCHAR(36) PRIMARY KEY,
            title VARCHAR(255) NOT NULL,
            description TEXT,
            status VARCHAR(32) NOT NULL DEFAULT 'pending',
            created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        """)
    )


async def down(conn: AsyncConnection) -> None:
    await conn.execute(text("DROP TABLE IF EXISTS todos;"))

CREATE TABLE IF NOT EXISTS makes the up function idempotent — running it twice won't fail. down simply drops the table. Both operations run inside a single transaction managed by the runner.

Registering Migrations

The Migrator class in server/migration/src/lib.py is the registry. Every migration module you create must be added here:

python
# server/migration/src/lib.py
class Migrator:
    @staticmethod
    def migrations() -> tuple[Migration, ...]:
        return (
            Migration(
                name="m20260408_000001_create_todos_table",
                up=m20260408_000001_create_todos_table.up,
                down=m20260408_000001_create_todos_table.down,
            ),
        )

The order of entries in the tuple is the order migrations are applied. upgrade walks through them forward; downgrade walks through them in reverse.

The Runner

The runner in server/migration/src/runner.py applies the migrations from the registry in a transaction:

python
# server/migration/src/runner.py
async def upgrade(engine: AsyncEngine) -> None:
    async with engine.begin() as conn:
        for migration in Migrator.migrations():
            await migration.up(conn)


async def downgrade(engine: AsyncEngine) -> None:
    async with engine.begin() as conn:
        for migration in reversed(Migrator.migrations()):
            await migration.down(conn)

engine.begin() opens a connection and starts a transaction. If any migration fails, the entire transaction rolls back. You never end up in a half-applied state.

To bootstrap a fresh database:

bash
poetry run python scripts/init_db.py

scripts/init_db.py calls upgrade with the configured engine. This is the same command used in CI and staging setup, so the database state is always reproducible from scratch.

What Ad-Hoc Schema Changes Break

Running raw SQL outside the migration registry is tempting during incidents, but it creates drift you can't roll back:

python
# What not to do: ad-hoc mutation outside the registry
async def hotfix_schema(conn):
    await conn.execute(text("ALTER TABLE todos ADD COLUMN temp_flag TEXT"))

This change exists in production but not in your migration history. The next time someone bootstraps from scratch — new developer, new staging environment, CI teardown — the column is missing. If the application code depends on temp_flag, it fails. If it doesn't, the column just lives in production forever, invisible and untested.

Adding a New Migration

When your schema needs to change, the process is:

  1. Create a new module in server/migration/src/ following the naming convention
  2. Implement up and down functions
  3. Register it in Migrator.migrations() in lib.py
  4. Run poetry run python scripts/init_db.py to apply it locally

That's the full cycle. Every schema change is versioned, reversible, and reproducible. Before a production deploy, validate that both upgrade and downgrade run cleanly on a staging copy of the database — that's the only way to confirm your down migration actually works before you need it.