Alembic op.get_bind 怎么在迁移脚本中执行参数化 SQL

Alembic 迁移脚本需要把一批旧数据改成新状态时，最稳妥的写法是：在 upgrade() 或 downgrade() 中调用 op.get_bind() 拿到当前 Connection，再用 SQLAlchemy 的 text() 和绑定参数执行 SQL。变量放在参数字典里，SQL 结构保持不变；不要把用户数据拼进字符串，也不要在迁移函数里自行创建、提交或关闭连接。

官方资料：https://alembic.sqlalchemy.org/en/latest/api/runtime.html

先说结论：需要 Connection 时才用 op.get_bind

Alembic 的 op.execute() 适合把一个迁移操作交给当前变更上下文，例如执行一条 DDL 或不需要读取结果的 SQL。它的返回值不是给业务逻辑使用的结果集。若迁移需要查询旧值、根据结果决定后续更新，或者需要检查受影响行数，就应该拿到当前连接：

from alembic import op


def upgrade():
    # 取得 Alembic 当前迁移上下文中的连接，不另建 Engine。
    connection = op.get_bind()
    # 后续的 text、select、update 等 SQLAlchemy Core 语句都交给它执行。
    result = connection.execute(...)
    return result.rowcount

这里的关键不是“换一种写法执行 SQL”，而是获得一个可以继续调用 execute()、读取 Result、保持当前事务语义的 SQLAlchemy Connection。Alembic 官方运行时文档也把 get_bind() 定义为返回当前 bind；在 online 模式下，它就是当前用于发出 SQL 的连接。

为什么不直接把变量拼进 SQL 字符串

迁移里的变量可能来自配置、旧表数据或预先计算的映射。把它们直接拼接到 SQL 中，会让引号、类型转换和特殊字符都变成调用方的责任；当迁移跨数据库方言时，问题更难定位。

SQLAlchemy 的 text() 使用命名占位符，执行时再把值交给连接。需要固定类型时，可以用 bindparam() 明确声明：

from alembic import op
from sqlalchemy import String, bindparam, text


def upgrade():
    connection = op.get_bind()
    statement = text("""
        UPDATE account
        SET display_name = :new_name
        WHERE tenant_id = :tenant_id
          AND display_name = :old_name
    """).bindparams(
        # 显式类型让不同数据库驱动按预期处理字符串参数。
        bindparam("new_name", type_=String(200)),
        bindparam("old_name", type_=String(200)),
        bindparam("tenant_id"),
    )

    # SQL 模板和值分开传递，避免拼接用户数据或旧表数据。
    result = connection.execute(statement, {
        "tenant_id": 7,
        "old_name": "未命名",
        "new_name": "待完善",
    })

    # rowcount 只作为本次语句的影响行数，不代表整批迁移已经提交。
    if result.rowcount == 0:
        raise RuntimeError("没有找到需要迁移的账号")

bindparam() 不是把值写回 SQL 字符串，而是让语句携带一个待执行时填充的参数定义。执行时的字典键必须和占位符名称一致；缺少某个键通常会在执行阶段报参数缺失错误。

需要读取数据时，让 Result 留在当前迁移步骤里

数据迁移常见的第二种情况是先查旧值，再按旧值生成新值。此时 op.execute() 不能替代连接执行，因为你需要使用返回的 Result。可以用 mappings() 让每一行按列名访问：

from alembic import op
from sqlalchemy import text


def upgrade():
    connection = op.get_bind()
    rows = connection.execute(
        text("""
            SELECT id, display_name
            FROM account
            WHERE display_name IS NULL
            ORDER BY id
        """),
    ).mappings().all()

    for row in rows:
        # 先在 Python 中形成确定的新值，再通过参数绑定写回数据库。
        new_name = f"账号-{row['id']}"
        connection.execute(
            text("""
                UPDATE account
                SET display_name = :new_name
                WHERE id = :account_id
            """),
            {"new_name": new_name, "account_id": row["id"]},
        )

这个例子故意把“查询”和“更新”放在同一个 Connection 上。这样，迁移逻辑使用的是同一个上下文和事务范围；如果中途抛出异常，是否回滚由 Alembic 的迁移配置和外层调用决定，而不是由循环里的某一次 execute() 单独决定。

对旧写法的影响：不要在迁移函数里接管事务

从应用代码迁移到 Alembic 时，最容易带过来的习惯是自己创建 Engine、开启新的事务、执行后立即提交。迁移脚本不应该这样做。op.get_bind() 的价值恰恰在于复用 Alembic 已经配置好的连接和事务环境：

如果只需要执行一次静态 SQL，继续用 op.execute() 更容易读；如果需要 Connection 的能力，再使用 op.get_bind()。不要为了“看起来更底层”而把所有迁移都改成连接代码。

online 和 offline 模式的边界要提前写清楚

op.get_bind() 依赖已经配置好的迁移上下文；官方文档说明，在 online 模式下它返回真正的数据库连接。生成离线 SQL 时，Alembic 主要把操作编译成输出文本，并不存在可以查询数据库的实时连接。

因此，依赖结果集的迁移应至少满足下面的判断：

• 只在 online 模式执行需要查询旧数据的分支。
• 需要支持 offline 输出时，把数据变换改成数据库内可编译的静态 SQL，或拆成预先生成数据的迁移步骤。
• 不要把“本地能拿到 Connection”误认为所有 Alembic 执行方式都能拿到连接。

最小的上线前检查是：确认迁移命令使用的模式、目标数据库方言和参数类型；再检查语句的影响范围是否由主键或唯一条件约束。这里的检查是迁移设计的一部分，不是用字符串拼接来绕过参数绑定。

一份可直接套用的决策清单

• 只是执行不读结果的 SQL：优先考虑 op.execute()。
• 需要参数化 DML：用 op.get_bind() 获取当前连接，配合 text() 和参数字典。
• 需要固定类型或复杂参数：用 bindparam() 明确声明类型。
• 需要查询后再更新：用同一个连接完成查询和写入，避免在脚本中重新创建 Engine。
• 不要在迁移函数里手动提交、回滚或关闭 Alembic 管理的连接。
• 需要生成 offline SQL 时，重新评估所有依赖实时结果集的代码。

常见问题

op.get_bind() 返回的是 Engine 还是 Connection？

在 Alembic online 模式下，它返回当前用于发出迁移 SQL 的 SQLAlchemy Connection。不要把它当成应用层的全局 Engine，也不要在脚本里关闭它。

可以把参数直接写进 text() 吗？

固定不变的 SQL 结构可以写进 text()，运行时值应通过命名占位符和执行参数传入；需要类型信息时再使用 bindparam()。

rowcount 等于最终提交的行数吗？

不一定。它表示当前语句返回的影响行数，具体语义还受数据库驱动和语句类型影响；它不能代替迁移事务是否成功提交的判断。

offline 模式能不能继续用 op.get_bind？

不能把它当成在线连接使用。需要读取真实数据的逻辑依赖数据库连接，应改成可生成 SQL 的方案，或明确只在 online 模式执行。

把连接获取、参数绑定和事务管理分开看，op.get_bind() 的职责就很清楚：它让迁移脚本进入 Alembic 当前的 SQLAlchemy Connection，而不是让脚本接管数据库生命周期。