如何覆写Python中psycopg2的connection对象close方法排查调用位置

psycopg2的psycopg2.extensions.connection是C语言实现的扩展类型，默认禁止在实例层面动态新增属性、修改实例方法，所以你尝试给实例加oldclose属性、替换close方法时会触发属性只读、属性不存在的报错。

方案1：使用官方支持的自定义连接类（最可靠） #

psycopg2原生支持通过connection_factory参数指定自定义的连接子类，你可以在子类中覆写close方法加入调试逻辑，代码示例如下：

import traceback
import psycopg2
from psycopg2.extensions import connection

# 自定义连接子类，覆写close方法
class DebugConnection(connection):
    def close(self):
        # 打印调用栈信息到控制台
        print("\n=== 检测到数据库连接关闭调用，调用栈如下 ===")
        traceback.print_stack()
        # 执行原有关闭逻辑
        super().close()

直接使用psycopg2创建连接的用法 #

创建连接时传入connection_factory参数即可：

conn = psycopg2.connect(
    dbname="your_db",
    user="your_user",
    password="your_pass",
    host="your_host",
    connection_factory=DebugConnection
)

搭配SQLAlchemy使用的配置 #

创建引擎时通过connect_args传递自定义连接类：

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg2://your_user:your_pass@your_host/your_db",
    connect_args={"connection_factory": DebugConnection}
)

方案2：全局捕获close调用（无需修改连接创建代码） #

如果不方便修改连接创建的逻辑，可以使用sys.settrace全局监控函数调用，匹配到connection.close调用时打印栈信息：

import sys
import traceback
import psycopg2.extensions

def trace_calls(frame, event, arg):
    if event == 'call':
        # 检查调用的是不是psycopg2连接的close方法
        if frame.f_code.co_name == 'close':
            self_obj = frame.f_locals.get('self', None)
            if isinstance(self_obj, psycopg2.extensions.connection):
                print("\n=== 检测到数据库连接关闭调用，调用栈如下 ===")
                traceback.print_stack(frame)
    return trace_calls

# 启动全局追踪
sys.settrace(trace_calls)

注意这个方案会轻微影响全局性能，调试完成后记得移除相关代码。

如果你用了SQLAlchemy连接池，连接被关闭常见的触发场景包括：

• 连接池回收超时/失效连接
• 代码中显式调用了connection.close()
• 事务异常回滚后框架主动关闭连接
• 多线程场景下跨线程使用连接被保护逻辑关闭

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