如何基于SQLAlchemy引擎通过代码手动执行Alembic迁移（无初始化配置）

核心思路 #

直接调用Alembic底层API，跳过命令行工具和传统配置文件，用已有的SQLAlchemy引擎绑定迁移环境，手动触发升级/降级操作，或直接执行自定义迁移SQL/ORM操作。

方案1：基于现有迁移脚本目录执行迁移 #

如果已经有编写好的Alembic迁移脚本（migrations/versions下的版本文件），可通过以下代码直接执行：

from sqlalchemy import create_engine
from alembic.config import Config
from alembic.script import ScriptDirectory
from alembic.runtime.environment import EnvironmentContext
from alembic import command

# 1. 初始化你的SQLAlchemy引擎（替换为实际数据库URL）
engine = create_engine("mysql+pymysql://user:password@localhost/your_db")

# 2. 手动构建Alembic配置，无需ini文件
alembic_cfg = Config()
# 指定迁移脚本所在目录（绝对/相对路径均可）
alembic_cfg.set_main_option("script_location", "./migrations")
# 绑定数据库URL（直接复用SQLAlchemy引擎的URL）
alembic_cfg.set_main_option("sqlalchemy.url", engine.url.render_as_string(hide_password=False))

# 3. 加载脚本目录并执行迁移
script_dir = ScriptDirectory.from_config(alembic_cfg)

# 4. 在事务中执行升级（可替换为"downgrade" + 版本号执行降级）
with engine.begin() as connection:
    env_ctx = EnvironmentContext(
        alembic_cfg,
        script_dir,
        connection=connection,
        target_metadata=None  # 若需自动生成迁移，传入你的SQLAlchemy Base.metadata
    )
    with env_ctx.begin_transaction():
        command.upgrade(alembic_cfg, "head")

方案2：完全自定义迁移操作（无迁移脚本目录） #

如果不需要预定义的迁移脚本，可直接通过Alembic的Operations对象执行SQL/ORM级别的迁移操作，同时手动管理版本记录：

from sqlalchemy import create_engine, Integer, String
from alembic.operations import Operations
from alembic.migration import MigrationContext

# 1. 初始化SQLAlchemy引擎
engine = create_engine("postgresql://user:pass@localhost/your_db")

with engine.begin() as connection:
    # 2. 创建迁移上下文，绑定数据库连接
    migration_ctx = MigrationContext.configure(connection)
    # 3. 获取操作对象，用于执行迁移指令
    op = Operations(migration_ctx)

    # 4. 执行具体迁移操作示例
    # 创建新表
    op.create_table(
        'user_profiles',
        op.column('id', Integer, primary_key=True, autoincrement=True),
        op.column('user_id', Integer, nullable=False),
        op.column('bio', String(500), nullable=True)
    )
    # 添加外键约束
    op.create_foreign_key(
        'fk_user_profile_user',
        'user_profiles',
        'users',
        ['user_id'],
        ['id']
    )
    # 手动记录版本（若需跟踪迁移历史）
    op.execute("INSERT INTO alembic_version (version_num) VALUES ('custom_20240520_1234')")

关键注意事项 #

• 事务安全：始终用engine.begin()包裹迁移操作，确保出错时自动回滚
• 版本管理：方案2中需手动维护alembic_version表，避免重复执行迁移
• 环境适配：确保迁移操作与目标数据库兼容（如MySQL与PostgreSQL的SQL语法差异）
• 权限要求：数据库用户需具备创建/修改表、写入alembic_version表的权限

内容的提问来源于stack exchange，提问作者Denny Weinberg