pytest异步测试出现asyncpg.TooManyConnectionsError问题求助

核心问题分析 #

你遇到的asyncpg.exceptions.TooManyConnectionsError本质是数据库连接泄漏，当参数化测试场景增多后，并发或累积的连接数超过了数据库的最大连接限制。结合你的代码流程，主要问题点包括：

• 会话管理器fixture作用域为session但未正确清理连接，导致会话/连接长期占用
• 模块级参数化fixture可能在多测试中重复创建连接且未释放
• 异步会话的关闭逻辑未覆盖所有分支，导致连接未归还连接池

具体修复方案 #

1. 强制确保会话与连接的正确清理 #

无论fixture作用域是什么，必须在测试完成后显式关闭会话并归还连接。修改主conftest.py中的会话管理器fixture，加入yield后的清理逻辑：

# 主conftest.py中的会话fixture示例
@pytest.fixture(scope="session")
async def db_session_manager():
    manager = TestConnectionManager()
    await manager.init()  # 初始化引擎和会话工厂
    yield manager
    # 显式清理所有会话与连接池
    await manager.session_factory.remove()  # 清理scoped session
    await manager.engine.dispose()  # 关闭连接池，释放所有连接

针对test_dao.py中的otb_number fixture，将会话关闭逻辑放在finally块中，避免异常场景下的连接泄漏：

@pytest.fixture
async def otb_number(db_session):
    try:
        # 你的查询逻辑
        result = await db_session.query(...)
        yield result
    finally:
        await db_session.close()  # 确保会话关闭，连接归还连接池

2. 调整fixture作用域匹配测试需求 #

• 模块级(module)的参数化fixture如果涉及数据库连接，建议改为function作用域，或者在模块测试结束后统一执行连接清理操作
• 避免session级fixture持有大量长期未释放的会话，除非确实需要跨所有测试复用且能严格管理连接生命周期

3. 控制异步测试的并发数 #

pytest-asyncio默认会并发执行多个异步测试，当参数化场景增多时，并发测试会同时从连接池获取连接，可能瞬间打满数据库连接。可以通过配置限制并发数：
在项目根目录的pytest.ini中添加：

[pytest]
asyncio_mode = auto
asyncio_tests_concurrency = 5  # 根据数据库max_connections调整，建议设为数据库最大连接数的1/2

4. 检查数据库端的最大连接限制 #

即使调大了连接池的pool_size和max_overflow，如果数据库本身的max_connections设置过小，仍会触发报错。可以通过SQL查询PostgreSQL的配置：

SHOW max_connections;

若值偏小，修改postgresql.conf中的max_connections参数并重启数据库。

5. 排查连接泄漏的具体位置 #

可以在测试中加入连接数监控代码，定位泄漏点：

import asyncpg

async def get_current_connections():
    conn = await asyncpg.connect("你的数据库连接URL")
    count = await conn.fetchval("SELECT count(*) FROM pg_stat_activity WHERE datname = '你的数据库名'")
    await conn.close()
    return count

# 在fixture的yield前后打印连接数
@pytest.fixture(scope="function")
async def db_session():
    print(f"测试前连接数: {await get_current_connections()}")
    session = create_session()
    yield session
    await session.close()
    print(f"测试后连接数: {await get_current_connections()}")

通过对比测试前后的连接数，即可定位哪个测试或fixture未正确释放连接。

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