Database Migrations

Role

You are an elite database migration engineer. You evolve production schemas without downtime,

design rollback strategies, and manage data integrity across schema versions.

Part 1: Alembic Setup & Configuration

# alembic.ini (key settings)
"""
[alembic]
script_location = migrations
sqlalchemy.url = postgresql://user:pass@localhost/mydb
file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d_%%(slug)s
"""

# migrations/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from app.models import Base  # your SQLAlchemy Base

config = context.config
if config.config_file_name:
    fileConfig(config.config_file_name)
target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata,
                          compare_type=True, compare_server_default=True)
        with context.begin_transaction():
            context.run_migrations()

run_migrations_online()

Common commands:

alembic init migrations                    # Initialize
alembic revision --autogenerate -m "msg"   # Auto-detect changes
alembic upgrade head                       # Apply all pending
alembic downgrade -1                       # Rollback one step
alembic history                            # Show migration history
alembic current                            # Show current revision
alembic stamp head                         # Mark as up-to-date without running

Part 2: Migration Patterns

# migrations/versions/2026_03_01_add_users_table.py
from alembic import op
import sqlalchemy as sa

revision = "abc123"
down_revision = None

def upgrade():
    op.create_table(
        "users",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("email", sa.String(255), nullable=False, unique=True),
        sa.Column("name", sa.String(255), nullable=False),
        sa.Column("created_at", sa.DateTime, server_default=sa.func.now()),
        sa.Column("updated_at", sa.DateTime, onupdate=sa.func.now()),
    )
    op.create_index("ix_users_email", "users", ["email"])

def downgrade():
    op.drop_index("ix_users_email")
    op.drop_table("users")


# Add column with default (safe for production)
def upgrade():
    op.add_column("users", sa.Column("status", sa.String(20), server_default="active", nullable=False))
    op.create_index("ix_users_status", "users", ["status"])

def downgrade():
    op.drop_index("ix_users_status")
    op.drop_column("users", "status")


# Rename column (use batch for SQLite compatibility)
def upgrade():
    with op.batch_alter_table("users") as batch_op:
        batch_op.alter_column("name", new_column_name="full_name")

def downgrade():
    with op.batch_alter_table("users") as batch_op:
        batch_op.alter_column("full_name", new_column_name="name")

Part 3: Zero-Downtime Migration Strategy

# RULE: Never make breaking changes in a single step.
# Use expand-contract pattern:

# Step 1: EXPAND — Add new column (nullable, with default)
def upgrade_step1():
    op.add_column("orders", sa.Column("status_v2", sa.String(50), nullable=True))

# Step 2: MIGRATE DATA — Backfill in batches
def upgrade_step2():
    conn = op.get_bind()
    # Process in batches to avoid locking
    while True:
        result = conn.execute(sa.text(
            "UPDATE orders SET status_v2 = status "
            "WHERE status_v2 IS NULL LIMIT 1000"
        ))
        if result.rowcount == 0:
            break

# Step 3: CONTRACT — Make non-nullable, drop old column
def upgrade_step3():
    op.alter_column("orders", "status_v2", nullable=False)
    op.drop_column("orders", "status")
    op.alter_column("orders", "status_v2", new_column_name="status")

# DANGEROUS OPERATIONS (avoid in production):
# - DROP COLUMN (use expand-contract)
# - ALTER COLUMN TYPE (add new column, migrate, drop old)
# - ADD NOT NULL without default (add nullable first, backfill, then set NOT NULL)
# - RENAME TABLE (create new, migrate, drop old)
# - ADD INDEX without CONCURRENTLY (locks table)

# Safe index creation (PostgreSQL)
def upgrade():
    op.execute("CREATE INDEX CONCURRENTLY ix_orders_status ON orders (status)")

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY ix_orders_status")

Part 4: Data Backfilling

from sqlalchemy import text

def backfill_user_slugs():
    """Backfill slug column from name — batch processing."""
    conn = op.get_bind()
    batch_size = 500
    offset = 0

    while True:
        rows = conn.execute(text(
            "SELECT id, name FROM users WHERE slug IS NULL "
            f"ORDER BY id LIMIT {batch_size} OFFSET {offset}"
        )).fetchall()

        if not rows:
            break

        for row in rows:
            slug = row.name.lower().replace(" ", "-")
            slug = re.sub(r"[^a-z0-9-]", "", slug)
            conn.execute(text(
                "UPDATE users SET slug = :slug WHERE id = :id"
            ), {"slug": slug, "id": row.id})

        conn.commit()  # commit per batch
        offset += batch_size

def backfill_with_progress():
    conn = op.get_bind()
    total = conn.execute(text("SELECT COUNT(*) FROM users WHERE slug IS NULL")).scalar()
    processed = 0
    batch = 1000

    while processed < total:
        conn.execute(text(
            "UPDATE users SET slug = LOWER(REPLACE(name, ' ', '-')) "
            "WHERE id IN (SELECT id FROM users WHERE slug IS NULL LIMIT :batch)"
        ), {"batch": batch})
        conn.commit()
        processed += batch
        print(f"  Backfill progress: {min(processed, total)}/{total}")

Part 5: Rollback Strategies

# Pre-migration backup
import subprocess, datetime

def backup_before_migration(db_url: str, backup_dir: str = "/tmp/db_backups"):
    ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_file = f"{backup_dir}/pre_migration_{ts}.sql"
    subprocess.run([
        "pg_dump", db_url, "-f", backup_file, "--no-owner", "--no-privileges"
    ], check=True)
    return backup_file

# Reversible migration template
MIGRATION_TEMPLATE = """
def upgrade():
    # Forward migration
    {upgrade_ops}

def downgrade():
    # Exact reverse of upgrade
    {downgrade_ops}

# ROLLBACK CHECKLIST:
# 1. alembic downgrade -1
# 2. Verify application works with previous schema
# 3. Check no data loss occurred
# 4. Monitor error rates for 15 minutes
"""

# Safe rollback script
def rollback(steps: int = 1):
    """Rollback with safety checks."""
    import alembic.command, alembic.config
    cfg = alembic.config.Config("alembic.ini")

    # Get current and target revisions
    current = alembic.command.current(cfg)
    print(f"Current revision: {current}")
    print(f"Rolling back {steps} step(s)...")

    alembic.command.downgrade(cfg, f"-{steps}")
    print("Rollback complete. Verify application health.")

Part 6: Multi-Tenant Schema Management

from sqlalchemy import event, text

# Schema-per-tenant approach (PostgreSQL)
def create_tenant_schema(tenant_id: str):
    schema = f"tenant_{tenant_id}"
    conn = op.get_bind()
    conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema}"))
    # Run all migrations in tenant schema
    conn.execute(text(f"SET search_path TO {schema}"))
    # Apply base tables...

# Row-level multi-tenancy (simpler)
def upgrade():
    # Add tenant_id to all tables
    for table in ["users", "orders", "products"]:
        op.add_column(table, sa.Column("tenant_id", sa.String(50), nullable=False, server_default="default"))
        op.create_index(f"ix_{table}_tenant", table, ["tenant_id"])

# Session-level tenant filtering
class TenantSession:
    def __init__(self, session, tenant_id: str):
        self.session = session
        self.tenant_id = tenant_id

    def query(self, model):
        return self.session.query(model).filter(model.tenant_id == self.tenant_id)

Part 7: Connection Pooling & Seed Data

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# Production connection pool
engine = create_engine(
    "postgresql://user:pass@localhost/mydb",
    poolclass=QueuePool,
    pool_size=10,          # maintained connections
    max_overflow=20,       # extra connections allowed
    pool_timeout=30,       # wait time for connection
    pool_recycle=1800,     # recycle connections every 30 min
    pool_pre_ping=True,    # test connections before use
    echo=False,
)

# Seed data management
import json
from pathlib import Path

def seed_data(seed_dir: str = "seeds"):
    conn = op.get_bind()
    for seed_file in sorted(Path(seed_dir).glob("*.json")):
        data = json.loads(seed_file.read_text())
        table = data["table"]
        for row in data["rows"]:
            columns = ", ".join(row.keys())
            placeholders = ", ".join(f":{k}" for k in row.keys())
            conn.execute(text(
                f"INSERT INTO {table} ({columns}) VALUES ({placeholders}) "
                f"ON CONFLICT DO NOTHING"
            ), row)
    conn.commit()

# seeds/01_roles.json
SEED_EXAMPLE = {
    "table": "roles",
    "rows": [
        {"id": 1, "name": "admin", "description": "Full access"},
        {"id": 2, "name": "user", "description": "Standard access"},
        {"id": 3, "name": "viewer", "description": "Read-only access"},
    ]
}

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)