Alembic 迁移脚本怎么执行带参数的 SQL

在 Alembic 迁移脚本里执行带参数的 SQL，关键不在于给 op.execute() 随手多传一个字典，而在于先判断当前迁移是“连接数据库执行”，还是“用 --sql 生成 SQL 文本”。在线迁移可以从 op.get_bind() 取得当前连接，再把 text() 语句和参数交给 SQLAlchemy；离线模式没有真实连接，普通绑定参数就没有地方传递。

官方地址：https://alembic.sqlalchemy.org/en/latest/ops.html#alembic.operations.Operations.execute

一句话记忆：简单且不需要绑定参数的迁移 SQL 可以用 op.execute()；真正要执行参数化 SQL 时，用 op.get_bind() 获取当前连接并调用 Connection.execute()。如果只是生成离线 SQL，则使用 inline_literal() 明确把值写进输出文本。

先看清 op.execute 能做什么

Alembic 的 Operations.execute() 会把给定的 SQL 放到当前迁移上下文中执行。传入普通字符串时，Alembic 会把它转成 SQLAlchemy 的 text() 构造；也可以直接传入 SQLAlchemy Core 表达式。这个接口适合迁移文件里边界清楚的简单语句，但它的返回值不是查询结果集。

更重要的是，官方文档明确区分了在线和离线语境：在 --sql 模式中，语句会写入输出流，参数化语句不适合这种模式；如果需要连接数据库并正常传递参数，应使用迁移上下文提供的 bind。也就是说，“把参数传给哪个 API”之前，要先确认迁移是否真的拥有一个可执行的数据库连接。

在线迁移用当前连接执行绑定参数

在线执行时，直接从 op.get_bind() 取得当前迁移上下文使用的连接。SQL 文本中的命名占位符用冒号开头，实际值单独放在参数字典里，避免把外部值拼接到 SQL 字符串中。下面的例子只更新符合条件的用户状态，参数名和占位符保持一一对应。

from alembic import op
from sqlalchemy import text


def upgrade() -> None:
    # 获取当前迁移上下文的连接，不另建一个绕过 Alembic 的 Engine。
    connection = op.get_bind()
    statement = text(
        "UPDATE user_account "
        "SET status = :status "
        "WHERE user_id = :user_id"
    )

    # 值与命名参数分开传递，避免把参数直接拼进 SQL 文本。
    connection.execute(
        statement,
        {"status": "active", "user_id": 42},
    )


def downgrade() -> None:
    # 回滚时使用同样的连接边界，把状态恢复为待处理。
    connection = op.get_bind()
    statement = text(
        "UPDATE user_account "
        "SET status = :status "
        "WHERE user_id = :user_id"
    )
    connection.execute(
        statement,
        {"status": "pending", "user_id": 42},
    )

这里的重点有三个：连接来自 Alembic 当前上下文，语句使用 SQLAlchemy 的 text()，参数作为单独映射传给 execute()。迁移事务由调用 Alembic 的环境负责，不要为了“确保提交”而在脚本里私自创建第二个连接或随意提交。

离线 --sql 时改用 inline_literal

如果发布流程需要先执行 alembic upgrade head --sql，当前脚本的目标是生成可审阅的 SQL 文本，而不是连接数据库。此时可以使用 op.inline_literal() 将迁移时已确定的值内联到 SQL 表达式中。它不是绑定参数的替代品，而是离线生成场景下明确的字面量表达。

from alembic import op
from sqlalchemy import table, column, String


def upgrade() -> None:
    # 声明迁移脚本需要的最小表结构，避免依赖会变化的应用模型。
    account = table(
        "user_account",
        column("status", String),
        column("user_id", String),
    )

    # 离线 --sql 没有真实连接，用 inline_literal 生成确定的 SQL 字面量。
    op.execute(
        account.update()
        .where(account.c.user_id == op.inline_literal("42"))
        .values(status=op.inline_literal("active"))
    )

这个写法适用于值已经由迁移脚本固定下来的情况，例如一次明确的数据修正。若值来自用户输入、环境变量或运行时查询，就不要为了迁就离线模式把它直接拼进 SQL；应把迁移改成在线执行，或者先设计一个能由发布系统审核的静态 SQL 产物。

四个容易混淆的边界

还有一个小坑：普通字符串传给 op.execute() 后会被当作 text() 处理，因此字符串里的 :name 可能被识别成绑定参数。如果冒号只是数据内容而不是占位符，需要按文档使用反斜杠转义。反过来，真正的参数化语句不要把占位符改成字符串格式化。

发布前怎么做最小回归检查

先用目标数据库的测试库执行一次在线升级，确认参数名、字段类型和事务边界都符合预期；如果生产流程走 --sql，再单独生成 SQL 文本，检查其中是否出现未处理的绑定占位符。两种检查对应两种运行模式，不能拿离线输出代替在线执行，也不能因为在线测试成功就假设离线脚本同样可用。

最终可以按下面的清单收口：

• 需要真实连接和绑定参数时，使用 op.get_bind()。
• 使用 text() 时，让参数字典独立于 SQL 文本。
• 离线 --sql 只生成文本，确定值用 inline_literal() 表达。
• 升级和降级使用相同的连接来源，并让 Alembic 环境管理事务。

相关问题

可以直接给 op.execute 传参数字典吗？不要把在线参数化执行建立在这个假设上。需要参数时，先用 op.get_bind() 取得连接，再调用连接的 execute()。

inline_literal 是不是更安全的绑定参数？不是。它会把值作为 SQL 字面量写入语句，主要服务于离线生成；在线动态值仍应使用绑定参数，并让值和 SQL 文本分离。