FastAPI(ASGI)下如何正确配置SQLAlchemy Core异步连接解决并发报错

你遇到的连接关闭报错完全是因为连接管理逻辑设计错误，核心问题有3个：

• 全局共享Request类属性导致连接串用：FastAPI的Request是全局类对象，你直接给Request.db_conn赋值相当于所有请求共享同一个连接变量，高并发下多个请求会互相覆盖连接引用，前一个请求结束触发连接关闭后，后一个请求还在调用已经被释放的连接，直接触发ResourceClosedError
• 代码拼写错误：get_async_engine里的future: bool = Ture应该是True，engine.begine()应该是engine.begin()，即使是笔误也会额外触发运行时异常
• 依赖设计不符合FastAPI规范：依赖应该直接返回/ yield当前请求独立的连接对象，而不是挂载到全局变量共享

1. 修正数据库连接管理逻辑（service/utils/db.py） #

from sqlalchemy.ext.asyncio import create_async_engine, AsyncConnection
from service.utils.config import DB_DSN

# 全局只初始化一次引擎，启动时不会提前建连接，第一次请求时才会初始化连接池
ENGINE = create_async_engine(
    str(DB_DSN), 
    echo=True, 
    future=True,
    # 高并发场景可选配置连接池参数，总和不要超过Postgres的max_connections（默认100）
    pool_size=20,  # 常驻核心连接数
    max_overflow=30,  # 峰值时额外临时创建的最大连接数
    pool_recycle=3600  # 1小时自动回收旧连接，避免数据库主动断开导致无效连接
)

async def get_async_conn() -> AsyncConnection:
    # 每个请求独立生成一个连接，请求结束自动提交/回滚并关闭连接
    async with ENGINE.begin() as conn:
        yield conn

2. 接口层通过依赖注入获取独立连接 #

不要调用全局Request的属性，直接在接口参数中注入当前请求的独立连接：

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncConnection
from service.utils.db import get_async_conn

router = APIRouter(...)

@router.get("")
async def get_items(conn: AsyncConnection = Depends(get_async_conn)):
    query = await conn.execute(...)  # 直接用注入的独立连接
    items = query.fetchall()
    return items

如果你习惯通过请求对象拿连接，也可以将连接挂载到当前请求实例的state而非全局类上，修改依赖如下：

from fastapi import Request

async def get_async_conn(request: Request):
    async with ENGINE.begin() as conn:
        request.state.db_conn = conn  # 存到当前请求的私有state，不会和其他请求冲突
        yield

接口中调用方式：

@router.get("")
async def get_items(request: Request):
    query = await request.state.db_conn.execute(...)
    items = query.fetchall()

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