Alembic迁移脚本冲突的智能检测与优雅合并之道
title: Alembic迁移脚本冲突的智能检测与优雅合并之道


# check_migration_conflicts.py
from alembic.config import Config
from alembic.script import ScriptDirectory
def detect_conflicts():
config = Config("alembic.ini")
scripts = ScriptDirectory.from_config(config)
# 获取当前分支的所有版本
heads = scripts.get_heads()
if len(heads) > 1:
print(f"⚠️ 检测到多个头版本:{heads}")
# 可视化显示分支结构
for revision in heads:
script = scripts.get_revision(revision)
print(f"分支 {revision}:")
for rev in script.iterate_revisions(script.down_revision, False):
print(f" ← {rev.revision}")
else:
print("✅ 无版本冲突")
if __name__ == "__main__":
detect_conflicts()
python check_migration_conflicts.py
alembic history --verbose
alembic revision -m "merge_branch" --head a1b2c3d4,b5e6f7g8
# migrations/versions/xxxx_merge_branch.py
def upgrade():
# 按正确顺序执行两个分支的修改
op.execute("ALTER TABLE users ADD COLUMN merged_flag BOOLEAN")
op.alter_column('posts', 'content_type',
existing_type=sa.VARCHAR(length=50),
nullable=False)
# 添加合并标记
op.create_table(
'migration_merge_records',
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('merged_version', sa.String(32))
)
import pytest
from alembic.command import upgrade, downgrade
from alembic.config import Config
@pytest.fixture
def alembic_config():
return Config("alembic.ini")
def test_merged_migration_upgrade(alembic_config):
try:
upgrade(alembic_config, "head")
# 验证合并后的表结构
with alembic_config.connection() as conn:
result = conn.execute("SHOW TABLES LIKE 'migration_merge_records'")
assert result.fetchone() is not None
finally:
downgrade(alembic_config, "base")
def test_conflict_resolution_consistency(alembic_config):
upgrade(alembic_config, "head")
downgrade(alembic_config, "-1")
upgrade(alembic_config, "+1")
# 验证回滚后重新升级是否一致
with alembic_config.connection() as conn:
result = conn.execute("DESC users")
columns = [row[0] for row in result]
assert 'merged_flag' in columns
alembic.util.exc.CommandError: Multiple head revisions are present
sqlalchemy.exc.OperationalError: (MySQL Error)无法修改字段类型
sqlalchemy.exc.ProgrammingError: 表'migration_merge_records'已存在
评论
发表评论