如何使用psycopg设置客户端侧查询/套接字执行超时？

直接设置客户端套接字超时 #

psycopg2（psycopg的旧版本）和psycopg3都支持通过操作底层套接字实现纯客户端侧的超时控制，不受数据库状态影响，即使数据库故障也能触发超时终止操作。

psycopg2的实现方式 #

建立连接后，可获取底层套接字并设置超时：

import psycopg2
import socket

conn = psycopg2.connect("dbname=test user=postgres")
# 获取连接的底层套接字
sock = conn.get_socket()
# 设置套接字超时时间（单位：秒）
sock.settimeout(10.0)

# 执行查询时，超过10秒无响应会抛出socket.timeout异常
try:
    cur = conn.cursor()
    cur.execute("SELECT pg_sleep(20);")  # 模拟无响应的查询
except socket.timeout:
    print("查询超时，已终止")
    conn.close()

注意：该超时会作用于该连接的所有操作，包括查询、结果读取等。若需针对单个查询设置不同超时，可在执行前后动态调整套接字超时，但需注意线程安全。

psycopg3的实现方式 #

psycopg3通过raw属性获取底层套接字：

import psycopg
import socket

conn = psycopg.connect("dbname=test user=postgres")
sock = conn.raw.connection.socket
sock.settimeout(10.0)

try:
    with conn.cursor() as cur:
        cur.execute("SELECT pg_sleep(20);")
except socket.timeout:
    print("查询超时")
    conn.close()

其他客户端侧超时实现方案 #

如果不想直接操作套接字，还可以采用以下两种方案：

• 线程包装查询+超时关闭连接：启动单独线程执行查询，主线程设置超时等待，超时则关闭连接终止查询：

import psycopg2
import threading

query_result = None
query_error = None

def run_query(conn):
    global query_result, query_error
    try:
        cur = conn.cursor()
        cur.execute("SELECT pg_sleep(20);")
        query_result = cur.fetchall()
    except Exception as e:
        query_error = e

conn = psycopg2.connect("dbname=test user=postgres")
query_thread = threading.Thread(target=run_query, args=(conn,))
query_thread.start()

# 等待10秒，超时则关闭连接
query_thread.join(timeout=10)
if query_thread.is_alive():
    print("查询超时，强制终止")
    conn.close()
else:
    if query_error:
        print(f"查询出错: {query_error}")
    else:
        print(f"查询结果: {query_result}")

• IO多路复用监控套接字：使用selectors模块监控套接字的可读状态，设置超时时间，超时则判定为无响应并关闭连接。

关于PEP249无等效机制的说明 #

PEP249作为Python数据库API规范，核心目标是提供最小通用接口，只覆盖各数据库驱动的共性功能。套接字超时属于驱动底层细节，不同数据库的网络交互模型差异较大，因此PEP249未将其纳入强制规范。而JDBC的setNetworkTimeout是基于Java平台统一网络模型设计的，Python生态中各数据库驱动的底层实现差异更大，这类细节被交给驱动自行实现。

内容的提问来源于stack exchange，提问作者acongo-i