Alembic 重置自动化脚本说明文档

1. 背景与用途

在使用 Alembic 进行数据库迁移管理时，项目可能出现以下问题：

• 迁移版本历史断裂，某些迁移文件丢失或被删除导致无法执行迁移
• 数据库中的 Alembic 版本表 alembic_version 与代码迁移文件不匹配
• 需要快速重置 Alembic 迁移状态，生成全新的初始迁移文件

针对上述场景，本脚本旨在实现一键：

• 删除数据库中 Alembic 版本控制表
• 清理旧的 Alembic 迁移文件目录
• 重新初始化 Alembic 迁移目录结构
• 自动写入符合项目业务需求的 env.py 配置
• 自动生成新的“初始迁移”迁移文件
• 标记数据库当前版本为最新，避免重复执行历史迁移

极大简化手工操作步骤，提升开发效率。

2. 脚本文件：reset_alembic.py

2.1 功能概述

• 删除 alembic_version 表（如果存在）
• 删除旧迁移文件夹（默认 migrations）
• 运行 alembic init 初始化迁移目录
• 自动覆盖写入符合当前项目需求的 env.py 文件（用于模型元数据导入和数据库连接）
• 自动执行 alembic revision --autogenerate 生成初始迁移脚本
• 执行 alembic stamp head 标记数据库版本

2.2 关键配置项

• DATABASE_URL （数据库连接字符串）：

  格式请确保正确，示例：

  mysql+pymysql://root:password@127.0.0.1:3306/your_database

  根据项目实际替换用户名、密码、主机和数据库名。

• ALEMBIC_DIR：

  Alembic 迁移目录，一般为 migrations，如果项目中迁移目录名称不同，请修改此项。

• ALEMBIC_INI：

  Alembic 配置文件路径，默认 alembic.ini，如果你的配置文件在其他位置或名称，需相应修改。

2.3 重要逻辑模块

• drop_alembic_version_table()：

  使用 SQLAlchemy 连接数据库，删除 alembic_version 表，解除版本锁定。

• remove_old_migrations()：

  删除旧的 Alembic 迁移目录及其中所有迁移文件。

• init_alembic()：

  执行 Alembic 初始化命令，新建迁移目录。

• overwrite_env_py()：

  将预定义的 env.py 重写到迁移目录中，确保 Alembic 正确加载项目模型和数据库连接。

• generate_revision()：

  使用 Alembic 的自动生成功能，基于当前模型创建初始迁移脚本。

• stamp_head()：

  标记数据库当前版本为最新，避免 Alembic 重复应用历史迁移。

3. 使用说明

3.1 环境准备

• 确保安装 Python 环境，版本推荐3.7及以上。

• 安装依赖库：

  pip install alembic sqlalchemy pymysql

• 确认项目中存在 alembic.ini，且配置项 sqlalchemy.url 指向正确的数据库。

3.2 操作步骤

1. 将 reset_alembic.py 脚本放置于项目根目录或合适位置。

2. 打开脚本，修改以下配置：

   • DATABASE_URL: 换成你项目数据库连接字符串
   • ALEMBIC_DIR: 如项目中迁移目录名非 migrations 请修改
   • ALEMBIC_INI: 配置文件路径，如非位于根目录按实际填写

3. 运行脚本：

   python reset_alembic.py

4. 脚本执行完毕后，会完成 Alembic 重置并生成新的迁移文件，你可以正常使用 Alembic 管理数据库迁移。

3.3 注意事项

• 建议仅在开发环境或测试环境使用该脚本。
• 该过程会删除数据库中的 Alembic 版本表，谨防误操作导致历史迁移丢失。
• 使用前请备份数据库和代码。
• 脚本内置的 env.py 内容需根据项目架构和 ORM 模型实际情况调整。

4. 相关说明

• Alembic 的 env.py 是迁移引擎的核心，负责加载模型元数据与数据库引擎。
• 自动生成迁移依赖于模型元数据与数据库实际结构的差异分析。
• stamp head 指令仅标记版本，不修改数据库结构。

5. 脚本示例附录

请参考本页或项目源码中的 reset_alembic.py 完整代码。

注意是否需要打开 每个项目独立的db model 的注释

from apis.models import *  # 每个项目独立的db model  
from commonlib.models import *  # 确保所有的表都被检测到  
from commonlib.commonDbEngine import Base, create_db_engine  

reset_alembic.py 文件

import os
import shutil
import subprocess
from sqlalchemy import create_engine, text

from commonlib.commonDbConfig import getDdataBaseConfigFunc

dbc = getDdataBaseConfigFunc("local")
DATABASE_URL = f"mysql+pymysql://{dbc.db_user}:{dbc.db_pwd}@{dbc.db_host}:{dbc.db_port}/{dbc.db_warehouse}"
ALEMBIC_DIR = "migrations"
ALEMBIC_INI = "alembic.ini"

ENV_PY_CONTENT = """\
from logging.config import fileConfig  

from sqlalchemy import engine_from_config  
from sqlalchemy import pool  

from alembic import context  

from apis.models import *  # 每个项目独立的db model  
from commonlib.models import *  # 确保所有的表都被检测到  
from commonlib.commonDbEngine import Base, create_db_engine  

config = context.config  

if config.config_file_name is not None:  
    fileConfig(config.config_file_name)  

target_metadata = Base.metadata  
print("检测到的表：", Base.metadata.tables.keys())  


def run_migrations_offline() -> None:  
    url = config.get_main_option("sqlalchemy.url")  
    context.configure(  
        url=url,  
        target_metadata=target_metadata,  
        literal_binds=True,  
        dialect_opts={"paramstyle": "named"},  
    )  

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


def run_migrations_online() -> None:  
    connectable = create_db_engine()  
    with connectable.connect() as connection:  
        context.configure(connection=connection, target_metadata=target_metadata)  
        with context.begin_transaction():  
            context.run_migrations()  


if context.is_offline_mode():  
    run_migrations_offline()  
else:  
    run_migrations_online()
"""


def drop_alembic_version_table():
    engine = create_engine(DATABASE_URL)
    with engine.connect() as conn:
        print("删除 alembic_version 表（如果存在）...")
        conn.execute(text("DROP TABLE IF EXISTS alembic_version"))
        print("删除完成。\n")


def remove_old_migrations():
    if os.path.exists(ALEMBIC_DIR):
        print(f"删除旧迁移目录：{ALEMBIC_DIR} ...")
        shutil.rmtree(ALEMBIC_DIR)
        print("删除完成。\n")
    else:
        print("没有找到旧迁移目录，跳过删除。\n")


def init_alembic():
    print("初始化 Alembic 迁移目录...")
    subprocess.run(["alembic", "init", ALEMBIC_DIR], check=True)
    print("初始化完成。\n")


def overwrite_env_py():
    env_py_path = os.path.join(ALEMBIC_DIR, "env.py")
    print(f"覆盖写入 {env_py_path} ...")
    with open(env_py_path, "w", encoding="utf-8") as f:
        f.write(ENV_PY_CONTENT)
    print("env.py 覆盖完成。\n")


def generate_revision():
    print("自动生成初始迁移文件...")
    subprocess.run(
        [
            "alembic",
            "-c",
            ALEMBIC_INI,
            "revision",
            "--autogenerate",
            "-m",
            "initial migration",
        ],
        check=True,
    )
    print("初始迁移文件生成完成。\n")


def stamp_head():
    print("执行 alembic stamp head 将数据库版本标记为最新...")
    subprocess.run(["alembic", "-c", ALEMBIC_INI, "stamp", "head"], check=True)
    print("数据库版本标记完成。\n")


def main():
    drop_alembic_version_table()
    remove_old_migrations()
    init_alembic()
    overwrite_env_py()
    generate_revision()
    stamp_head()
    print("Alembic 重置并初始化完成！")


if __name__ == "__main__":
    main()