++++

Engineering

Apr 2026×11 min read

How this template handles schema evolution with a custom async migration runner — no Alembic required, just versioned modules with explicit up and down functions.

6. Migrations

Driptanil DattaSoftware Developer

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:

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:

# 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:

# 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:

# 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:

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:

# 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.