🐍 如何在 FastAPI 中优雅地记录数据库查询日志？

让每一行 SQL 都“看得见”，让性能瓶颈无处隐藏。

在开发高性能 Web 应用时，数据库查询往往是性能瓶颈的第一嫌疑人。但在 FastAPI 这类现代异步框架中，如何在不破坏异步上下文、不影响性能的前提下，干净、可追踪、可过滤地记录每一次 SQL 执行？今天我们就来拆解一个实用的中间件实现。

为什么需要记录数据库查询日志？

调试复杂查询：快速定位慢查询或错误语句。
性能分析：结合 trace ID，将 SQL 与具体 API 请求关联。
安全审计：记录敏感操作（如 DELETE、UPDATE）。
避免 N+1 问题：直观发现重复或冗余查询。

但盲目打印所有 SQL 会带来两个问题：

大量系统查询干扰日志（如 SELECT current_schema()）；
异步上下文丢失，无法与请求日志关联。

接下来，我们通过一个自定义中间件，优雅解决这些问题。

核心思路：监听 SQLAlchemy 事件 + 异步日志

我们基于 SQLAlchemy 的事件系统，在每次执行 SQL 前触发钩子，同时利用 FastAPI 的请求上下文，将 SQL 日志与当前请求绑定。

第一步：监听数据库执行事件

from sqlalchemy import event
from sqlalchemy.engine import Engine

@event.listens_for(Engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    if not self._is_system_query(statement):
        formatted_sql = self._format_sql_with_params(statement, parameters)
        clean_sql = self._remove_newline_escapes(formatted_sql)
        asyncio.create_task(self._log_database_query(clean_sql))

✅ 使用 before_cursor_execute 事件，在 SQL 发送到数据库前拦截。
✅ 通过 asyncio.create_task() 异步记录，避免阻塞主流程。

第二步：过滤系统查询

PostgreSQL 等数据库在连接时会自动执行大量元数据查询。我们通过正则匹配过滤掉它们：

def _is_system_query(self, statement: str) -> bool:
    system_patterns = [
        r'^\s*select pg_catalog',
        r'^\s*select current_schema',
        r'^\s*show standard_conforming_strings'
    ]
    stripped = statement.strip()
    if stripped in ['BEGIN (implicit)', 'COMMIT']:
        return True
    return any(re.match(p, statement, re.IGNORECASE) for p in system_patterns)

📌 这能减少 80% 以上的无用日志，让日志清爽可读。

第三步：安全拼接 SQL 与参数

直接打印 statement 和 parameters 是分离的。我们需要将它们合并成可读的完整 SQL：

def _format_sql_with_params(self, statement: str, parameters: Any) -> str:
    if isinstance(parameters, (list, tuple)):
        for i, param in enumerate(parameters):
            placeholder = f'${i+1}'
            if isinstance(param, str):
                statement = statement.replace(placeholder, f"'{param}'")
            else:
                # 同时处理 PostgreSQL 的类型转换，如 $1::INTEGER
                statement = re.sub(f'\\${i+1}::\\w+', str(param), statement)
                if placeholder in statement:
                    statement = statement.replace(placeholder, str(param))
    return statement

⚠️ 注意：这只是用于日志展示，绝不能用于实际执行！避免 SQL 注入误解。

第四步：与请求上下文绑定（Trace ID）

关键来了！如何让这条 SQL 日志和当前 API 请求关联？

我们借助 FastAPI 的 request.state 存储 trace ID，并在记录时复用统一的日志通道：

async def _log_database_query(self, sql_statement: str):
    try:
        req = get_current_request()  # 从上下文获取当前请求
        trace_id = getattr(req.state, 'traceid', None)
        if trace_id:
            await _log_request_event('database_query', {'sql': sql_statement})
        else:
            logger.info(f"[DB Query] {sql_statement}")
    except LookupError:
        logger.info(f"[DB Query] {sql_statement}")

🌟 这样，你在日志中就能看到：

1
2
3

[trace=abc123] [event=request_start] ...
[trace=abc123] [event=database_query] {"sql": "SELECT * FROM users WHERE id = '1001'"}
[trace=abc123] [event=request_end] ...