csrd-migration 0.3.109

Versioned schema migrations for csrd database adapters

csrd-migration

Versioned, idempotent schema migrations for csrd.repository database adapters.

Quick start

from csrd.migration import Migration, MigrationRunner
from csrd.repository import SQLiteAdapter

MIGRATIONS = [
    Migration(
        version="001",
        description="create users table",
        up="CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT NOT NULL)",
        down="DROP TABLE users",
    ),
]

async def setup_db():
    adapter = SQLiteAdapter("app.db")
    await adapter.connect()
    runner = MigrationRunner()
    await runner.apply_all(adapter, MIGRATIONS)

API

Migration

Frozen dataclass representing a single schema change.

Field	Type	Description
version	str	Unique version identifier (e.g. "001", "002")
description	str	Human-readable description
up	str	SQL to apply the migration
down	str	SQL to revert the migration

MigrationRunner

Applies and tracks migrations against any ABCDatabaseAdapter.

Method	Description
apply_all(adapter, migrations)	Apply all pending migrations in order
rollback(adapter, migrations, *, to_version=None)	Roll back applied migrations in reverse order. Pass to_version to stop at a specific version, or None to roll back all
reapply_all(adapter, migrations)	Roll back all then re-apply — clean schema reset without dropping the database
current_version(adapter)	Return the latest applied version, or None
pending(adapter, migrations)	Return migrations not yet applied

The runner creates a _csrd_migrations table automatically to track applied versions. Re-running is idempotent — already-applied migrations are skipped.

Design

• Adapter-agnostic — works with SQLiteAdapter, PGAdapter, MariaAdapter
• Idempotent — safe to call on every startup
• Ordered — migrations are applied in list order
• Future-ready — the Migration dataclass is the stable interface for a future model→SQL translation layer (see docs/AUGMENT_PLAN.md)