proxywhirl.migrations

Database migration utilities for ProxyWhirl.

This module provides a programmatic API for running Alembic migrations, following ProxyWhirl’s library-first architecture. It allows applications to manage database schema versions without requiring command-line tools.

Key Features: - Programmatic migration execution - Current revision checking - Pending migrations detection - Database initialization - Async SQLite support

Best Practices: - Always backup databases before migrations - Test migrations in development first - Use check_pending_migrations() in CI/CD - Call run_migrations() on application startup

Example:

from proxywhirl.migrations import run_migrations, get_current_revision

# Run all pending migrations
await run_migrations("sqlite+aiosqlite:///./mydb.db")

# Check current version
revision = await get_current_revision("sqlite+aiosqlite:///./mydb.db")
print(f"Database at revision: {revision}")

Functions

check_pending_migrations([database_url])

Check if there are pending migrations that need to be applied.

downgrade_migrations([database_url, target_revision])

Downgrade database to a previous revision.

get_current_revision([database_url])

Get the current migration revision of the database.

get_head_revision([database_url])

Get the latest (head) migration revision available.

get_migration_history([database_url])

Get the migration history showing all applied migrations.

initialize_database([database_url])

Initialize a new database with the latest schema.

run_migrations([database_url, target_revision])

Run all pending migrations up to target revision.

stamp_revision([database_url, revision])

Stamp the database with a specific revision without running migrations.

Module Contents

async proxywhirl.migrations.check_pending_migrations(database_url=None)[source]

Check if there are pending migrations that need to be applied.

This is useful for CI/CD pipelines and application health checks.

Parameters:

database_url (str | None) – SQLAlchemy async database URL

Returns:

True if pending migrations exist, False otherwise

Return type:

bool

Example:

if await check_pending_migrations("sqlite+aiosqlite:///./db.sqlite"):
    print("WARNING: Pending migrations detected!")
    await run_migrations()
async proxywhirl.migrations.downgrade_migrations(database_url=None, target_revision='-1')[source]

Downgrade database to a previous revision.

Use with caution in production. Always backup data before downgrading.

Parameters:

  • database_url (str | None) – SQLAlchemy async database URL

  • target_revision (str) – Target revision to downgrade to. Defaults to “-1” (previous). Can be specific revision ID, relative revision, or “base” for complete rollback.

Raises:
Return type:

None

Example:

# Downgrade one revision
await downgrade_migrations("sqlite+aiosqlite:///./db.sqlite")

# Downgrade to specific revision
await downgrade_migrations("sqlite+aiosqlite:///./db.sqlite", "58c0fadfa0ca")

# Rollback all migrations
await downgrade_migrations("sqlite+aiosqlite:///./db.sqlite", "base")
async proxywhirl.migrations.get_current_revision(database_url=None)[source]

Get the current migration revision of the database.

Parameters:

database_url (str | None) – SQLAlchemy async database URL

Returns:

Current revision ID, or None if database is uninitialized

Return type:

str | None

Example:

revision = await get_current_revision("sqlite+aiosqlite:///./db.sqlite")
if revision:
    print(f"Database at revision: {revision}")
else:
    print("Database not initialized")
async proxywhirl.migrations.get_head_revision(database_url=None)[source]

Get the latest (head) migration revision available.

Parameters:

database_url (str | None) – SQLAlchemy async database URL (used to get config)

Returns:

Head revision ID

Return type:

str

Example:

head = await get_head_revision()
current = await get_current_revision()
if head != current:
    print(f"Migrations pending: {current} -> {head}")
async proxywhirl.migrations.get_migration_history(database_url=None)[source]

Get the migration history showing all applied migrations.

Parameters:

database_url (str | None) – SQLAlchemy async database URL

Returns:

Migration records with keys revision, down_revision, description.

Return type:

list[dict[str, str | None]]

Example:

history = await get_migration_history()
for migration in history:
    print(f"{migration['revision']}: {migration['description']}")
async proxywhirl.migrations.initialize_database(database_url=None)[source]

Initialize a new database with the latest schema.

This is a convenience function that creates all tables at the current head revision without running individual migrations. Equivalent to running all migrations from scratch.

Parameters:

database_url (str | None) – SQLAlchemy async database URL

Raises:

Exception – If database already has schema or initialization fails

Return type:

None

Example:

# Initialize new database
await initialize_database("sqlite+aiosqlite:///./newdb.sqlite")
async proxywhirl.migrations.run_migrations(database_url=None, target_revision='head')[source]

Run all pending migrations up to target revision.

This function executes Alembic migrations programmatically, allowing applications to manage schema versions without CLI tools. It uses async SQLite support for non-blocking execution.

Parameters:

  • database_url (str | None) – SQLAlchemy async database URL (e.g., “sqlite+aiosqlite:///./db.sqlite”). If None, uses default from alembic.ini

  • target_revision (str) – Target revision to migrate to. Defaults to “head” (latest). Can be specific revision ID, relative revision (e.g., “+1”, “-2”), or branch label.

Raises:
Return type:

None

Example:

# Migrate to latest
await run_migrations("sqlite+aiosqlite:///./proxywhirl.db")

# Migrate to specific revision
await run_migrations("sqlite+aiosqlite:///./db.sqlite", "58c0fadfa0ca")

# Rollback one revision
await run_migrations("sqlite+aiosqlite:///./db.sqlite", "-1")
async proxywhirl.migrations.stamp_revision(database_url=None, revision='head')[source]

Stamp the database with a specific revision without running migrations.

This is useful when importing an existing database that matches a specific schema version, or for recovering from migration issues.

Use with extreme caution! This bypasses actual schema changes.

Parameters:

  • database_url (str | None) – SQLAlchemy async database URL

  • revision (str) – Revision to stamp database with

Return type:

None

Example:

# Stamp database as being at head revision
await stamp_revision("sqlite+aiosqlite:///./db.sqlite", "head")