如何在psycopg3的Alembic迁移中执行事务外的语句？

问题场景 #

尝试通过Alembic的autocommit_block执行PostgreSQL的REINDEX INDEX CONCURRENTLY（该命令无法在事务内运行），代码如下：

with op.get_context().autocommit_block():
    op.execute("REINDEX INDEX CONCURRENTLY idx_something")

但在Psycopg3环境下触发AssertionError，报错详情：

@contextmanager
def autocommit_block(self) -> Iterator[None]:
    # ...
    _in_connection_transaction = self._in_connection_transaction()

    if self.impl.transactional_ddl and self.as_sql:
        self.impl.emit_commit()

    elif _in_connection_transaction:
>       assert self._transaction is not None
E       AssertionError

.venv/lib/python3.10/site-packages/alembic/runtime/migration.py:330: AssertionError

环境版本 #

• Psycopg 3.2.9
• Alembic 1.16.4
• SQLAlchemy 2.0.42

问题原因 #

这是Alembic与Psycopg3的兼容性问题：Alembic的autocommit_block逻辑检查到连接处于事务中，但无法找到对应的_transaction对象引用，触发断言失败。

可行解决方案 #

绕过Alembic的autocommit_block，手动控制连接的自动提交状态，直接操作原始连接：

方案一：使用SQLAlchemy连接操作 #

from sqlalchemy import text

# 获取绑定的数据库连接
conn = op.get_bind().connect()
try:
    # 切换到自动提交模式
    conn.execution_options(isolation_level="AUTOCOMMIT")
    # 执行REINDEX命令
    conn.execute(text("REINDEX INDEX CONCURRENTLY idx_something"))
finally:
    # 恢复默认隔离级别，避免影响后续迁移步骤
    conn.execution_options(isolation_level="READ COMMITTED")
    conn.close()

方案二：直接使用Psycopg3原生连接 #

# 获取SQLAlchemy连接，再拿到Psycopg3原生连接
sa_conn = op.get_bind()
pg_conn = sa_conn.connection
try:
    # 开启自动提交
    pg_conn.autocommit = True
    with pg_conn.cursor() as cur:
        cur.execute("REINDEX INDEX CONCURRENTLY idx_something")
finally:
    # 恢复自动提交状态
    pg_conn.autocommit = False

说明 #

两种方案都直接避开了Alembic的事务管理逻辑，确保REINDEX CONCURRENTLY在自动提交模式下执行，既满足PostgreSQL的命令要求，又不会触发Alembic内部的断言错误。

内容的提问来源于stack exchange，提问作者Joel Shellman