FastAPI应用频繁出现未关闭连接警告，求助排查原因

问题描述 #

开发的FastAPI应用频繁抛出大量警告：
ResourceWarning: unclosed connection <asyncpg.connection.Connection object at 0x>
以及
unclosed resource <TCPTransport closed=False reading=False 0x5a17c850d000>

已确认所有数据库操作均通过上下文管理器实现，示例代码如下：

同步数据库操作 #

with create_session() as session:
        session.add(db_models)
        session.commit()

异步数据库操作 #

query = select(model.source_identifier).filter(model.source_identifier == id_to_check,
                                               model.data["embedding_model"].astext==CURRENT_EMBEDDING_MODEL.value)
async with create_async_session() as session:
    row = await session.execute(query)
    row_exists = bool(row.scalar())
return row_exists

异步会话创建实现 #

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, AsyncEngine, async_sessionmaker

def _create_async_engine() -> AsyncEngine:
    hostname = os.environ.get('Postgres__Hostname')
    username = os.environ.get('Postgres__Username')
    password = os.environ.get('Postgres__Password')
    database_name = os.environ.get('Postgres__Name')
    return create_async_engine(
        f"postgresql+asyncpg://{username}:{password}@{hostname}/{database_name}"
    )

_asyncsessionmaker = async_sessionmaker()
_asyncsessionmaker.configure(bind=_create_async_engine())

def create_async_session() -> AsyncSession:
    return _asyncsessionmaker()

警告出现无规律且成批触发，并非每次请求后都会出现，需排查根源。

可能的原因及解决办法 #

1. 异常场景下会话未正确关闭 #

虽然使用了async with上下文管理器，但如果代码块内抛出未捕获的异常，会话可能无法正常完成回滚或关闭流程。可以通过显式的finally块确保会话关闭，或者完整捕获异常：

async def check_exists(id_to_check):
    session = None
    try:
        session = create_async_session()
        async with session:
            query = select(model.source_identifier).filter(...)
            row = await session.execute(query)
            return bool(row.scalar())
    finally:
        if session:
            await session.close()

2. 连接池配置不合理 #

默认的SQLAlchemy连接池可能存在未及时回收连接的情况，可通过显式配置参数优化：

def _create_async_engine() -> AsyncEngine:
    # 环境变量获取逻辑不变
    return create_async_engine(
        f"postgresql+asyncpg://{username}:{password}@{hostname}/{database_name}",
        pool_pre_ping=True,  # 获取连接前自动检测有效性
        pool_recycle=300,    # 每5分钟强制回收连接，避免数据库端主动关闭后客户端未感知
        pool_size=10,        # 连接池默认大小
        max_overflow=20      # 超出池大小的临时连接上限
    )

3. 异步会话初始化方式优化 #

当前代码先创建async_sessionmaker实例再通过configure绑定引擎，可改为直接初始化时绑定，避免潜在的上下文问题：

# 优化后的异步会话创建逻辑
_async_engine = _create_async_engine()
# 直接绑定引擎，同时关闭commit后的实例过期（按需选择）
_asyncsessionmaker = async_sessionmaker(bind=_async_engine, expire_on_commit=False)

def create_async_session() -> AsyncSession:
    return _asyncsessionmaker()

4. 用FastAPI依赖管理会话生命周期 #

使用FastAPI的Dependency机制管理会话，确保每个请求结束后无论成功失败都自动清理会话：

from fastapi import Depends

async def get_async_session():
    async with _asyncsessionmaker() as session:
        yield session

# 路由中使用依赖
@app.get("/check/{id}")
async def check_item(id: str, session: AsyncSession = Depends(get_async_session)):
    query = select(model.source_identifier).filter(...)
    row = await session.execute(query)
    return {"exists": bool(row.scalar())}

yield语法会在请求结束后自动触发上下文管理器的退出逻辑，保证会话关闭。

5. 升级asyncpg版本 #

SQLAlchemy异步层依赖asyncpg，旧版本可能存在已知的资源泄漏问题，执行以下命令升级到最新稳定版：

pip install --upgrade asyncpg

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