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_code401, 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( levellogging.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.pyLogger 名称通常会是app.services.user_service这样可以根据模块名称判断日志来自哪里也可以为不同模块设置不同级别。例如可以让数据库模块输出WARNING以上日志而业务模块保留INFOlogging.getLogger( app.database ).setLevel(logging.WARNING) logging.getLogger( app.services ).setLevel(logging.INFO)四、什么是结构化日志传统文本日志通常是用户登录成功 user_id1001 ip127.0.0.1结构化日志则使用固定格式例如 JSON{ timestamp: 2026-08-04T10:20:30Z, level: INFO, message: 用户登录成功, user_id: 1001, ip: 127.0.0.1 }JSON 日志有几个明显优势可以按照字段搜索可以统计不同接口的错误数量可以筛选指定用户或请求方便接入集中式日志平台不需要依赖复杂的文本解析规则。可以自定义一个简单的 JSON Formatterimport 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_asciiFalse, )配置输出处理器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 IDRequest ID 是为每次 HTTP 请求生成的唯一编号。假设一次请求经历以下步骤接收 HTTP 请求 ↓ 验证用户身份 ↓ 查询数据库 ↓ 调用外部接口 ↓ 保存处理结果 ↓ 返回响应如果这些步骤都使用相同的 Request ID就可以从大量日志中筛选出完整调用过程。例如{ request_id: f937d8..., message: 开始处理请求 }{ request_id: f937d8..., message: 数据库查询完成 }{ request_id: f937d8..., message: 外部接口调用超时 }发生故障时只需要搜索一个 Request ID就能快速查看该请求的全部日志。六、使用 ContextVar 保存 Request IDFastAPI 是异步 Web 框架同一个进程可能同时处理多个请求。不能简单地使用全局变量保存 Request IDcurrent_request_id 如果多个请求并发执行全局变量会相互覆盖。Python 的ContextVar可以为不同异步上下文保存独立数据from contextvars import ContextVar request_id_context ContextVar( request_id, default-, )读取 Request IDrequest_id request_id_context.get()设置 Request IDtoken request_id_context.set( example-request-id )使用完成后恢复之前的值request_id_context.reset(token)七、通过中间件生成 Request IDFastAPI 中间件可以在每个请求进入和返回时执行。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)中间件做了几件事情优先读取客户端传入的X-Request-ID如果不存在则生成 UUID记录请求开始时间请求完成后计算耗时将 Request ID 写入响应头请求结束后清理上下文。客户端发现接口异常时可以把响应头中的 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 添加到 Handlerhandler logging.StreamHandler() handler.setFormatter(JsonFormatter()) handler.addFilter( RequestContextFilter() )之后业务代码只需要正常记录日志logger.info(开始查询用户信息)Formatter 输出时会自动获得当前请求的 Request ID。九、记录更多结构化字段为了让 JSON 日志包含接口、状态码和耗时可以扩展 Formatterclass 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_asciiFalse, )业务代码可以增加字段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_infoTrue, )不要在捕获异常后只记录日志却继续返回成功结果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_code500, 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用户登录 fusername{username}, fpassword{password} )以下信息不应该直接写入日志用户密码完整 Access TokenRefresh TokenCookie数据库密码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, headersheaders, jsonrequest_data, )下游服务读取X-Request-ID后继续使用。除了 Request ID还可以单独设计 Trace ID 和 Span IDTrace ID标识完整调用链Span ID标识调用链中的某一个步骤Request ID标识一次具体 HTTP 请求。对于服务数量不多的项目一个统一的 Request ID 已经能够解决很多排查问题。系统进一步复杂后可以引入完整的分布式追踪方案。十六、后台任务如何关联请求日志FastAPI 创建 Celery 等异步任务后原始 HTTP 请求已经结束后台 Worker 无法自动获得之前的上下文。创建任务时可以显式传递 Request IDtask process_document.delay( document_iddocument_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( filenameapp.log, maxBytes100 * 1024 * 1024, backupCount10, encodingutf-8, )也可以按时间轮转from logging.handlers import ( TimedRotatingFileHandler, ) file_handler ( TimedRotatingFileHandler( filenameapp.log, whenmidnight, interval1, backupCount30, encodingutf-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_codeHTTP 状态码duration_ms接口处理时间error_type错误分类task_id后台任务编号并不是每条日志都需要包含全部字段但同一类日志应尽量保持结构一致。二十、常见误区误区一日志越多越好过量日志会增加成本也会降低排查效率。应该记录有明确用途的信息。误区二发生异常时只记录错误字符串只记录str(error)往往缺少堆栈信息应根据情况使用logger.exception()。误区三记录完整请求体方便排查请求体可能包含密码、Token 和用户内容。应该采用字段白名单而不是默认记录全部内容。误区四Request ID 可以替代用户权限校验Request ID 只用于追踪请求不具备身份认证或权限控制能力。误区五只在发生故障后增加日志缺少事前设计时故障发生后往往无法还原现场。应该在核心链路上线前规划关键日志。二十一、总结一套基础的 FastAPI 日志体系可以按照以下思路建设使用logging替代print()为不同模块创建独立 Logger使用 JSON 输出结构化日志为每个请求生成 Request ID通过ContextVar隔离并发请求上下文记录接口状态码和处理耗时使用logger.exception()保存异常堆栈对敏感字段进行脱敏在跨服务和后台任务中传递 Request ID设置日志采样、轮转和保存期限。高质量日志的目标不是把所有数据都记录下来而是在系统出现问题时能够快速回答以下几个问题哪个请求发生了问题问题发生在哪个服务具体失败在哪一个步骤影响了哪些用户或任务接口在哪个环节消耗了时间系统是否能够恢复能够回答这些问题的日志才是真正有价值的生产日志。