如何在FastAPI+SQLAlchemy中收集单HTTP请求的所有数据库查询？

别从日志里费劲提取了，用SQLAlchemy的事件监听机制结合FastAPI的请求上下文，就能精准统计单个请求内的所有数据库查询，简单又靠谱。

核心思路 #

利用SQLAlchemy的执行事件（同步用after_execute、异步用after_execute_async），在每次数据库查询完成后，把查询语句、参数等信息绑定到当前请求的专属存储中；请求结束时统一统计查询数量，超过阈值就触发告警。

实现代码 #

1. 同步SQLAlchemy引擎示例

from fastapi import FastAPI, Request
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.event import listens_for

app = FastAPI()

# 初始化同步数据库引擎
engine = create_engine("postgresql://user:password@localhost/db_name")

# 告警逻辑：替换成你的通知方式（邮件、企业微信等）
def send_alert(request_path: str, query_count: int, queries: list):
    print(f"⚠️ 告警：请求 {request_path} 执行了 {query_count} 次查询，超过阈值！")
    print("查询详情：")
    for idx, q in enumerate(queries, 1):
        print(f"{idx}. 语句: {q['statement']} | 参数: {q['parameters']}")

# 注册SQLAlchemy查询完成事件监听
@listens_for(Engine, "after_execute")
def track_query(conn, clauseelement, multiparams, params, result, execution_options):
    # 从数据库连接中获取当前FastAPI请求对象
    request = getattr(conn, "_fastapi_request", None)
    if not request:
        return
    
    # 记录查询信息到请求的state中
    query_info = {
        "statement": str(clauseelement),
        "parameters": multiparams or params,
        "execution_time": result.execution_time if hasattr(result, "execution_time") else 0
    }
    if not hasattr(request.state, "queries"):
        request.state.queries = []
    request.state.queries.append(query_info)

# FastAPI中间件：绑定请求到数据库连接，请求结束后统计查询
@app.middleware("http")
async def bind_request_to_db(request: Request, call_next):
    with engine.connect() as conn:
        # 将当前请求对象附加到数据库连接上
        setattr(conn, "_fastapi_request", request)
        # 处理请求
        response = await call_next(request)
        # 统计并判断是否告警
        query_count = len(getattr(request.state, "queries", []))
        THRESHOLD = 10  # 自定义查询数量阈值
        if query_count > THRESHOLD:
            send_alert(request.url.path, query_count, request.state.queries)
    return response

2. 异步SQLAlchemy引擎适配

如果使用异步引擎，只需要调整事件监听和中间件逻辑：

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio.engine import AsyncEngine

# 初始化异步数据库引擎
async_engine = create_async_engine("postgresql+asyncpg://user:password@localhost/db_name")

# 异步事件监听
@listens_for(AsyncEngine, "after_execute_async")
async def track_async_query(conn, clauseelement, multiparams, params, result, execution_options):
    request = getattr(conn, "_fastapi_request", None)
    if not request:
        return
    
    query_info = {
        "statement": str(clauseelement),
        "parameters": multiparams or params,
        "execution_time": result.execution_time if hasattr(result, "execution_time") else 0
    }
    if not hasattr(request.state, "queries"):
        request.state.queries = []
    request.state.queries.append(query_info)

# 异步中间件
@app.middleware("http")
async def bind_request_to_async_db(request: Request, call_next):
    async with async_engine.connect() as conn:
        setattr(conn, "_fastapi_request", request)
        response = await call_next(request)
        query_count = len(getattr(request.state, "queries", []))
        THRESHOLD = 10
        if query_count > THRESHOLD:
            await send_alert(request.url.path, query_count, request.state.queries)  # 告警函数需改为异步
    return response

关键优势 #

• 精准度高：直接从SQLAlchemy执行链路获取查询数据，不受日志格式、日志级别限制
• 请求隔离：每个请求的查询独立统计，不会和其他请求的查询混淆
• 扩展性强：可额外记录执行时间、参数等信息，方便后续排查n+1问题

额外优化建议 #

统计到异常请求后，用SQLAlchemy的selectinload或joinedload预加载关联数据，直接解决n+1问题；同时给高频查询添加合适的索引，优化整体性能。

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