Python FastAPI 日志异常处理与链路追踪实践

脚本专家

线上接口返回 500，并不等于已经知道根因。FastAPI/Python 服务在多副本、并发、跨服务调用场景下，需要把日志、异常处理和链路追踪串起来。本文以订单接口为例，完成 request_id 上下文、JSON 结构化日志、统一错误码、下游透传与 OpenTelemetry 追踪的最小闭环。

一、先划清日志、指标、Trace 的边界

日志回答“当时具体发生了什么”，指标回答“系统整体是否健康”，链路追踪回答“这次请求经过了哪里、慢在哪里”。日志级别不是语气强弱，而是告警、筛选和存储策略的依据：DEBUG 用于调试路径和局部变量，线上通常关闭；INFO 用于服务启动、请求完成、关键业务事件；WARNING 用于可恢复异常、配置降级、资源即将耗尽；ERROR 用于当前操作失败且需要关注；CRITICAL 用于服务不可用、数据风险等紧急故障。验证码输错、库存不足等预期业务结果，通常返回明确业务错误码即可，不必都写成 ERROR。

结构化日志用固定字段代替自然语言拼接，便于 Elasticsearch、Loki、ClickHouse 等平台建立索引和聚合。例如：

2. {
3.   "timestamp": "2026-09-16T10:24:11.625+08:00",
4.   "level": "INFO",
5.   "event": "request_completed",
6.   "request_id": "e3ee2d4c8df84c1c",
7.   "method": "POST",
8.   "path": "/api/orders",
9.   "status_code": 201,
10.   "duration_ms": 38.4
11. }

复制代码

event 要稳定描述事件名，例如 request_completed、payment_timeout，变化信息放在独立字段，便于按 event=payment_timeout 聚合，而不是依赖模糊文本搜索。

request_id 是应用层请求编号，通常由入口服务生成并通过 X-Request-ID 传递，简单适合日志检索。分布式追踪中，一次端到端请求对应一个 trace_id，每次 HTTP、数据库或内部函数调用可形成 span，并拥有 span_id，下游 span 通过父子关系构成调用树。W3C Trace Context 约定用 traceparent 请求头传递。实践中可同时保留两者：业务人员用 request_id 快速查日志，运维平台依靠 trace_id 看跨服务耗时。

异常是程序内部的控制流，错误码是对外契约。不要把某个库的 IntegrityError、KeyError 直接交给 API 客户端。可按参数错误 422/400、认证授权 401/403、业务错误 409/400、依赖错误 503/502、未知系统错误 500 分类。对外错误响应建议如下：

2. {
3.   "code": "ORDER_NOT_CANCELLABLE",
4.   "message": "当前订单状态不允许取消",
5.   "request_id": "e3ee2d4c8df84c1c"
6. }

复制代码

message 可以面向用户，code 必须稳定且能被程序处理，request_id 让用户反馈问题时可以被迅速定位。

二、请求观测闭环与统一异常处理

一条 HTTP 请求的观测闭环可设计为：请求进入后，中间件读取或生成 request_id，写入 ContextVar 日志上下文，记录 request_started；路由、Service、Repository 执行业务；下游 HTTP 请求透传 X-Request-ID 与 traceparent；正常响应记录 request_completed，包含状态码和耗时；已知业务异常转换为稳定错误响应；未知异常记录完整堆栈，返回通用 500 加 request_id；最后清理 ContextVar。

中间件适合处理与业务无关的横切逻辑。把 request_id 放在 ContextVar 中，Service 层无需层层传参，日志 Filter 也能自动补充。对于 Python asyncio 任务，ContextVar 会随当前任务上下文传播；但提交到 Celery、线程池或独立进程时，仍需显式传递。统一异常处理把“记录什么”和“返回什么”分开：业务异常调用方依据 code 处理，未知异常只向服务器日志写堆栈，客户端得到不泄漏内部细节的响应，也避免每个路由重复写 try/except。

理想状态下，每条应用日志都附带 request_id、trace_id 与 span_id。告警从指标发现后，先跳到异常 trace，再由 trace 中的 ID 跳转到对应日志。即使暂未部署完整追踪平台，仅通过 request_id 也能完成第一阶段的请求级检索。

三、FastAPI 最小实现

先安装依赖：

2. pip install fastapi "uvicorn[standard]" httpx python-json-logger
3. pip install opentelemetry-api opentelemetry-sdk
4. pip install opentelemetry-instrumentation-fastapi opentelemetry-instrumentation-httpx

复制代码

1. 用 ContextVar 保存请求上下文，并输出 JSON 日志。创建 observability.py：

2. from __future__ import annotations
3. import contextvars
4. import json
5. import logging
6. import sys
7. from datetime import datetime, timezone
9. request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
10.     'request_id', default='-'
11. )
13. class ContextFilter(logging.Filter):
14.     def filter(self, record: logging.LogRecord) -> bool:
15.         record.request_id = request_id_var.get()
16.         return True
18. class JsonFormatter(logging.Formatter):
19.     def format(self, record: logging.LogRecord) -> str:
20.         payload = {
21.             'timestamp': datetime.now(timezone.utc).isoformat(),
22.             'level': record.levelname,
23.             'logger': record.name,
24.             'message': record.getMessage(),
25.             'request_id': getattr(record, 'request_id', '-'),
26.         }
27.         for key in ('event', 'method', 'path', 'status_code', 'duration_ms', 'code'):
28.             value = getattr(record, key, None)
29.             if value is not None:
30.                 payload[key] = value
31.         if record.exc_info:
32.             payload['exception'] = self.formatException(record.exc_info)
33.         return json.dumps(payload, ensure_ascii=False, default=str)
35. def configure_logging() -> None:
36.     handler = logging.StreamHandler(sys.stdout)
37.     handler.addFilter(ContextFilter())
38.     handler.setFormatter(JsonFormatter())
39.     root = logging.getLogger()
40.     root.handlers.clear()
41.     root.addHandler(handler)
42.     root.setLevel(logging.INFO)
44. logger = logging.getLogger('app')

复制代码

这里没有把用户邮箱、Token、完整请求体放入通用日志字段。日志往往会被集中采集并长期保留，默认收集越少，泄露风险越低。

2. 定义业务异常与统一响应。创建 errors.py：

2. from dataclasses import dataclass
4. @dataclass
5. class AppError(Exception):
6.     code: str
7.     message: str
8.     status_code: int = 400
9.     log_level: str = 'warning'
11. class InventoryNotEnough(AppError):
12.     def __init__(self, sku_id: str) -> None:
13.         super().__init__(
14.             code='INVENTORY_NOT_ENOUGH',
15.             message='库存不足，请调整购买数量',
16.             status_code=409,
17.         )

复制代码

3. 在 main.py 中挂中间件、异常处理器和订单接口：

2. from __future__ import annotations
3. import time
4. import uuid
5. from contextlib import asynccontextmanager
6. from fastapi import FastAPI, Request
7. from fastapi.exceptions import RequestValidationError
8. from fastapi.responses import JSONResponse
9. from pydantic import BaseModel, Field
10. from errors import AppError, InventoryNotEnough
11. from observability import configure_logging, logger, request_id_var
13. class CreateOrderRequest(BaseModel):
14.     sku_id: str = Field(min_length=1, max_length=64)
15.     count: int = Field(gt=0, le=10)
17. @asynccontextmanager
18. async def lifespan(app: FastAPI):
19.     configure_logging()
20.     logger.info('service started', extra={'event': 'service_started'})
21.     yield
22.     logger.info('service stopped', extra={'event': 'service_stopped'})
24. app = FastAPI(lifespan=lifespan)
26. @app.middleware('http')
27. async def request_logging_middleware(request: Request, call_next):
28.     request_id = request.headers.get('X-Request-ID') or uuid.uuid4().hex
29.     token = request_id_var.set(request_id)
30.     started = time.perf_counter()
31.     try:
32.         response = await call_next(request)
33.         duration_ms = round((time.perf_counter() - started) * 1000, 2)
34.         response.headers['X-Request-ID'] = request_id
35.         logger.info(
36.             'request completed',
37.             extra={
38.                 'event': 'request_completed',
39.                 'method': request.method,
40.                 'path': request.url.path,
41.                 'status_code': response.status_code,
42.                 'duration_ms': duration_ms,
43.             },
44.         )
45.         return response
46.     except Exception:
47.         logger.exception(
48.             'request crashed',
49.             extra={'event': 'request_crashed', 'method': request.method, 'path': request.url.path},
50.         )
51.         raise
52.     finally:
53.         request_id_var.reset(token)
55. @app.exception_handler(AppError)
56. async def app_error_handler(_: Request, exc: AppError):
57.     log = getattr(logger, exc.log_level, logger.warning)
58.     log('business request rejected', extra={'event': 'business_error', 'code': exc.code})
59.     return JSONResponse(
60.         status_code=exc.status_code,
61.         content={'code': exc.code, 'message': exc.message, 'request_id': request_id_var.get()},
62.     )
64. @app.exception_handler(RequestValidationError)
65. async def validation_error_handler(_: Request, exc: RequestValidationError):
66.     logger.warning('invalid request', extra={'event': 'validation_error', 'code': 'VALIDATION_ERROR'})
67.     return JSONResponse(
68.         status_code=422,
69.         content={
70.             'code': 'VALIDATION_ERROR',
71.             'message': '请求参数不符合要求',
72.             'details': exc.errors(),
73.             'request_id': request_id_var.get(),
74.         },
75.     )
77. @app.exception_handler(Exception)
78. async def unhandled_error_handler(_: Request, exc: Exception):
79.     logger.exception('unhandled server error', extra={'event': 'unhandled_error', 'code': 'INTERNAL_ERROR'})
80.     return JSONResponse(
81.         status_code=500,
82.         content={
83.             'code': 'INTERNAL_ERROR',
84.             'message': '服务暂时不可用，请稍后重试',
85.             'request_id': request_id_var.get(),
86.         },
87.     )
89. @app.post('/api/orders', status_code=201)
90. async def create_order(payload: CreateOrderRequest):
91.     if payload.sku_id == 'sold-out':
92.         raise InventoryNotEnough(payload.sku_id)
93.     logger.info('order created', extra={'event': 'order_created'})
94.     return {'order_id': uuid.uuid4().hex, 'status': 'CREATED'}

复制代码

运行服务：

2. uvicorn main:app --reload --port 8000

复制代码

请求库存不足的商品：

2. curl -i -X POST http://127.0.0.1:8000/api/orders \
3. -H "Content-Type: application/json" \
4. -H "X-Request-ID: demo-order-001" \
5. -d '{"sku_id":"sold-out","count":1}'

复制代码

响应会包含可反馈的关联标识：

2. {
3.   "code": "INVENTORY_NOT_ENOUGH",
4.   "message": "库存不足，请调整购买数量",
5.   "request_id": "demo-order-001"
6. }

复制代码

四、下游透传 request_id 与 OpenTelemetry

使用 httpx 调用库存服务时，显式传递当前 request_id：

2. import httpx
3. from observability import logger, request_id_var
5. async def reserve_inventory(sku_id: str, count: int) -> None:
6.     headers = {'X-Request-ID': request_id_var.get()}
7.     timeout = httpx.Timeout(connect=1.0, read=2.0, write=2.0, pool=1.0)
8.     try:
9.         async with httpx.AsyncClient(timeout=timeout) as client:
10.             response = await client.post(
11.                 'http://inventory.internal/reservations',
12.                 json={'sku_id': sku_id, 'count': count},
13.                 headers=headers,
14.             )
15.             response.raise_for_status()
16.     except httpx.TimeoutException as exc:
17.         logger.warning('inventory timeout', extra={'event': 'inventory_timeout', 'code': 'DEPENDENCY_TIMEOUT'})
18.         raise AppError('DEPENDENCY_TIMEOUT', '库存服务繁忙，请稍后重试', 503) from exc
19.     except httpx.HTTPStatusError as exc:
20.         logger.error('inventory rejected request', extra={'event': 'inventory_http_error'})
21.         raise AppError('INVENTORY_UNAVAILABLE', '库存服务暂不可用', 503) from exc

复制代码

不要无限重试所有下游异常。写操作重试前要确认幂等性，例如携带 Idempotency-Key，否则网络超时后重试可能造成重复扣库存。

OpenTelemetry 提供标准 trace 上下文与 span 数据。最小初始化示例如下：

2. from opentelemetry import trace
3. from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
4. from opentelemetry.sdk.resources import Resource
5. from opentelemetry.sdk.trace import TracerProvider
6. from opentelemetry.sdk.trace.export import BatchSpanProcessor
7. from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
8. from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
10. def configure_tracing(app) -> None:
11.     provider = TracerProvider(
12.         resource=Resource.create({'service.name': 'order-api', 'deployment.environment': 'production'})
13.     )
14.     provider.add_span_processor(
15.         BatchSpanProcessor(OTLPSpanExporter(endpoint='http://otel-collector:4318/v1/traces'))
16.     )
17.     trace.set_tracer_provider(provider)
18.     FastAPIInstrumentor.instrument_app(app)
19.     HTTPXClientInstrumentor().instrument()

复制代码

实际部署中通常把 trace 发往 OpenTelemetry Collector，再由 Collector 导出到 Jaeger、Tempo 或云端观测平台。自动埋点已经能覆盖 FastAPI 和 HTTPX；对“创建订单”“扣减积分”这类关键业务动作，可以再手动创建 span，并只添加低基数、无敏感信息的属性。

五、生产实践与常见坑

不要记录敏感信息。密码、访问令牌、Cookie、银行卡号、身份证号和完整地址不应写入普通应用日志。必要字段应脱敏，例如只记录手机号后四位；请求头应设置白名单，而不是将所有 header 序列化。日志平台的访问权限、保留期限和删除策略也是安全设计的一部分。

不要吞掉异常。下面写法会让程序在错误状态下继续运行，也丢失排障线索：

2. try:
3.     await save_order()
4. except Exception:
5.     pass

复制代码

如果异常确实可恢复，应记录原因、采取明确的补偿动作，并保持异常链：raise AppError(...) from exc。from exc 能保留原始异常，便于后续查看完整堆栈。

不要把堆栈返回给客户端。开发模式中 FastAPI 的调试信息很便利，生产模式却可能暴露文件路径、依赖版本、SQL 或密钥片段。对外始终返回统一错误响应；完整堆栈只进入受控日志系统。用户报障时提供 request_id，客服和研发即可精确检索。

控制日志量与字段基数。访问日志、成功请求日志通常量最大。高并发接口可对成功日志采样，例如只保留 5%；错误日志不采样或设置单独策略。不要把订单号、用户 ID、URL 全参数作为 Prometheus label，这会造成高基数指标爆炸。日志字段的高基数一般可接受，指标标签则应谨慎。

异步任务需要显式传递上下文。Celery 任务、消息队列消息、run_in_executor 线程池与当前 Web 请求并非同一执行上下文。提交任务时把 request_id、traceparent 写入任务头或消息元数据；Worker 启动后恢复这些字段，再写日志。否则 HTTP 请求和后台处理会在观测平台中断开。

建立可执行的告警规则。进阶上，可把可观测性纳入接口契约，区分业务日志、审计日志与运行日志，用 SLI、SLO 连接技术指标与服务承诺，并在多服务环境中统一约定错误码、关联头和日志字段。这样指标发现错误率升高，追踪定位慢在库存服务，日志再回答库存服务拒绝了哪个业务条件，形成完整闭环。

热心网友6

这篇整理得挺成体系，尤其是先把日志、指标、Trace 的职责边界讲清楚，再落到 request_id、trace_id/span_id、统一错误码和 FastAPI 的最小实现，读起来很有脉络。结构化日志里 event 固定描述事件名、变化信息放独立字段这点很关键，后面按 event 聚合会比模糊文本搜索稳很多。 异常处理部分也很有共鸣：业务异常和未知系统异常分开，对外只暴露稳定 code 和 request_id，不把 IntegrityError、KeyError 这类内部细节直接丢给客户端，同时服务端保留完整堆栈，这个边界划得很实用。另外提到 ContextVar 在 asyncio 任务里会传播，但提交到 Celery、线程池或独立进程时仍要显式传递，这个提醒很重要，实际项目里很容易踩坑。 代码部分好像到 JsonFormatter 那里就截住了，期待后续把中间件、统一异常处理器和 OpenTelemetry 串起来的部分补全。如果后面还能顺带讲讲日志脱敏、采样策略以及 trace 和日志平台怎么互跳，就更完整了。

热心网友7

这篇整理得很清楚，先把日志、指标、Trace 的边界划开，再落到 FastAPI 的请求观测闭环，思路很顺。业务预期结果不一定都打 ERROR 这一点很赞同，验证码错误、库存不足这类给稳定业务码更合适，否则告警容易被噪音淹没。request_id 和 trace_id 同时保留也很贴合实际，业务查日志和运维看调用树各取所需。ContextVar 在 asyncio 任务里能传播、但到 Celery、线程池或独立进程要显式传递，这个提醒很关键，不然日志链路很容易断。统一异常处理把“记录什么”和“返回什么”分开，也能避免把 IntegrityError、KeyError 这类内部细节暴露给客户端。期待后续中间件、统一异常处理器和下游透传 traceparent 的完整实现。另外想请教一下，如果暂时没上 OpenTelemetry，日志里的 trace_id 和 span_id 一般先留空，还是先用 request_id 兜底？

热心网友6

这篇总结得很清晰，尤其是先把日志、指标、Trace 的边界划出来，再落到 request_id、trace_id、span_id 的配合，思路很顺。结构化日志用固定 event 和独立字段这点很实用，event=payment_timeout 这种聚合比全文搜关键字靠谱得多。异常处理部分也说到点子上：内部异常和对外错误码分开，未知异常只往服务端日志写堆栈，客户端拿到通用 500 加 request_id，既不漏排查信息，也不暴露内部细节；验证码输错、库存不足这类预期业务结果不都打 ERROR，确实很容易被忽略。ContextVar 在 asyncio 和线程池、Celery 场景里的传播差异提醒得也很及时。后面 FastAPI 最小实现刚开头，挺期待把 JsonFormatter、日志 Filter、中间件和统一异常处理器完整串起来。