alembic

Alembic 数据库迁移管理技能

概述

此技能全面介绍了如何在客户支持环境中使用 Alembic 管理数据库迁移，涵盖从初始设置到复杂的生产部署场景，重点在于保持数据完整性并尽量减少对支持运营的影响。

核心概念

什么是 Alembic？

Alembic 是一款轻量级数据库迁移工具，可与 SQLAlchemy 配合使用。它通过版本控制迁移脚本，帮助你管理数据库架构随时间发生的变更。对于客户支持系统而言，这意味着：

• 版本控制：跟踪支持数据库中的所有架构变更
• 可复现性：在开发、预发布和生产环境中应用相同的迁移
• 回滚能力：安全地撤销有问题的变更
• 团队协作：合并多位开发者所做的架构变更
• 数据保留：在架构转换期间迁移数据

支持系统中的迁移生命周期

1. 开发：开发新功能时在本地创建迁移
2. 测试：在预发布环境中验证迁移
3. 审查：在部署到生产环境前审查迁移脚本
4. 部署：以尽可能短的停机时间将迁移应用到生产环境
5. 监控：跟踪迁移状态并处理故障
6. 回滚：如果生产环境出现问题，则回滚

安装和初始设置

安装 Alembic

# Install Alembic with PostgreSQL support
pip install alembic psycopg2-binary sqlalchemy

# Or add to requirements.txt
alembic>=1.13.0
sqlalchemy>=2.0.0
psycopg2-binary>=2.9.0

在项目中初始化 Alembic

# Initialize Alembic (creates alembic/ directory and alembic.ini)
alembic init alembic

# For multiple database support
alembic init --template multidb alembic

这会创建：

• alembic/：包含迁移脚本的目录
• alembic/versions/：存放单个迁移文件的位置
• alembic/env.py：迁移环境配置
• alembic.ini：Alembic 配置文件

配置数据库连接

编辑 alembic.ini 以设置数据库 URL：

# For development
sqlalchemy.url = postgresql://user:password@localhost/support_dev

# For production (use environment variables)
sqlalchemy.url = postgresql://%(DB_USER)s:%(DB_PASSWORD)s@%(DB_HOST)s/%(DB_NAME)s

更好的做法是在 env.py 中使用环境变量：

import os
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# Import your models
from myapp.models import Base

# This is the Alembic Config object
config = context.config

# Override sqlalchemy.url from environment
db_url = os.getenv('DATABASE_URL', 'postgresql://localhost/support_dev')
config.set_main_option('sqlalchemy.url', db_url)

# Set up target metadata for autogenerate
target_metadata = Base.metadata

创建迁移

手动创建迁移

需要精确控制时，可手动创建迁移：

# Create empty migration file
alembic revision -m "add ticket priority column"

这会生成类似 versions/abc123_add_ticket_priority_column.py 的文件：

"""add ticket priority column

Revision ID: abc123
Revises: def456
Create Date: 2025-01-15 10:30:00.000000
"""

from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = 'abc123'
down_revision = 'def456'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Add priority column to tickets table
    op.add_column('tickets',
        sa.Column('priority', sa.String(20), nullable=True, server_default='normal')
    )

    # Create index for performance
    op.create_index('ix_tickets_priority', 'tickets', ['priority'])

def downgrade() -> None:
    # Remove index first
    op.drop_index('ix_tickets_priority', 'tickets')

    # Remove column
    op.drop_column('tickets', 'priority')

自动生成迁移

让 Alembic 自动检测架构变更：

# Generate migration by comparing models to database
alembic revision --autogenerate -m "add customer satisfaction table"

重要：务必审查自动生成的迁移！它们可能会遗漏：

• 重命名列（会表现为删除列并添加新列）
• 需要进行数据转换的列类型变更
• 复杂约束
• 数据迁移

自动生成的迁移示例：

"""add customer satisfaction table

Revision ID: xyz789
Revises: abc123
Create Date: 2025-01-15 11:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = 'xyz789'
down_revision = 'abc123'
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Auto-generated - review before running!
    op.create_table(
        'customer_satisfaction',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('ticket_id', sa.Integer(), nullable=False),
        sa.Column('rating', sa.Integer(), nullable=False),
        sa.Column('feedback', sa.Text(), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(['ticket_id'], ['tickets.id'], ondelete='CASCADE'),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index('ix_satisfaction_ticket_id', 'customer_satisfaction', ['ticket_id'])
    op.create_index('ix_satisfaction_created_at', 'customer_satisfaction', ['created_at'])

def downgrade() -> None:
    op.drop_index('ix_satisfaction_created_at', 'customer_satisfaction')
    op.drop_index('ix_satisfaction_ticket_id', 'customer_satisfaction')
    op.drop_table('customer_satisfaction')

数据迁移

在架构变更期间迁移数据

需要转换现有数据时：

"""convert ticket status to new enum

Revision ID: data001
Revises: xyz789
Create Date: 2025-01-15 12:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column

revision = 'data001'
down_revision = 'xyz789'

def upgrade() -> None:
    # Create new status column
    op.add_column('tickets',
        sa.Column('status_new', sa.String(50), nullable=True)
    )

    # Migrate data using bulk update
    tickets = table('tickets',
        column('status', sa.String),
        column('status_new', sa.String)
    )

    # Map old statuses to new ones
    status_mapping = {
        'open': 'OPEN',
        'in_progress': 'IN_PROGRESS',
        'pending': 'WAITING_ON_CUSTOMER',
        'resolved': 'RESOLVED',
        'closed': 'CLOSED'
    }

    connection = op.get_bind()
    for old_status, new_status in status_mapping.items():
        connection.execute(
            tickets.update().where(
                tickets.c.status == old_status
            ).values(status_new=new_status)
        )

    # Make new column non-nullable now that data is migrated
    op.alter_column('tickets', 'status_new', nullable=False)

    # Drop old column and rename new one
    op.drop_column('tickets', 'status')
    op.alter_column('tickets', 'status_new', new_column_name='status')

def downgrade() -> None:
    # Reverse the migration
    op.add_column('tickets',
        sa.Column('status_old', sa.String(50), nullable=True)
    )

    tickets = table('tickets',
        column('status', sa.String),
        column('status_old', sa.String)
    )

    # Reverse mapping
    reverse_mapping = {
        'OPEN': 'open',
        'IN_PROGRESS': 'in_progress',
        'WAITING_ON_CUSTOMER': 'pending',
        'RESOLVED': 'resolved',
        'CLOSED': 'closed'
    }

    connection = op.get_bind()
    for new_status, old_status in reverse_mapping.items():
        connection.execute(
            tickets.update().where(
                tickets.c.status == new_status
            ).values(status_old=old_status)
        )

    op.alter_column('tickets', 'status_old', nullable=False)
    op.drop_column('tickets', 'status')
    op.alter_column('tickets', 'status_old', new_column_name='status')

使用分批处理进行大规模数据迁移

对于大型表，请分批处理数据：

"""add computed resolution time to tickets

Revision ID: data002
Revises: data001
Create Date: 2025-01-15 13:00:00.000000
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column, select

revision = 'data002'
down_revision = 'data001'

def upgrade() -> None:
    # Add new column
    op.add_column('tickets',
        sa.Column('resolution_time_seconds', sa.Integer(), nullable=True)
    )

    connection = op.get_bind()
    tickets = table('tickets',
        column('id', sa.Integer),
        column('created_at', sa.DateTime),
        column('resolved_at', sa.DateTime),
        column('resolution_time_seconds', sa.Integer)
    )

    # Process in batches to avoid memory issues
    batch_size = 1000
    offset = 0

    while True:
        # Get batch of tickets that need processing
        batch = connection.execute(
            select(
                tickets.c.id,
                tickets.c.created_at,
                tickets.c.resolved_at
            ).where(
                sa.and_(
                    tickets.c.resolved_at.isnot(None),
                    tickets.c.resolution_time_seconds.is_(None)
                )
            ).limit(batch_size).offset(offset)
        ).fetchall()

        if not batch:
            break

        # Update batch
        for row in batch:
            if row.resolved_at and row.created_at:
                resolution_time = (row.resolved_at - row.created_at).total_seconds()
                connection.execute(
                    tickets.update().where(
                        tickets.c.id == row.id
                    ).values(resolution_time_seconds=int(resolution_time))
                )

        offset += batch_size

    # Now make column non-nullable for future rows
    op.alter_column('tickets', 'resolution_time_seconds',
        nullable=False, server_default='0')

def downgrade() -> None:
    op.drop_column('tickets', 'resolution_time_seconds')

运行迁移

将数据库升级到最新版本

# Upgrade to latest revision (head)
alembic upgrade head

# See what would be executed (SQL only, don't run)
alembic upgrade head --sql

# Upgrade one step at a time
alembic upgrade +1

# Upgrade to specific revision
alembic upgrade abc123

降级数据库

# Downgrade one revision
alembic downgrade -1

# Downgrade to specific revision
alembic downgrade abc123

# Downgrade to base (empty database)
alembic downgrade base

# Generate SQL for downgrade without executing
alembic downgrade -1 --sql

检查当前状态

# Show current database revision
alembic current

# Show current revision with details
alembic current --verbose

# Show migration history
alembic history

# Show history with current revision marked
alembic history --indicate-current

# Show specific revision range
alembic history -r base:head

分支与合并

为什么要创建迁移分支？

在客户支持系统中，你可能会遇到：

• 功能分支：并行开发的新功能
• 热修复分支：无法等待功能开发完成的紧急修复
• 团队分支：多个团队负责不同模块

创建分支

# Create base for new branch
alembic revision -m "create reporting branch" \
    --head=base \
    --branch-label=reporting \
    --version-path=alembic/versions/reporting

# Add migration to specific branch
alembic revision -m "add report tables" \
    --head=reporting@head

分支结构示例：

base
├── main branch
│   ├── abc123: initial schema
│   ├── def456: add tickets
│   └── ghi789: add users
└── reporting branch
    ├── rep001: create reports table
    └── rep002: add scheduled reports

使用多个分支

# Show all branch heads
alembic heads

# Show branch points
alembic branches

# Upgrade specific branch
alembic upgrade reporting@head

# Upgrade all branches
alembic upgrade heads

合并分支

功能准备好合并时：

# Merge two branches
alembic merge -m "merge reporting into main" \
    main@head reporting@head

生成的合并迁移：

"""merge reporting into main

Revision ID: merge001
Revises: ghi789, rep002
Create Date: 2025-01-15 14:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = 'merge001'
down_revision = ('ghi789', 'rep002')  # Multiple parents
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Usually empty for simple merges
    # Add code if you need to reconcile conflicting changes
    pass

def downgrade() -> None:
    pass

跨分支依赖

一个分支依赖另一个分支时：

# Create migration that depends on specific revision from another branch
alembic revision -m "reporting needs user table" \
    --head=reporting@head \
    --depends-on=def456  # Revision from main branch

测试迁移

迁移的单元测试

# tests/test_migrations.py
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def alembic_config():
    """Provide Alembic configuration for testing"""
    config = Config("alembic.ini")
    config.set_main_option(
        "sqlalchemy.url",
        "postgresql://localhost/support_test"
    )
    return config

@pytest.fixture
def test_db(alembic_config):
    """Create test database and apply migrations"""
    # Create engine
    engine = create_engine(
        alembic_config.get_main_option("sqlalchemy.url")
    )

    # Run migrations to head
    command.upgrade(alembic_config, "head")

    yield engine

    # Cleanup - downgrade to base
    command.downgrade(alembic_config, "base")
    engine.dispose()

def test_migration_creates_tickets_table(test_db):
    """Test that migrations create expected tables"""
    inspector = inspect(test_db)
    tables = inspector.get_table_names()

    assert 'tickets' in tables
    assert 'users' in tables
    assert 'customer_satisfaction' in tables

def test_tickets_table_structure(test_db):
    """Test ticket table has correct columns"""
    inspector = inspect(test_db)
    columns = {col['name']: col for col in inspector.get_columns('tickets')}

    assert 'id' in columns
    assert 'priority' in columns
    assert 'status' in columns
    assert 'created_at' in columns
    assert 'resolution_time_seconds' in columns

    # Check column types
    assert columns['priority']['type'].python_type == str
    assert columns['status']['type'].python_type == str

def test_migration_upgrade_downgrade_cycle(alembic_config):
    """Test that upgrade -> downgrade -> upgrade works"""
    # Start at base
    command.downgrade(alembic_config, "base")

    # Upgrade to head
    command.upgrade(alembic_config, "head")

    # Downgrade one step
    command.downgrade(alembic_config, "-1")

    # Upgrade back to head
    command.upgrade(alembic_config, "head")

    # Should complete without errors

def test_data_migration_preserves_data(test_db):
    """Test that data migrations don't lose data"""
    from sqlalchemy.orm import sessionmaker
    from myapp.models import Ticket

    Session = sessionmaker(bind=test_db)
    session = Session()

    # Insert test data
    ticket = Ticket(
        title="Test ticket",
        status="OPEN",
        priority="high"
    )
    session.add(ticket)
    session.commit()
    ticket_id = ticket.id
    session.close()

    # Run a migration that modifies tickets table
    # (This would be a specific revision)
    # command.upgrade(alembic_config, "specific_revision")

    # Verify data still exists
    session = Session()
    retrieved = session.query(Ticket).filter_by(id=ticket_id).first()
    assert retrieved is not None
    assert retrieved.title == "Test ticket"
    session.close()

集成测试

# tests/test_migration_integration.py
import pytest
from alembic import command
from alembic.config import Config
from alembic.script import ScriptDirectory
from alembic.runtime.migration import MigrationContext

def test_no_pending_migrations(alembic_config, test_db):
    """Ensure all migrations are applied in test environment"""
    script = ScriptDirectory.from_config(alembic_config)

    with test_db.connect() as connection:
        context = MigrationContext.configure(connection)
        current_heads = set(context.get_current_heads())
        script_heads = set(script.get_heads())

        assert current_heads == script_heads, \
            f"Database has pending migrations. Current: {current_heads}, Expected: {script_heads}"

def test_migration_order_is_valid(alembic_config):
    """Verify migration chain has no gaps or conflicts"""
    script = ScriptDirectory.from_config(alembic_config)

    # Get all revisions
    revisions = list(script.walk_revisions())

    # Check each revision has valid down_revision
    for revision in revisions:
        if revision.down_revision is not None:
            if isinstance(revision.down_revision, tuple):
                # Merge point
                for down_rev in revision.down_revision:
                    assert script.get_revision(down_rev) is not None
            else:
                assert script.get_revision(revision.down_revision) is not None

def test_check_command_detects_drift(alembic_config, test_db):
    """Test that check command detects schema drift"""
    # This test verifies that `alembic check` works correctly
    try:
        command.check(alembic_config)
        # If no exception, database matches models
        assert True
    except Exception as e:
        # If exception, there's drift between DB and models
        pytest.fail(f"Schema drift detected: {e}")

测试迁移性能

# tests/test_migration_performance.py
import time
import pytest
from alembic import command

def test_migration_completes_within_time_limit(alembic_config):
    """Ensure migrations complete within acceptable time"""
    # Downgrade to base
    command.downgrade(alembic_config, "base")

    # Time the upgrade
    start = time.time()
    command.upgrade(alembic_config, "head")
    duration = time.time() - start

    # Assert completes within 60 seconds
    assert duration < 60, f"Migration took {duration}s, exceeds 60s limit"

@pytest.mark.slow
def test_data_migration_with_large_dataset(alembic_config, test_db):
    """Test data migration performance with realistic data volume"""
    from sqlalchemy.orm import sessionmaker
    from myapp.models import Ticket

    Session = sessionmaker(bind=test_db)
    session = Session()

    # Create 10,000 test tickets
    tickets = [
        Ticket(
            title=f"Test ticket {i}",
            status="OPEN",
            priority="normal"
        )
        for i in range(10000)
    ]
    session.bulk_save_objects(tickets)
    session.commit()
    session.close()

    # Run data migration and measure time
    start = time.time()
    command.upgrade(alembic_config, "data002")  # Specific data migration
    duration = time.time() - start

    # Should process 10k records in reasonable time
    assert duration < 30, f"Data migration took {duration}s for 10k records"

CI/CD 集成

GitHub Actions 工作流

# .github/workflows/migrations.yml
name: Database Migrations

on:
  pull_request:
    paths:
      - 'alembic/versions/**'
      - 'myapp/models/**'
      - 'alembic.ini'
      - 'alembic/env.py'
  push:
    branches:
      - main
      - develop

jobs:
  test-migrations:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: support_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov

      - name: Run migration tests
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost/support_test
        run: |
          # Test upgrade to head
          alembic upgrade head

          # Test downgrade to base
          alembic downgrade base

          # Test upgrade again
          alembic upgrade head

          # Run pytest for migration tests
          pytest tests/test_migrations.py -v

      - name: Check for schema drift
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost/support_test
        run: |
          alembic check

      - name: Validate migration history
        run: |
          # Check for multiple heads (should be only one)
          HEADS_COUNT=$(alembic heads | wc -l)
          if [ "$HEADS_COUNT" -gt 1 ]; then
            echo "ERROR: Multiple heads detected. Please merge branches."
            alembic heads
            exit 1
          fi

  review-migration-sql:
    runs-on: ubuntu-latest
    if: github.event_name == 'pull_request'

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Generate SQL for review
        run: |
          # Generate SQL without executing
          alembic upgrade head --sql > migration.sql

      - name: Upload SQL artifact
        uses: actions/upload-artifact@v3
        with:
          name: migration-sql
          path: migration.sql

      - name: Comment PR with SQL
        uses: actions/github-script@v6
        with:
          script: |
            const fs = require('fs');
            const sql = fs.readFileSync('migration.sql', 'utf8');

            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: `## Migration SQL\n\n\`\`\`sql\n${sql}\n\`\`\``
            });

部署脚本

#!/bin/bash
# scripts/deploy_migrations.sh

set -e  # Exit on error

echo "Starting database migration deployment..."

# Environment variables
DB_HOST="${DB_HOST:-localhost}"
DB_NAME="${DB_NAME:-support_prod}"
DB_USER="${DB_USER:-postgres}"
DATABASE_URL="postgresql://${DB_USER}:${DB_PASSWORD}@${DB_HOST}/${DB_NAME}"

# Configuration
BACKUP_DIR="./backups"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
BACKUP_FILE="${BACKUP_DIR}/pre_migration_${TIMESTAMP}.sql"

# Create backup directory
mkdir -p "$BACKUP_DIR"

# 1. Backup database before migration
echo "Creating database backup..."
pg_dump "$DATABASE_URL" > "$BACKUP_FILE"
echo "Backup created: $BACKUP_FILE"

# 2. Check current migration status
echo "Current migration status:"
alembic current

# 3. Show pending migrations
echo "Pending migrations:"
alembic history --verbose | grep -A 5 "head"

# 4. Run migrations with timeout
echo "Running migrations..."
timeout 300 alembic upgrade head || {
    echo "ERROR: Migration failed or timed out!"
    echo "Restoring from backup..."
    psql "$DATABASE_URL" < "$BACKUP_FILE"
    exit 1
}

# 5. Verify migration success
echo "Verifying migration status..."
CURRENT_REV=$(alembic current | grep "Rev:" | awk '{print $2}')
HEAD_REV=$(alembic heads | awk '{print $1}')

if [ "$CURRENT_REV" != "$HEAD_REV" ]; then
    echo "ERROR: Migration incomplete. Current: $CURRENT_REV, Expected: $HEAD_REV"
    echo "Restoring from backup..."
    psql "$DATABASE_URL" < "$BACKUP_FILE"
    exit 1
fi

echo "Migration completed successfully!"
echo "Current revision: $CURRENT_REV"

# 6. Cleanup old backups (keep last 10)
echo "Cleaning up old backups..."
ls -t "$BACKUP_DIR"/*.sql | tail -n +11 | xargs -r rm

echo "Deployment complete!"

生产环境最佳实践

部署前检查清单

• 已在开发环境中测试迁移
• 已使用类似生产环境的数据在预发布环境中测试迁移
• 至少有一位团队成员审查了迁移
• 已测试并验证降级路径
• 已评估对大型表的性能影响
• 已制定数据库备份计划
• 已记录回滚流程
• 已安排维护窗口（如有需要）
• 已通知团队部署事宜
• 已配置监控警报

零停机迁移

对于无法停机的关键支持系统：

阶段 1：添加变更

"""add new column (phase 1)

Revision ID: zd001
"""

def upgrade() -> None:
    # Add new column as nullable
    op.add_column('tickets',
        sa.Column('new_field', sa.String(100), nullable=True)
    )

def downgrade() -> None:
    op.drop_column('tickets', 'new_field')

阶段 2：数据迁移（后台）

"""populate new column (phase 2)

Revision ID: zd002
"""

def upgrade() -> None:
    # Update in small batches during low-traffic periods
    connection = op.get_bind()

    batch_size = 100
    while True:
        result = connection.execute(
            """
            UPDATE tickets
            SET new_field = calculate_value(old_field)
            WHERE new_field IS NULL
            LIMIT {batch_size}
            """.format(batch_size=batch_size)
        )
        if result.rowcount == 0:
            break

        # Small delay to reduce database load
        import time
        time.sleep(0.1)

def downgrade() -> None:
    connection = op.get_bind()
    connection.execute("UPDATE tickets SET new_field = NULL")

阶段 3：设为必需

"""make new column required (phase 3)

Revision ID: zd003
"""

def upgrade() -> None:
    # Now that all rows have values, make it non-nullable
    op.alter_column('tickets', 'new_field',
        nullable=False,
        server_default='default_value'
    )

def downgrade() -> None:
    op.alter_column('tickets', 'new_field',
        nullable=True,
        server_default=None
    )

阶段 4：删除旧列（可选）

"""remove old column (phase 4)

Revision ID: zd004
"""

def upgrade() -> None:
    op.drop_column('tickets', 'old_field')

def downgrade() -> None:
    op.add_column('tickets',
        sa.Column('old_field', sa.String(100), nullable=True)
    )

处理迁移失败

# alembic/env.py additions for error handling

from alembic import context
import logging

logger = logging.getLogger('alembic.env')

def run_migrations_online():
    """Run migrations in 'online' mode with error handling"""

    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,
            transaction_per_migration=True,  # Rollback individual migrations
            compare_type=True,
            compare_server_default=True
        )

        try:
            with context.begin_transaction():
                context.run_migrations()

        except Exception as e:
            logger.error(f"Migration failed: {e}")
            logger.error("Rolling back transaction...")
            # Transaction automatically rolled back
            raise

        else:
            logger.info("Migration completed successfully")

高级配置

自定义迁移模板

为组织创建自定义模板：

# alembic/script.py.mako
"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

Author: ${author if author else 'Support Team'}
Jira: ${jira_ticket if jira_ticket else 'N/A'}
"""

from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}


def upgrade() -> None:
    """Apply migration changes"""
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    """Revert migration changes"""
    ${downgrades if downgrades else "pass"}

多数据库支持

对于使用独立数据库的系统（例如，主数据库 + 分析数据库）：

# alembic/env.py for multiple databases

def run_migrations_online():
    """Run migrations for multiple databases"""

    # Configuration for each database
    engines = {
        'main': {
            'url': os.getenv('MAIN_DB_URL'),
            'target_metadata': main_metadata
        },
        'analytics': {
            'url': os.getenv('ANALYTICS_DB_URL'),
            'target_metadata': analytics_metadata
        }
    }

    for name, config in engines.items():
        logger.info(f"Running migrations for {name} database")

        engine = create_engine(config['url'])

        with engine.connect() as connection:
            context.configure(
                connection=connection,
                target_metadata=config['target_metadata'],
                upgrade_token=f"{name}_upgrade",
                downgrade_token=f"{name}_downgrade"
            )

            with context.begin_transaction():
                context.run_migrations(engine_name=name)

故障排除

常见问题和解决方案

多个头节点错误

# Problem: "Multiple heads exist"
# Solution: Merge the branches
alembic merge heads -m "merge branches"

迁移不同步

# Problem: Database revision doesn't match migration history
# Solution: Stamp database to specific revision
alembic stamp head

# Or stamp to specific revision
alembic stamp abc123

清理失败的迁移

# Problem: Migration failed midway
# Solution: Manual cleanup

# 1. Check current state
alembic current

# 2. Manually fix database issues
psql $DATABASE_URL

# 3. Stamp to correct revision
alembic stamp previous_working_revision

# 4. Try migration again
alembic upgrade head

循环依赖

# Problem: "Circular dependency detected"
# Solution: Use depends_on instead of down_revision
alembic revision -m "fix circular dependency" \
    --head=branch_a@head \
    --depends-on=branch_b_revision

总结

本技能介绍了在客户支持系统中全面使用 Alembic 的方法：

1. 设置：安装、配置和初始化
2. 创建迁移：手动和自动生成的方法
3. 数据迁移：在架构变更期间转换数据
4. 运行迁移：升级、降级和状态命令
5. 分支：管理并行开发流
6. 测试：单元、集成和性能测试
7. CI/CD：自动化和部署策略
8. 生产环境：零停机迁移和最佳实践
9. 高级功能：自定义模板和多数据库支持
10. 故障排除：常见问题和解决方案

请始终牢记：

• 审查自动生成的迁移
• 在生产环境中应用迁移前进行全面测试
• 执行重大迁移前做好备份
• 规划回滚方案
• 监控迁移性能
• 记录复杂迁移

更多示例请参阅此技能包中的 EXAMPLES.md。