使用SQLAlchemy+asyncpg+pgbouncer时遇DuplicatePreparedStatementError

错误详情 #

{
  "status": "healthy",
  "database": "disconnected",
  "details": {
    "status": "disconnected",
    "error": "(sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.DuplicatePreparedStatementError'>: prepared statement \"__asyncpg_stmt_1__\" already exists\nHINT: \nNOTE: pgbouncer with pool_mode set to \"transaction\" or\n\"statement\" does not support prepared statements properly.\nYou have two options:\n\n* if you are using pgbouncer for connection pooling to a\n single server, switch to the connection pool functionality\n provided by asyncpg, it is a much better option for this\n purpose;\n\n* if you have no option of avoiding the use of pgbouncer,\n then you can set statement_cache_size to 0 when creating\n the asyncpg connection object.\n\n[SQL: select pg_catalog.version()]"
  }
}

环境信息 #

• 数据库：PostgreSQL（Supabase）
• 连接池：pgbouncer（事务模式）
• Python：3.11
• SQLAlchemy：2.x（异步版）
• 驱动：asyncpg
• 部署环境：Heroku

当前配置 #

def get_ssl_args():
    """Get SSL arguments based on the database URL."""
    try:
        parsed_url = urlparse(settings.DATABASE_URL)
        
        if "supabase" in settings.DATABASE_URL:
            return {
                "ssl": "require",
                "server_settings": {
                    "application_name": "finance_advisor_agent",
                    "statement_timeout": "60000",
                    "idle_in_transaction_session_timeout": "60000",
                    "client_min_messages": "warning"
                },
                "statement_cache_size": 0,  # Disable prepared statements
            }
        # ... other configurations
    except Exception as e:
        logger.error(f"Error parsing database URL: {str(e)}")
        raise

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=False,
    poolclass=AsyncAdaptedQueuePool,
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=1800,
    connect_args=get_ssl_args(),
    pool_pre_ping=True,
    execution_options={
        "compiled_cache": None,
        "isolation_level": "READ COMMITTED"
    },
    pool_reset_on_return='commit'
)

已尝试的解决措施 #

• 在connect_args中设置statement_cache_size=0以禁用预准备语句
• 通过"compiled_cache": None禁用编译缓存
• 启用pool_pre_ping进行连接健康检查
• 设置连接池回收机制以避免 stale 连接

问题咨询 #

1. 当前禁用预准备语句的配置是否正确？
2. 如何彻底解决这个偶发的错误？

解答 #

1. 禁用预准备语句的配置是否正确？ #

你的statement_cache_size=0配置方向正确，但可能存在配置未正确传递的情况：

• 检查Supabase的数据库URL是否确实被识别为包含"supabase"字符串，确保get_ssl_args()返回的字典包含statement_cache_size参数。
• 确认SQLAlchemy异步引擎的connect_args参数，是否将所有配置正确转发给asyncpg驱动。

2. 彻底解决偶发错误的额外措施 #

（1）强制禁用SQLAlchemy的预准备语句支持

在创建异步引擎时添加use_prepared_statements=False参数，这是SQLAlchemy 2.x针对PostgreSQL异步驱动的专属配置，直接禁用驱动层的预准备语句：

engine = create_async_engine(
    settings.DATABASE_URL,
    # ... 其他现有配置
    use_prepared_statements=False,  # 新增：强制禁用预准备语句
)

（2）调整pgbouncer连接模式（若有权限）

若能修改Supabase的pgbouncer配置，将pool_mode从transaction改为session模式。session模式支持预准备语句，但需注意连接复用逻辑的变化——若Supabase默认不允许修改此配置，可咨询官方支持。

（3）优化连接池配置，避免连接复用冲突

• 将pool_reset_on_return设置为'hard'，确保连接归还时彻底重置状态，避免遗留的预准备语句缓存：

  pool_reset_on_return='hard'
  

• 降低max_overflow数值，减少临时连接的创建频率，避免连接池过度复用导致的语句缓存冲突。

（4）验证asyncpg连接参数的传递

打印connect_args的返回值，确认statement_cache_size=0确实被包含在内。也可尝试直接在数据库URL中添加参数（仅用于测试）：

postgresql+asyncpg://user:pass@host:port/db?statement_cache_size=0

（5）排查SQLAlchemy内部查询

错误中显示的select pg_catalog.version()是SQLAlchemy的内部健康检查语句，添加use_prepared_statements=False后可避免该语句触发预准备语句。

内容的提问来源于stack exchange，提问作者Anas Ansari