Module

data.migrate

Forward-only SQL migration runner.

Migrations are numbered.sqlfiles in a directory::

migrations/
    001_create_users.sql
    002_add_email_index.sql
    003_create_orders.sql

Applied migrations are tracked in a_chirp_migrationstable. Each migration runs inside a transaction — if it fails, the migration is rolled back and no further migrations are applied.

Usage::

from chirp.data import Database, migrate

db = Database("sqlite:///app.db")
await db.connect()
await migrate(db, "migrations/")

Or integrated with the app::

app = App(db="sqlite:///app.db", migrations="migrations/")
2Classes5Functions

Classes

Migration 3 ▼
A single migration file.

A single migration file.

Attributes

Name Type Description
version int —
name str —
sql str —
MigrationResult 4 ▼
Result of running migrations.

Result of running migrations.

Attributes

Name Type Description
applied list[str] —
already_applied int —
total_available int —

Methods

summary 0 str ▼
property
def summary(self) -> str
Returns
str

Functions

_discover_migrations 1 list[Migration] ▼
Discover and parse migration files from a directory. Files must match the patt…
def _discover_migrations(directory: str | Path) -> list[Migration]

Discover and parse migration files from a directory.

Files must match the patternNNN_description.sqlwhere NNN is a zero-padded integer version number. Files are sorted by version.

Parameters
Name Type Description
directory str | Path
Returns
list[Migration]
_ensure_tracking_table 1 None ▼
Create the migration tracking table if it doesn't exist.
async
async def _ensure_tracking_table(db: Database) -> None
Parameters
Name Type Description
db Database
_get_applied_versions 1 set[int] ▼
Get the set of already-applied migration versions.
async
async def _get_applied_versions(db: Database) -> set[int]
Parameters
Name Type Description
db Database
Returns
set[int]
_apply_migration 2 None ▼
Apply a single migration. Uses ``execute_script`` for SQLite to support multi-…
async
async def _apply_migration(db: Database, migration: Migration) -> None

Apply a single migration.

Usesexecute_scriptfor SQLite to support multi-statement migration files (e.g. CREATE TABLE + CREATE INDEX in one file). The tracking record is inserted separately after the migration succeeds.

Parameters
Name Type Description
db Database
migration Migration
migrate 2 MigrationResult ▼
Apply pending migrations from a directory. Discovers ``.sql`` files, compares …
async
async def migrate(db: Database, directory: str | Path) -> MigrationResult

Apply pending migrations from a directory.

Discovers.sqlfiles, compares against the tracking table, and applies missing migrations in version order. Each migration runs in its own transaction.

Parameters
Name Type Description
db Database

Connected database instance.
directory str | Path

Path to the migrations directory.
Returns
MigrationResult