如何在Django中测试PostgreSQL statement_timeout并捕获超时错误

先说说你当前测试失败的核心原因：你直接覆盖了_CONN_TIMEOUT和_STATEMENT_TIMEOUT变量，但Django的DATABASES配置在服务启动时就已经解析完成了，后续修改这些变量并不会自动同步到OPTIONS里的实际配置，所以断言失败，超时也没触发。下面一步步解决你的两个问题：

1. 如何测试PostgreSQL的statement_timeout和connect_timeout？ #

测试statement_timeout #

要让超时配置生效，你需要直接修改DATABASES里的OPTIONS配置，并且关闭现有数据库连接（避免连接池复用旧配置的连接）。然后执行一个明确的慢查询（比如用PostgreSQL的pg_sleep函数）来触发超时。

修改后的测试用例如下：

from django.db import connection
from django.test import TestCase
from django.conf import settings

class DbTimeoutTest(TestCase):
    def test_db_statement_timeout(self):
        mock_s_timeout = 1
        # 保存原始配置，测试完恢复
        original_options = settings.DATABASES["default"]["OPTIONS"]["options"]
        
        try:
            # 直接修改DATABASES的statement_timeout配置
            settings.DATABASES["default"]["OPTIONS"]["options"] = f"-c statement_timeout={mock_s_timeout}ms"
            # 关闭现有连接，确保新连接使用新配置
            connection.close()
            
            # 验证配置是否更新成功
            self.assertEqual(
                settings.DATABASES["default"]["OPTIONS"]["options"],
                f"-c statement_timeout={mock_s_timeout}ms"
            )
            
            # 执行一个肯定会超时的慢查询
            with self.assertRaises(Exception) as ctx:
                with connection.cursor() as cursor:
                    cursor.execute("SELECT pg_sleep(2);")  # 睡眠2秒，远超1ms的超时设置
            
            # 验证错误是statement_timeout导致的
            self.assertIn("statement timeout", str(ctx.exception).lower())
        finally:
            # 恢复原始配置，避免影响其他测试用例
            settings.DATABASES["default"]["OPTIONS"]["options"] = original_options
            connection.close()

测试connect_timeout #

连接超时的测试需要模拟一个无法快速响应的数据库地址（比如不存在的IP、未监听的端口），然后修改connect_timeout配置，尝试连接并验证超时行为：

def test_db_connect_timeout(self):
        mock_conn_timeout = 1
        # 保存原始数据库配置
        original_db_config = settings.DATABASES["default"].copy()
        original_options = settings.DATABASES["default"]["OPTIONS"].copy()
        
        try:
            # 修改为一个无法快速连接的地址（根据你的环境调整，确保会超时）
            settings.DATABASES["default"]["HOST"] = "192.168.99.99"
            settings.DATABASES["default"]["PORT"] = "5433"
            settings.DATABASES["default"]["OPTIONS"]["connect_timeout"] = mock_conn_timeout
            
            # 关闭现有连接，使用新配置创建连接
            connection.close()
            
            # 尝试连接，预期触发超时
            with self.assertRaises(Exception) as ctx:
                with connection.cursor() as cursor:
                    cursor.execute("SELECT 1;")
            
            # 验证错误是连接超时导致的
            self.assertIn("connection timeout", str(ctx.exception).lower())
        finally:
            # 恢复原始配置
            settings.DATABASES["default"] = original_db_config
            settings.DATABASES["default"]["OPTIONS"] = original_options
            connection.close()

2. 如何在代码中捕获语句超时错误？ #

PostgreSQL的statement超时会抛出特定的数据库异常，你可以直接捕获对应的异常类：

方式1：捕获Django包装的OperationalError #

Django会将PostgreSQL的异常包装为django.db.utils.OperationalError，你可以判断错误信息来区分是否是语句超时：

from django.db.utils import OperationalError
from django.db import connection

def execute_slow_query():
    try:
        with connection.cursor() as cursor:
            cursor.execute("SELECT pg_sleep(10);")  # 超过你配置的3秒超时
            return cursor.fetchone()
    except OperationalError as e:
        if "statement timeout" in str(e).lower():
            print("查询超时，已终止")
            return None
        else:
            # 其他操作错误（比如权限问题、连接失败），重新抛出
            raise

方式2：捕获psycopg的具体异常（更精准） #

如果你使用的是psycopg2/psycopg3，可以直接捕获QueryCanceled异常（这是statement超时对应的具体异常类）：

import psycopg2
from django.db import connection

def execute_slow_query():
    try:
        with connection.cursor() as cursor:
            cursor.execute("SELECT pg_sleep(10);")
            return cursor.fetchone()
    except psycopg2.errors.QueryCanceled as e:
        if "statement timeout" in str(e).lower():
            print("语句超时被PostgreSQL终止")
            return None
    # 其他数据库错误仍需捕获
    except Exception as e:
        print(f"数据库操作出错：{str(e)}")
        raise

对于ORM操作（比如Book.objects.create()），同样可以在外层捕获这些异常，逻辑完全一致。

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