Psycopg3随机返回None问题排查求助

核心可能原因 #

1. 多线程共享Connection实例导致游标冲突 #

Flask默认是多线程运行模式，若你的optimus_connection是全局实例（比如模块级别初始化），多个请求线程会共用同一个conn和cursor。当线程A执行查询后，线程B的查询会覆盖游标结果，导致线程A的fetchone()拿到空值或错误结果——这是生产环境偶现、本地难以复现的典型场景（本地请求量小，线程冲突概率低）。

2. 重试逻辑未覆盖空结果场景 #

当前的retry装饰器仅处理InterfaceError和OperationalError，但查询返回None时不会触发异常，因此无法自动重试。而你手动重试能拿到结果，说明空结果场景需要纳入重试范围（仅限确定存在数据的查询）。

3. 数据库连接失效但未触发异常 #

Postgres可能因idle_in_transaction_session_timeout、网络波动等原因断开连接，此时连接处于半开状态，执行SELECT可能不抛出异常，而是返回空结果。

具体修复步骤 #

1. 确保每个请求使用独立的数据库连接 #

不要全局复用Connection实例，改为在每个请求生命周期内创建独立连接：

# Flask 请求钩子示例
from flask import g

@app.before_request
def before_request():
    # 每个请求创建新的Connection
    g.db = Connection(conn_string=CONN_STRING)

@app.teardown_request
def teardown_request(exception):
    # 请求结束后关闭连接
    db = getattr(g, 'db', None)
    if db is not None:
        db.conn.close()

# 在接口中使用
@app.route("/your-api")
def your_api():
    params = {"store_link_or_domain": "dipen28"}
    row = g.db.execute(
        query=store_detail_by_link_query,
        params=params,
    ).fetchone()
    # 后续逻辑

2. 扩展重试逻辑到空结果场景 #

针对确定存在数据的查询，添加结果检查并触发重试：
首先修改retry装饰器，支持自定义异常：

def retry(fn):
    @wraps(fn)
    def wrapper(*args, **kw):
        cls = args[0]
        exec = None
        for x in range(cls._reconnectTries):
            try:
                return fn(*args, **kw)
            except (InterfaceError, OperationalError, ValueError) as e:
                if isinstance(e, ValueError):
                    logger.warning(f"查询返回空结果，开始重试（第{x+1}次）")
                else:
                    logger.warning(f"数据库连接异常: {e}, 类型: {type(e)}")
                    logger.info(f"等待{cls._reconnectIdle}秒后重连")
                    time.sleep(cls._reconnectIdle)
                    cls._connect()
                exec = e
        logger.exception(f"重试次数耗尽，退出系统: {exec}")
        import sys
        sys.exit(exec)
    return wrapper

然后在Connection类中添加带结果检查的方法：

@retry
def execute_fetchone_for_existing_data(self, **kwargs):
    result = self.execute(**kwargs)
    row = result.fetchone()
    if row is None:
        # 抛出异常触发重试
        raise ValueError("预期存在数据，但查询返回空")
    return row

调用时改用这个方法：

row = optimus_connection.execute_fetchone_for_existing_data(
    query=store_detail_by_link_query,
    params=params,
)

3. 优化连接配置，避免半开连接 #

在_connect方法中添加TCP保活和超时参数，减少连接失效概率：

def _connect(self):
    self.conn = psycopg.connect(
        self._conn_string,
        connect_timeout=10,
        keepalives=1,
        keepalives_idle=30,
        keepalives_interval=10,
        keepalives_count=5
    )
    self.conn.autocommit = True
    self.cursor = self.conn.cursor()

同时检查Postgres配置，调整idle_in_transaction_session_timeout（建议设置为5分钟左右），避免闲置连接被强制断开。

4. 增强日志排查 #

在execute方法中添加查询日志，方便定位问题：

@retry
def execute(self, **kwargs):
    if "query" in kwargs:
        kwargs["query"] = kwargs["query"].replace("\n", " ")
        kwargs["query"] = " ".join(kwargs["query"].split())
    # 记录查询语句和参数
    logger.debug(f"执行查询: {kwargs.get('query')}, 参数: {kwargs.get('params')}")
    result = self.cursor.execute(**kwargs)
    # 记录返回行数
    logger.debug(f"查询返回{self.cursor.rowcount}行")
    return result

5. 改用官方连接池（推荐） #

放弃自定义Connection类，使用psycopg官方的psycopg_pool连接池，它能自动管理连接复用、失效重连和线程安全：

from psycopg_pool import ConnectionPool

# 全局初始化连接池
pool = ConnectionPool(CONN_STRING, min_size=5, max_size=20)

# 在接口中使用
@app.route("/your-api")
def your_api():
    params = {"store_link_or_domain": "dipen28"}
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(store_detail_by_link_query, params)
            row = cur.fetchone()
    # 后续逻辑

内容的提问来源于stack exchange，提问作者Dipendra bhatt