FastAPI 日志实战：用结构化日志和 Trace ID 快速定位线上问题

项目刚开始开发时，很多人习惯直接使用print()输出调试信息：

print("用户登录成功") print(user_id) print(result)

这种方式在本地开发阶段比较方便，但项目上线后，很快就会遇到问题：

• 不知道日志发生在什么时间；

• 无法区分不同用户的请求；

• 多条并发请求的日志混在一起；

• 只能搜索文本，难以统计分析；

• 异常日志缺少调用链信息；

• 可能无意中输出密码、Token 等敏感数据。

对于 Web 服务来说，日志不仅是调试工具，也是观察系统运行状态、定位故障和分析性能问题的重要依据。

本文将以 FastAPI 为例，介绍如何实现一套基础的日志体系，包括日志级别、结构化 JSON、Request ID、异常记录、耗时统计和敏感信息脱敏。

一、为什么不建议直接使用 print？

下面是一段常见代码：

@app.post("/users/login") def login(request: LoginRequest): print("收到登录请求") user = find_user(request.username) if not user: print("用户不存在") raise HTTPException( status_code=401, detail="登录失败", ) print("登录成功") return { "user_id": user.id }

当系统只有一个用户时，这些输出似乎足够使用。

但如果每秒有几百个请求，日志可能变成：

收到登录请求 收到登录请求 用户不存在 收到登录请求 登录成功 登录成功 用户不存在

这时很难判断：

• 哪几行属于同一个请求；

• 哪个用户登录失败；

• 请求来自哪个接口；

• 接口执行了多长时间；

• 具体在哪一步发生错误。

因此，生产环境需要使用标准日志模块，并为每条请求建立可追踪的上下文。

二、认识 Python 日志级别

Python 标准库提供了logging模块。

最基本的用法如下：

import logging logger = logging.getLogger(__name__) logger.debug("调试信息") logger.info("普通运行信息") logger.warning("需要注意的问题") logger.error("业务或系统错误") logger.critical("严重故障")

常见日志级别可以这样理解：

级别	使用场景
DEBUG	开发调试、变量状态、详细流程
INFO	正常业务流程、服务启动、任务完成
WARNING	可以继续运行，但需要关注
ERROR	当前操作失败，需要排查
CRITICAL	服务无法继续运行或发生严重故障

生产环境通常使用INFO级别，避免输出过多调试内容。

配置示例：

import logging logging.basicConfig( level=logging.INFO, format=( "%(asctime)s " "%(levelname)s " "%(name)s " "%(message)s" ), )

输出效果：

2026-08-04 10:20:30 INFO app.main 服务启动成功

三、为每个模块创建 Logger

不建议整个项目只使用一个固定名称的 Logger。

可以在不同模块中通过__name__创建：

import logging logger = logging.getLogger(__name__)

假设当前文件是：

app/services/user_service.py

Logger 名称通常会是：

app.services.user_service

这样可以根据模块名称判断日志来自哪里，也可以为不同模块设置不同级别。

例如，可以让数据库模块输出WARNING以上日志，而业务模块保留INFO：

logging.getLogger( "app.database" ).setLevel(logging.WARNING) logging.getLogger( "app.services" ).setLevel(logging.INFO)

四、什么是结构化日志？

传统文本日志通常是：

用户登录成功 user_id=1001 ip=127.0.0.1

结构化日志则使用固定格式，例如 JSON：

{ "timestamp": "2026-08-04T10:20:30Z", "level": "INFO", "message": "用户登录成功", "user_id": 1001, "ip": "127.0.0.1" }

JSON 日志有几个明显优势：

• 可以按照字段搜索；

• 可以统计不同接口的错误数量；

• 可以筛选指定用户或请求；

• 方便接入集中式日志平台；

• 不需要依赖复杂的文本解析规则。

可以自定义一个简单的 JSON Formatter：

import json import logging from datetime import ( datetime, timezone, ) class JsonFormatter(logging.Formatter): def format( self, record: logging.LogRecord, ) -> str: log_data = { "timestamp": datetime.now( timezone.utc ).isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), } if hasattr(record, "request_id"): log_data["request_id"] = ( record.request_id ) if hasattr(record, "user_id"): log_data["user_id"] = ( record.user_id ) if record.exc_info: log_data["exception"] = ( self.formatException( record.exc_info ) ) return json.dumps( log_data, ensure_ascii=False, )

配置输出处理器：

handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) root_logger = logging.getLogger() root_logger.setLevel(logging.INFO) root_logger.handlers.clear() root_logger.addHandler(handler)

之后日志会以 JSON 形式输出。

五、为什么需要 Request ID？

Request ID 是为每次 HTTP 请求生成的唯一编号。

假设一次请求经历以下步骤：

接收 HTTP 请求 ↓ 验证用户身份 ↓ 查询数据库 ↓ 调用外部接口 ↓ 保存处理结果 ↓ 返回响应

如果这些步骤都使用相同的 Request ID，就可以从大量日志中筛选出完整调用过程。

例如：

{ "request_id": "f937d8...", "message": "开始处理请求" }

{ "request_id": "f937d8...", "message": "数据库查询完成" }

{ "request_id": "f937d8...", "message": "外部接口调用超时" }

发生故障时，只需要搜索一个 Request ID，就能快速查看该请求的全部日志。

六、使用 ContextVar 保存 Request ID

FastAPI 是异步 Web 框架，同一个进程可能同时处理多个请求。

不能简单地使用全局变量保存 Request ID：

current_request_id = ""

如果多个请求并发执行，全局变量会相互覆盖。

Python 的ContextVar可以为不同异步上下文保存独立数据：

from contextvars import ContextVar request_id_context = ContextVar( "request_id", default="-", )

读取 Request ID：

request_id = request_id_context.get()

设置 Request ID：

token = request_id_context.set( "example-request-id" )

使用完成后恢复之前的值：

request_id_context.reset(token)

七、通过中间件生成 Request ID

FastAPI 中间件可以在每个请求进入和返回时执行。

import time from uuid import uuid4 from fastapi import ( FastAPI, Request, ) app = FastAPI() @app.middleware("http") async def request_context_middleware( request: Request, call_next, ): request_id = request.headers.get( "X-Request-ID" ) if not request_id: request_id = str(uuid4()) token = request_id_context.set( request_id ) start_time = time.perf_counter() try: response = await call_next(request) process_time_ms = ( time.perf_counter() - start_time ) * 1000 logger.info( "请求处理完成", extra={ "request_id": request_id, "method": request.method, "path": request.url.path, "status_code": ( response.status_code ), "duration_ms": round( process_time_ms, 2, ), }, ) response.headers[ "X-Request-ID" ] = request_id return response finally: request_id_context.reset(token)

中间件做了几件事情：

1. 优先读取客户端传入的X-Request-ID；

2. 如果不存在，则生成 UUID；

3. 记录请求开始时间；

4. 请求完成后计算耗时；

5. 将 Request ID 写入响应头；

6. 请求结束后清理上下文。

客户端发现接口异常时，可以把响应头中的 Request ID 提供给技术人员，方便定位日志。

八、让所有日志自动携带 Request ID

如果每次写日志都手动传递：

logger.info( "查询完成", extra={ "request_id": request_id }, )

代码会比较重复。

可以通过logging.Filter自动添加：

class RequestContextFilter( logging.Filter ): def filter( self, record: logging.LogRecord, ) -> bool: record.request_id = ( request_id_context.get() ) return True

把 Filter 添加到 Handler：

handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) handler.addFilter( RequestContextFilter() )

之后业务代码只需要正常记录日志：

logger.info("开始查询用户信息")

Formatter 输出时会自动获得当前请求的 Request ID。

九、记录更多结构化字段

为了让 JSON 日志包含接口、状态码和耗时，可以扩展 Formatter：

class JsonFormatter(logging.Formatter): extra_fields = [ "request_id", "user_id", "method", "path", "status_code", "duration_ms", "task_id", ] def format( self, record: logging.LogRecord, ) -> str: log_data = { "timestamp": datetime.now( timezone.utc ).isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), } for field in self.extra_fields: value = getattr( record, field, None, ) if value is not None: log_data[field] = value if record.exc_info: log_data["exception"] = ( self.formatException( record.exc_info ) ) return json.dumps( log_data, ensure_ascii=False, )

业务代码可以增加字段：

logger.info( "用户资料更新成功", extra={ "user_id": user_id, }, )

输出：

{ "timestamp": "2026-08-04T02:20:30Z", "level": "INFO", "logger": "app.services.user", "message": "用户资料更新成功", "request_id": "f937d8...", "user_id": 1001 }

字段名称最好在整个项目中保持统一。

不要在某些模块中使用request_id，另一些模块又使用requestId或trace_no，否则会增加查询和统计难度。

十、正确记录异常堆栈

下面的写法只能输出错误描述：

try: run_task() except Exception as error: logger.error(str(error))

它不会自动记录完整的调用堆栈，排查问题时可能无法知道异常发生在哪一行。

更推荐使用：

try: run_task() except Exception: logger.exception( "任务执行失败" )

logger.exception()会自动记录当前异常堆栈。

也可以使用：

logger.error( "任务执行失败", exc_info=True, )

不要在捕获异常后只记录日志，却继续返回成功结果：

try: save_data() except Exception: logger.exception("保存失败") return { "status": "success" }

这样客户端会收到成功响应，但实际上数据并没有保存。

如果当前层无法恢复异常，应当记录必要上下文后继续抛出，或者转换成明确的业务异常。

十一、增加全局异常处理

可以在 FastAPI 中统一处理未捕获异常：

from fastapi import Request from fastapi.responses import JSONResponse @app.exception_handler(Exception) async def global_exception_handler( request: Request, error: Exception, ): logger.exception( "未处理的服务器异常", extra={ "method": request.method, "path": request.url.path, }, ) return JSONResponse( status_code=500, content={ "detail": "服务器内部错误", "request_id": ( request_id_context.get() ), }, )

返回给客户端的信息应该简洁，不要直接返回：

• Python 异常堆栈；

• 数据库错误详情；

• 服务器文件路径；

• SQL 语句；

• 内部服务地址；

• 配置和环境变量。

详细错误应该保存在受控日志系统中，客户端只需要获得错误类型和 Request ID。

十二、记录接口耗时

接口响应慢时，首先要判断时间消耗在哪一步。

可以通过上下文管理器记录代码块耗时：

import time from contextlib import contextmanager @contextmanager def log_duration( operation_name: str, ): start_time = time.perf_counter() try: yield finally: duration_ms = ( time.perf_counter() - start_time ) * 1000 logger.info( "操作耗时", extra={ "operation": operation_name, "duration_ms": round( duration_ms, 2, ), }, )

使用方式：

with log_duration("query_user"): user = query_user_from_database( user_id )

也可以分别记录：

• 数据库查询耗时；

• Redis 查询耗时；

• 外部 API 耗时；

• 文件处理耗时；

• AI 模型调用耗时；

• 响应序列化耗时。

只有知道时间消耗在哪一步，才能进行有针对性的性能优化。

十三、日志中不要记录敏感数据

日志通常会被长期保存，并可能被开发、运维和安全人员查看。

下面的写法存在风险：

logger.info( f"用户登录：" f"username={username}, " f"password={password}" )

以下信息不应该直接写入日志：

• 用户密码；

• 完整 Access Token；

• Refresh Token；

• Cookie；

• 数据库密码；

• API Key；

• 身份证号；

• 银行卡号；

• 未脱敏的手机号；

• 私密聊天和业务内容。

可以实现简单的脱敏函数：

def mask_phone(phone: str) -> str: if len(phone) < 7: return "***" return ( phone[:3] + "****" + phone[-4:] )

使用：

logger.info( "发送验证码", extra={ "phone": mask_phone(phone), }, )

对于 Token，可以只记录前几位摘要或对应的唯一编号，不要输出完整内容。

十四、同言翻译中的日志设计

实时翻译业务通常会经过多个处理步骤：

客户端建立连接 ↓ 接收输入内容 ↓ 语音识别或文本预处理 ↓ 执行翻译 ↓ 返回翻译结果 ↓ 保存会话状态

以同言翻译为例，可以为每次会话生成session_id，为每个请求或消息生成request_id。出现结果延迟或处理失败时，就可以根据这两个字段还原执行过程。

示例日志：

logger.info( "翻译任务开始", extra={ "session_id": session_id, "source_language": "zh-CN", "target_language": "en-US", }, )

翻译完成后记录耗时，但不直接记录完整原文和译文：

logger.info( "翻译任务完成", extra={ "session_id": session_id, "duration_ms": duration_ms, "input_length": len(source_text), "output_length": len( translated_text ), }, )

对于同言翻译这类可能涉及会议、商务沟通和个人对话的应用，日志设计应遵循“只记录排查问题所需的最少信息”原则。

相比记录完整内容，可以记录：

• 会话编号；

• 请求编号；

• 输入字符数量；

• 音频时长；

• 语言方向；

• 模型或服务版本；

• 处理耗时；

• 错误类型；

• 重试次数；

• 结果状态。

如果确实需要短期保存样本用于故障分析，也应该设置严格的权限、保存期限和清理机制，避免用户内容进入普通应用日志。

十五、跨服务传递 Request ID

当系统拆分成多个服务后，一次请求可能经过：

网关 ↓ 用户服务 ↓ 业务服务 ↓ 任务服务 ↓ 外部 API

如果每个服务都重新生成 Request ID，就无法把整条调用链关联起来。

调用下游服务时，应该继续传递：

headers = { "X-Request-ID": ( request_id_context.get() ) } response = http_client.post( service_url, headers=headers, json=request_data, )

下游服务读取X-Request-ID后继续使用。

除了 Request ID，还可以单独设计 Trace ID 和 Span ID：

• Trace ID：标识完整调用链；

• Span ID：标识调用链中的某一个步骤；

• Request ID：标识一次具体 HTTP 请求。

对于服务数量不多的项目，一个统一的 Request ID 已经能够解决很多排查问题。系统进一步复杂后，可以引入完整的分布式追踪方案。

十六、后台任务如何关联请求日志？

FastAPI 创建 Celery 等异步任务后，原始 HTTP 请求已经结束，后台 Worker 无法自动获得之前的上下文。

创建任务时可以显式传递 Request ID：

task = process_document.delay( document_id=document_id, request_id=( request_id_context.get() ), )

Worker 执行任务时重新设置上下文：

@celery_app.task def process_document( document_id: int, request_id: str, ): token = request_id_context.set( request_id ) try: logger.info( "开始处理文档" ) return run_document_task( document_id ) finally: request_id_context.reset( token )

这样就可以通过相同 Request ID，把创建任务和后台执行过程关联起来。

需要注意，不要为了传递日志上下文，把完整请求头或用户敏感信息放入任务消息。

十七、是否应该记录每一次成功请求？

记录所有请求可以提供完整信息，但高并发系统可能每天产生大量日志。

日志过多会带来：

• 存储成本上升；

• 查询速度下降；

• 有效错误被大量普通日志淹没；

• 日志传输占用网络资源；

• 序列化日志消耗 CPU。

可以根据业务重要程度进行调整：

• 错误请求完整记录；

• 慢请求完整记录；

• 核心接口保留成功日志；

• 高频普通接口适当采样；

• 健康检查日志降低级别或忽略；

• 静态资源请求不进入业务日志。

例如，只重点记录超过一秒的请求：

if process_time_ms > 1000: logger.warning( "发现慢请求", extra={ "path": request.url.path, "duration_ms": ( process_time_ms ), }, )

日志策略应该兼顾问题定位能力和运行成本。

十八、日志轮转与保存期限

如果直接把日志写入本地文件而不进行轮转，文件可能不断增大，最终占满磁盘。

Python 提供了按大小轮转的处理器：

from logging.handlers import ( RotatingFileHandler, ) file_handler = RotatingFileHandler( filename="app.log", maxBytes=100 * 1024 * 1024, backupCount=10, encoding="utf-8", )

也可以按时间轮转：

from logging.handlers import ( TimedRotatingFileHandler, ) file_handler = ( TimedRotatingFileHandler( filename="app.log", when="midnight", interval=1, backupCount=30, encoding="utf-8", ) )

在 Docker 和容器编排环境中，更常见的做法是把日志输出到标准输出，由容器平台或日志采集组件统一处理。

无论采用哪种方式，都应该明确日志的保存期限，避免无限期保存无价值或敏感数据。

十九、推荐的日志字段

一个实用的 Web 服务日志可以包含：

{ "timestamp": "2026-08-04T02:20:30Z", "level": "INFO", "service": "user-api", "environment": "production", "logger": "app.services.user", "message": "用户资料更新成功", "request_id": "f937d8...", "user_id": 1001, "method": "PUT", "path": "/users/me", "status_code": 200, "duration_ms": 35.8 }

推荐字段包括：

字段	用途
timestamp	日志发生时间
level	日志级别
service	服务名称
environment	运行环境
logger	日志来源模块
message	事件描述
request_id	请求唯一编号
user_id	用户编号，避免记录敏感身份信息
path	请求路径
status_code	HTTP 状态码
duration_ms	接口处理时间
error_type	错误分类
task_id	后台任务编号

并不是每条日志都需要包含全部字段，但同一类日志应尽量保持结构一致。

二十、常见误区

误区一：日志越多越好

过量日志会增加成本，也会降低排查效率。应该记录有明确用途的信息。

误区二：发生异常时只记录错误字符串

只记录str(error)往往缺少堆栈信息，应根据情况使用logger.exception()。

误区三：记录完整请求体方便排查

请求体可能包含密码、Token 和用户内容。应该采用字段白名单，而不是默认记录全部内容。

误区四：Request ID 可以替代用户权限校验

Request ID 只用于追踪请求，不具备身份认证或权限控制能力。

误区五：只在发生故障后增加日志

缺少事前设计时，故障发生后往往无法还原现场。应该在核心链路上线前规划关键日志。

二十一、总结

一套基础的 FastAPI 日志体系，可以按照以下思路建设：

1. 使用logging替代print()；

2. 为不同模块创建独立 Logger；

3. 使用 JSON 输出结构化日志；

4. 为每个请求生成 Request ID；

5. 通过ContextVar隔离并发请求上下文；

6. 记录接口状态码和处理耗时；

7. 使用logger.exception()保存异常堆栈；

8. 对敏感字段进行脱敏；

9. 在跨服务和后台任务中传递 Request ID；

10. 设置日志采样、轮转和保存期限。

高质量日志的目标不是把所有数据都记录下来，而是在系统出现问题时，能够快速回答以下几个问题：

• 哪个请求发生了问题？

• 问题发生在哪个服务？

• 具体失败在哪一个步骤？

• 影响了哪些用户或任务？

• 接口在哪个环节消耗了时间？

• 系统是否能够恢复？

能够回答这些问题的日志，才是真正有价值的生产日志。