data.migrate

Page actions AI-ready formats and sharing
Open LLM text
Share with AI
Ask Claude Ask ChatGPT Ask Gemini Ask Copilot
module

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…

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/")

data.migrate

Name Type Default Description
type —
qualified_name —
element_type —
description —
source_file —
line_number —
is_autodoc —
autodoc_element —
_autodoc_template —
_autodoc_url_path —
_autodoc_page_type —
title —
doc_content_hash —

Symbols on this page

function _checksum

Stable content hash of a migration's SQL body.

Hashes the already-.strip()-ed Migration.sqlexactly as stored at apply time so the drift recompute…

Jump to symbol class Migration

A single migration file.

Jump to symbol class MigrationResult

Result of running migrations.

Jump to symbol function _discover_migrations

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…

Jump to symbol async function _ensure_tracking_table

Create the migration tracking table if it doesn't exist.

Also brings a tracking table created by a pre-checksum Chirp version up to date:CREATE…

Jump to symbol async function _tracking_has_checksum_column

Report whether the tracking table already has achecksumcolumn.

Jump to symbol async function _get_applied_versions

Get applied migrations asversion -> (name, checksum).

checksum is Nonefor rows written before the checksum column existed (legacy → skipped by…

Jump to symbol async function _apply_migration

Apply a single migration.

Usesexecute_scriptfor multi-statement migration files (e.g. CREATE TABLE + CREATE INDEX in one file). SQLite gets an explicit BEGIN/COMMIT…

Jump to symbol async function migrate

Apply pending migrations from a directory.

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

Jump to symbol
_checksum
function 
def _checksum(sql: str) -> str

Stable content hash of a migration's SQL body.

Hashes the already-.strip()-ed Migration.sqlexactly as stored at apply time so the drift recompute matches byte-for-byte (a whitespace or normalization mismatch would be a false-positive drift error).

Parameters

Name Type Default Description
sql str —
Migration
class

A single migration file.

MigrationResult
class

Result of running migrations.

_discover_migrations
function 
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 Default Description
directory str | Path —
_ensure_tracking_table
function async 
async def _ensure_tracking_table(db: Database) -> None

Create the migration tracking table if it doesn't exist.

Also brings a tracking table created by a pre-checksum Chirp version up to date:CREATE TABLE IF NOT EXISTSnever alters an existing table, so the nullablechecksumcolumn is added idempotently here. Existing rows get NULL(legacy → skip-verify). Column existence is introspected per driver so theALTERonly runs when the column is genuinely absent — no broad exception swallowing.

Parameters

Name Type Default Description
db Database —
_tracking_has_checksum_column
function async 
async def _tracking_has_checksum_column(db: Database) -> bool

Report whether the tracking table already has achecksumcolumn.

Parameters

Name Type Default Description
db Database —
_get_applied_versions
function async 
async def _get_applied_versions(db: Database) -> dict[int, tuple[str | None, str | None]]

Get applied migrations asversion -> (name, checksum).

checksum is Nonefor rows written before the checksum column existed (legacy → skipped by the drift guard).

Parameters

Name Type Default Description
db Database —
_apply_migration
function async 
async def _apply_migration(db: Database, migration: Migration) -> None

Apply a single migration.

Usesexecute_scriptfor multi-statement migration files (e.g. CREATE TABLE + CREATE INDEX in one file). SQLite gets an explicit BEGIN/COMMIT wrapper becausesqlite3.executescriptdoes not honor the connection's surrounding transaction mode.

Parameters

Name Type Default Description
db Database —
migration Migration —
migrate
function 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 Default Description
db Database — Connected database instance.
directory str | Path — Path to the migrations directory.

View source · /home/runner/work/chirp/chirp/site/../src/chirp/data/migrate.py:1