Database Migrations Guide

ProxyWhirl uses Alembic for database schema migrations, providing robust version control for the SQLite database schema.

Overview

Alembic manages incremental changes to the database schema over time, allowing you to:

• Track schema versions in version control alongside code

• Upgrade databases to newer schema versions

• Rollback changes if needed

• Generate migrations automatically from model changes

• Test migrations before deploying to production

Quick Start

Initialize a New Database

For new applications, initialize the database with the latest schema:

from proxywhirl.migrations import initialize_database

# Initialize database at latest schema
await initialize_database("sqlite+aiosqlite:///./proxywhirl.db")

Run Migrations Programmatically

ProxyWhirl provides a library-first API for migrations:

from proxywhirl.migrations import (
    run_migrations,
    check_pending_migrations,
    get_current_revision,
)

# Check if migrations are needed
if await check_pending_migrations("sqlite+aiosqlite:///./mydb.db"):
    print("Migrations pending...")
    await run_migrations("sqlite+aiosqlite:///./mydb.db")

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

Run Migrations via CLI

For command-line usage, use the Alembic CLI:

# Upgrade to latest version
uv run alembic upgrade head

# Show current revision
uv run alembic current

# Show migration history
uv run alembic history --verbose

# Downgrade one revision
uv run alembic downgrade -1

# Generate new migration after model changes
uv run alembic revision --autogenerate -m "description"

Migration API Reference

Core Functions

initialize_database(database_url=None)

Initialize a new database with the latest schema. Equivalent to running all migrations from scratch.

Parameters:

• database_url (str, optional): SQLAlchemy async database URL. If None, uses default from alembic.ini.

Raises:

• ValueError: If database is already initialized

Example:

await initialize_database("sqlite+aiosqlite:///./newdb.sqlite")

---

run_migrations(database_url=None, target_revision="head")

Run all pending migrations up to the target revision.

Parameters:

• database_url (str, optional): SQLAlchemy async database URL

• target_revision (str): Target revision (default: “head”)

  • "head" - Latest revision

  • "58c0fadfa0ca" - Specific revision ID

  • "+1" - One revision forward

  • "-1" - One revision back

Example:

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

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

---

downgrade_migrations(database_url=None, target_revision="-1")

Downgrade database to a previous revision. Use with caution in production.

Parameters:

• database_url (str, optional): SQLAlchemy async database URL

• target_revision (str): Target revision (default: “-1”)

  • "-1" - Previous revision

  • "base" - Rollback all migrations

  • "58c0fadfa0ca" - Specific revision ID

Example:

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

# Rollback all migrations
await downgrade_migrations("sqlite+aiosqlite:///./db.sqlite", "base")

---

get_current_revision(database_url=None)

Get the current migration revision of the database.

Returns: Current revision ID (str) or None if uninitialized

Example:

revision = await get_current_revision("sqlite+aiosqlite:///./db.sqlite")
if revision:
    print(f"Database at revision: {revision}")
else:
    print("Database not initialized")

---

get_head_revision(database_url=None)

Get the latest (head) migration revision available in code.

Returns: Head revision ID (str)

Example:

head = await get_head_revision()
current = await get_current_revision()
if head != current:
    print(f"Migrations pending: {current} -> {head}")

---

check_pending_migrations(database_url=None)

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

Returns: True if pending migrations exist, False otherwise

Example:

if await check_pending_migrations("sqlite+aiosqlite:///./db.sqlite"):
    print("WARNING: Pending migrations detected!")
    await run_migrations()

---

get_migration_history(database_url=None)

Get the complete migration history.

Returns: List of dicts with keys: revision, down_revision, description

Example:

history = await get_migration_history()
for migration in history:
    print(f"{migration['revision']}: {migration['description']}")

---

stamp_revision(database_url=None, revision="head")

Stamp the database with a specific revision without running migrations.

⚠️ Use with extreme caution! This bypasses actual schema changes.

Useful for:

• Importing existing databases that match a schema version

• Recovering from migration issues

Example:

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

Creating New Migrations

Automatic Generation (Recommended)

After modifying SQLModel table definitions in proxywhirl/storage.py:

# Generate migration automatically
uv run alembic revision --autogenerate -m "add_column_xyz"

# Review generated migration in alembic/versions/
# Edit if needed (especially for data migrations)

# Test migration
uv run alembic upgrade head

# Verify with tests
uv run pytest tests/unit/test_migrations.py

Manual Migration

For complex data migrations or when autogenerate isn’t sufficient:

# Create empty migration
uv run alembic revision -m "custom_data_migration"

# Edit alembic/versions/TIMESTAMP_custom_data_migration.py
# Add upgrade() and downgrade() logic

Example manual migration:

def upgrade() -> None:
    """Add default values to existing records."""
    op.execute(
        """
        UPDATE proxies 
        SET health_status = 'unknown' 
        WHERE health_status IS NULL
        """
    )

def downgrade() -> None:
    """Revert changes."""
    pass  # Data migrations often can't be cleanly reverted

Best Practices

1. Always Test Migrations

# Create test database
uv run python -c "
from proxywhirl.migrations import initialize_database
import asyncio
asyncio.run(initialize_database('sqlite+aiosqlite:///./test.db'))
"

# Test upgrade
uv run alembic upgrade head

# Test downgrade
uv run alembic downgrade -1

# Test re-upgrade
uv run alembic upgrade head

# Run automated tests
uv run pytest tests/unit/test_migrations.py

2. Use Batch Operations for SQLite

SQLite has limited ALTER TABLE support. Always use batch operations:

def upgrade() -> None:
    with op.batch_alter_table('proxies', schema=None) as batch_op:
        batch_op.add_column(sa.Column('new_field', sa.String()))
        batch_op.create_index('ix_new_field', ['new_field'])

Alembic’s render_as_batch=True (configured in alembic/env.py) handles this automatically.

3. Never Import Models in Migrations

❌ Don’t do this:

from proxywhirl.storage import ProxyTable  # BAD!

def upgrade() -> None:
    # Model might change in future, breaking old migrations
    op.create_table(ProxyTable.__tablename__, ...)

✅ Do this instead:

def upgrade() -> None:
    # Static schema definition
    op.create_table(
        'proxies',
        sa.Column('url', sa.String(), nullable=False),
        sa.Column('protocol', sa.String(), nullable=True),
        # ...
    )

4. Make Migrations Reversible

Always implement both upgrade() and downgrade():

def upgrade() -> None:
    op.add_column('proxies', sa.Column('rating', sa.Float()))

def downgrade() -> None:
    op.drop_column('proxies', 'rating')

5. Use Comments

Add comments to complex migrations:

def upgrade() -> None:
    """Add proxy rating system.
    
    Adds a rating column (0.0-1.0) to track proxy quality based on
    success rate and response time. Existing proxies default to 0.5.
    """
    op.add_column('proxies', sa.Column('rating', sa.Float(), default=0.5))

6. Check Migrations in CI/CD

Add to your CI pipeline:

# Check for pending migrations
uv run alembic check || {
    echo "ERROR: Database schema changes detected!"
    echo "Generate migrations with:"
    echo "  uv run alembic revision --autogenerate -m 'description'"
    exit 1
}

7. Backup Before Production Migrations

# Backup database
cp proxywhirl.db proxywhirl.db.backup.$(date +%Y%m%d_%H%M%S)

# Run migration
uv run alembic upgrade head

# Verify
uv run alembic current

Application Integration

Startup Migration Check

Run migrations automatically on application startup:

from proxywhirl.migrations import check_pending_migrations, run_migrations
from loguru import logger

async def initialize_app():
    """Initialize application with database migrations."""
    db_url = "sqlite+aiosqlite:///./proxywhirl.db"
    
    if await check_pending_migrations(db_url):
        logger.info("Pending migrations detected, upgrading...")
        await run_migrations(db_url)
        logger.info("Migrations complete")
    else:
        logger.info("Database up to date")

Health Check Endpoint

For REST API:

from fastapi import FastAPI, HTTPException
from proxywhirl.migrations import get_current_revision, check_pending_migrations

app = FastAPI()

@app.get("/health/database")
async def database_health():
    """Check database health and migration status."""
    db_url = "sqlite+aiosqlite:///./proxywhirl.db"
    
    current = await get_current_revision(db_url)
    if current is None:
        raise HTTPException(
            status_code=500,
            detail="Database not initialized"
        )
    
    has_pending = await check_pending_migrations(db_url)
    
    return {
        "status": "warning" if has_pending else "healthy",
        "current_revision": current,
        "pending_migrations": has_pending,
    }

Troubleshooting

“Database already initialized” Error

This occurs when trying to initialize an already-initialized database:

# Check current revision first
revision = await get_current_revision(db_url)
if revision:
    print(f"Database already initialized at {revision}")
    await run_migrations(db_url)  # Use upgrade instead
else:
    await initialize_database(db_url)

Migration Conflicts

If multiple branches create migrations:

# Merge branches with migrations
uv run alembic merge heads -m "merge_branches"

# Or manually edit migration to specify down_revision

Corrupted Migration State

If migration state is corrupted:

# Option 1: Stamp with correct revision (if schema matches)
uv run alembic stamp head

# Option 2: Restore from backup
cp proxywhirl.db.backup proxywhirl.db

# Option 3: Reinitialize (LOSES DATA!)
rm proxywhirl.db
uv run alembic upgrade head

Testing Migration Recovery

Simulate failures:

async def test_migration_recovery():
    """Test migration error handling."""
    db_url = "sqlite+aiosqlite:///./test.db"
    
    try:
        await run_migrations(db_url)
    except Exception as e:
        logger.error(f"Migration failed: {e}")
        # Restore from backup
        await restore_from_backup()
        raise

Directory Structure

proxywhirl/
├── alembic.ini              # Alembic configuration
├── alembic/
│   ├── env.py              # Migration environment (async support)
│   ├── script.py.mako      # Migration template
│   └── versions/           # Migration scripts
│       └── 20251120_..._initial_schema.py
└── proxywhirl/
    ├── storage.py          # SQLModel table definitions
    └── migrations.py       # Programmatic migration API

Configuration

alembic.ini

Key configuration options:

[alembic]
# Migration scripts location
script_location = alembic

# Database URL (can be overridden programmatically)
sqlalchemy.url = sqlite+aiosqlite:///./proxywhirl.db

# Filename template for migrations
file_template = %%(year)d%%(month).2d%%(day).2d_%%(hour).2d%%(minute).2d_%%(rev)s_%%(slug)s

# Timezone for timestamps
timezone = UTC

env.py

Configured with:

• Async SQLite support via aiosqlite

• Batch mode for SQLite ALTER operations

• Naming conventions for consistent constraint names

• Type comparison enabled for detecting column changes

See Also

• Alembic Documentation

• SQLModel Documentation

• ProxyWhirl Configuration Guide

• ProxyWhirl API Reference