Stats
Actions
Tags
From example-skills
Manages database schema changes safely with Alembic migrations, zero-downtime strategies, rollbacks, and testing for SQL and data migrations.
How this skill is triggered — by the user, by Claude, or both
Slash command
/example-skills:database-migration-patternsThe summary Claude sees in its skill listing — used to decide when to auto-load this skill
Evolve database schemas safely with versioned, reversible, tested migrations.
Evolve database schemas safely with versioned, reversible, tested migrations.
# Initialize
alembic init alembic
# Create migration
alembic revision --autogenerate -m "add skills table"
# Run migrations
alembic upgrade head
# Rollback one step
alembic downgrade -1
# Show current state
alembic current
alembic history
# alembic/env.py
from app.models import Base
from app.config import settings
target_metadata = Base.metadata
def run_migrations_online():
connectable = create_async_engine(settings.database_url)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
def do_run_migrations(connection):
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
"""add skills table
Revision ID: abc123
Revises: def456
Create Date: 2026-03-20 10:00:00
"""
from alembic import op
import sqlalchemy as sa
revision = "abc123"
down_revision = "def456"
def upgrade():
op.create_table(
"skills",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("name", sa.String(64), nullable=False, unique=True),
sa.Column("description", sa.Text, nullable=False),
sa.Column("category", sa.String(32), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
op.create_index("ix_skills_category", "skills", ["category"])
def downgrade():
op.drop_index("ix_skills_category")
op.drop_table("skills")
# Safe: add nullable column first
def upgrade():
op.add_column("skills", sa.Column("tier", sa.String(20), nullable=True))
# Later migration: backfill then add constraint
def upgrade():
op.execute("UPDATE skills SET tier = 'community' WHERE tier IS NULL")
op.alter_column("skills", "tier", nullable=False, server_default="community")
# Phase 1: Add new column
def upgrade():
op.add_column("skills", sa.Column("skill_category", sa.String(32)))
op.execute("UPDATE skills SET skill_category = category")
# Phase 2: (after app updated to use new column)
def upgrade():
op.drop_column("skills", "category")
# Phase 1: Stop writing to column (app change)
# Phase 2: Remove column
def upgrade():
op.drop_column("skills", "deprecated_field")
# Use CONCURRENTLY for zero-downtime
def upgrade():
op.execute("CREATE INDEX CONCURRENTLY ix_skills_name ON skills (name)")
def downgrade():
op.drop_index("ix_skills_name")
"""backfill governance metadata
Revision ID: ghi789
"""
from alembic import op
import sqlalchemy as sa
def upgrade():
# Use raw SQL for large table updates
conn = op.get_bind()
conn.execute(sa.text("""
UPDATE skills
SET governance_phases = ARRAY['build']
WHERE governance_phases IS NULL
AND category IN ('development', 'data')
"""))
conn.execute(sa.text("""
UPDATE skills
SET governance_phases = ARRAY['prove']
WHERE governance_phases IS NULL
AND category IN ('security', 'documentation')
"""))
def downgrade():
conn = op.get_bind()
conn.execute(sa.text("UPDATE skills SET governance_phases = NULL"))
import pytest
from alembic.config import Config
from alembic import command
@pytest.fixture
def alembic_config():
config = Config("alembic.ini")
config.set_main_option("sqlalchemy.url", test_database_url)
return config
def test_upgrade_downgrade(alembic_config):
# Full upgrade
command.upgrade(alembic_config, "head")
# Full downgrade
command.downgrade(alembic_config, "base")
# Back to head
command.upgrade(alembic_config, "head")
def test_migration_data_integrity(alembic_config, db_session):
command.upgrade(alembic_config, "head")
# Insert test data
db_session.execute(sa.text("INSERT INTO skills (id, name, description, category) VALUES ('t1', 'test', 'Test skill', 'dev')"))
db_session.commit()
# Verify data survives next migration
command.upgrade(alembic_config, "head")
result = db_session.execute(sa.text("SELECT name FROM skills WHERE id = 't1'"))
assert result.scalar() == "test"
npx claudepluginhub a-organvm/a-i--skills --plugin document-skillsProvides a checklist for code reviews covering functionality, security, performance, maintainability, tests, and quality. Use for pull requests, audits, team standards, and developer training.