[python]Flask - Tracking ID的设计

前言

在实际业务中，根据 tracking_id 追溯一条请求的完整处理路径是比较常见的需求。借助 Flask 自带的全局对象 g 以及钩子函数，可以很容易地为每条请求添加 tracking_id，并在日志中自动记录。
主要内容：

• 如何为每条请求添加 tracking_id
  
• 如何为日志自动添加 tracking_id 记录
  
• 如何自定义响应类，实现统一的响应格式，并在响应头中添加 tracking_id
  
• 视图函数单元测试示例
  
• Gunicorn 配置
  

项目结构

虽然内容看起来很多，但 tracking_id 的实现其实很简单。本文按照生产项目的规范组织了代码，添加了 Gunicorn 配置和单元测试代码，以及规范了日志格式和 JSON 响应格式。

1. ├── apis
2. │   ├── common
3. │   │   ├── common.py
4. │   │   └── __init__.py
5. │   └── __init__.py
6. ├── gunicorn.conf.py
7. ├── handles
8. │   └── user.py
9. ├── logs
10. │   ├── access.log
11. │   └── error.log
12. ├── main.py
13. ├── middlewares
14. │   ├── __init__.py
15. │   └── tracking_id.py
16. ├── pkgs
17. │   └── log
18. │       ├── app_log.py
19. │       └── __init__.py
20. ├── pyproject.toml
21. ├── pytest.ini
22. ├── README.md
23. ├── responses
24. │   ├── __init__.py
25. │   └── json_response.py
26. ├── tests
27. │   └── apis
28. │       └── test_common.py
29. ├── tmp
30. │   └── gunicorn.pid
31. └── uv.lock

复制代码

安装依赖

1. uv add flask
2. uv add gunicorn gevent  # 生产环境部署一般依赖这两个
3. uv add --dev pytest           # 测试库

复制代码

实现添加 tracking_id 的中间件

代码文件：middlewares/tracking_id.py

1. from uuid import uuid4
3. from flask import Flask, Response, g, request
6. def tracking_id_middleware(app: Flask):
7.     """
8.     跟踪 ID 中间件
9.     为每个请求生成或获取跟踪 ID，用于追踪请求链路
10.     """
11.    
12.     @app.before_request
13.     def tracking_id_before_request():
14.         """
15.         请求前处理函数
16.         检查请求头中是否包含 X-Tracking-ID，如果没有则生成一个新的 UUID 作为跟踪 ID
17.         并将其存储到 Flask 的全局对象 g 中，供后续处理使用
18.         """
19.         # 从请求头中获取 X-Tracking-ID
20.         tracking_id = request.headers.get("X-Tracking-ID")
21.         if not tracking_id:
22.             # 如果请求头中没有 X-Tracking-ID，则生成一个新的 UUID
23.             tracking_id = str(uuid4())
24.         # 将跟踪 ID 存储到 Flask 的全局对象 g 中，供后续处理使用
25.         g.tracking_id = tracking_id
27.     @app.after_request
28.     def tracking_id_after_request(response: Response):
29.         """
30.         请求后处理函数
31.         将跟踪 ID 添加到响应头中，以便客户端知道本次请求的跟踪 ID
32.         """
33.         # 检查响应头中是否已经有 X-Tracking-ID
34.         tracking_id = response.headers.get("X-Tracking-ID", "")
35.         if not tracking_id:
36.             # 如果响应头中没有 X-Tracking-ID，则从全局对象 g 中获取
37.             tracking_id = g.get("tracking_id", "")
38.             # 将跟踪 ID 添加到响应头中
39.             response.headers["X-Tracking-ID"] = tracking_id
40.         return response
42.     # 返回应用实例
43.     return app

复制代码

代码文件 middlewares/__init__.py，方便其他模块导入

1. from .tracking_id import tracking_id_middleware
3. __all__ = [
4.     "tracking_id_middleware",
5. ]

复制代码

日志模块 - 自动记录 tracking_id

实现一个简单的输出到控制台的日志模块，日志格式为 JSON，自动添加 tracking_id 到日志中，避免手动在 logger.info() 这类方法中传入 tracking_id。
代码文件 pkgs/log/app_log.py

1. import json
2. import logging
3. import sys
5. from flask import g
8. class JSONFormatter(logging.Formatter):
9.     """日志格式化器，输出 JSON 格式的日志。"""
11.     def format(self, record: logging.LogRecord) -> str:
12.         log_record = {
13.             "@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
14.             "level": record.levelname,
15.             "name": record.name,
16.             # "processName": record.processName,  # 如需记录进程名可取消注释
17.             "tracking_id": getattr(record, "tracking_id", None),
18.             "loc": "%s:%d" % (record.filename, record.lineno),
19.             "func": record.funcName,
20.             "message": record.getMessage(),
21.         }
23.         return json.dumps(log_record, ensure_ascii=False, default=str)
26. class TrackingIDFilter(logging.Filter):
27.     """日志过滤器，为日志记录添加 tracking_id。"""
29.     def filter(self, record):
30.         record.tracking_id = g.get("tracking_id", None)
31.         return True
34. def _setup_console_handler(level: int) -> logging.StreamHandler:
35.     """设置控制台日志处理器。
37.     Args:
38.         level (int): 日志级别。
39.     """
40.     handler = logging.StreamHandler(sys.stdout)
41.     handler.setLevel(level)
42.     handler.setFormatter(JSONFormatter())
43.     return handler
46. def setup_app_logger(level: int = logging.INFO, name: str = "app") -> logging.Logger:
47.     logger = logging.getLogger(name)
49.     if logger.hasHandlers():
50.         return logger
52.     logger.setLevel(level)
53.     logger.propagate = False
55.     logger.addHandler(_setup_console_handler(level))
56.     logger.addFilter(TrackingIDFilter())
58.     return logger

复制代码

在 pkgs/log/__init__.py 中初始化 logger，实现单例调用。

1. from .app_log import setup_app_logger
3. logger = setup_app_logger()
5. __all__ = ["logger"]

复制代码

自定义响应类

规范 JSON 类型的响应格式，并在响应头中添加 X-Tracking-ID 和 X-DateTime。
代码文件 responses/json_response.py

1. import json
2. from datetime import datetime
3. from http import HTTPStatus
4. from typing import Any
6. from flask import Response, g, request
9. class JsonResponse(Response):
10.     def __init__(
11.         self,
12.         data: Any = None,
13.         code: HTTPStatus = HTTPStatus.OK,
14.         msg: str = "this is a json response",
15.     ):
16.         x_tracking_id = g.get("tracking_id", "")
17.         x_datetime = datetime.now().astimezone().isoformat(timespec="seconds")
18.         resp_headers = {
19.             "Content-Type": "application/json",
20.             "X-Tracking-ID": x_tracking_id,
21.             "X-DateTime": x_datetime,
22.         }
23.         try:
24.             resp = json.dumps(
25.                 {
26.                     "code": code.value,
27.                     "msg": msg,
28.                     "data": data,
29.                 },
30.                 ensure_ascii=False,
31.                 default=str,
32.             )
33.         except Exception as e:
34.             resp = json.dumps(
35.                 {
36.                     "code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
37.                     "msg": f"Response serialization error: {str(e)}",
38.                     "data": None,
39.                 }
40.             )
41.         super().__init__(response=resp, status=code.value, headers=resp_headers)
44. class Success(JsonResponse):
45.     def __init__(self, data: Any = None, msg: str = ""):
46.         if not msg:
47.             msg = f"{request.method} {request.path} success"
48.         super().__init__(data=data, code=HTTPStatus.OK, msg=msg)
51. class Fail(JsonResponse):
52.     def __init__(self, msg: str = "", data: Any = None):
53.         if not msg:
54.             msg = f"{request.method} {request.path} failed"
55.         super().__init__(data=data, code=HTTPStatus.INTERNAL_SERVER_ERROR, msg=msg)
58. class ArgumentNotFound(JsonResponse):
59.     def __init__(self, msg: str = "", data: Any = None):
60.         if not msg:
61.             msg = f"{request.method} {request.path} argument not found"
62.         super().__init__(data=data, code=HTTPStatus.BAD_REQUEST, msg=msg)
65. class ArgumentInvalid(JsonResponse):
66.     def __init__(self, msg: str = "", data: Any = None):
67.         if not msg:
68.             msg = f"{request.method} {request.path} argument invalid"
69.         super().__init__(data=data, code=HTTPStatus.BAD_REQUEST, msg=msg)
72. class AuthFailed(JsonResponse):
73.     """HTTP 状态码: 401"""
75.     def __init__(self, msg: str = "", data: Any = None):
76.         if not msg:
77.             msg = f"{request.method} {request.path} auth failed"
78.         super().__init__(data=data, code=HTTPStatus.UNAUTHORIZED, msg=msg)
81. class ResourceConflict(JsonResponse):
82.     """HTTP 状态码: 409"""
84.     def __init__(self, msg: str = "", data: Any = None):
85.         if not msg:
86.             msg = f"{request.method} {request.path} resource conflict"
87.         super().__init__(data=data, code=HTTPStatus.CONFLICT, msg=msg)
90. class ResourceNotFound(JsonResponse):
91.     """HTTP 状态码: 404"""
93.     def __init__(self, msg: str = "", data: Any = None):
94.         if not msg:
95.             msg = f"{request.method} {request.path} resource not found"
96.         super().__init__(data=data, code=HTTPStatus.NOT_FOUND, msg=msg)
99. class ResourceForbidden(JsonResponse):
100.     """HTTP 状态码: 403"""
102.     def __init__(self, msg: str = "", data: Any = None):
103.         if not msg:
104.             msg = f"{request.method} {request.path} resource forbidden"
105.         super().__init__(data=data, code=HTTPStatus.FORBIDDEN, msg=msg)

复制代码

代码文件 responses/__init__.py，方便其他模块调用。

1. from .json_response import (
2.     ArgumentInvalid,
3.     ArgumentNotFound,
4.     AuthFailed,
5.     Fail,
6.     JsonResponse,
7.     ResourceConflict,
8.     ResourceForbidden,
9.     ResourceNotFound,
10.     Success,
11. )
13. __all__ = [
14.     "JsonResponse",
15.     "Success",
16.     "Fail",
17.     "ArgumentNotFound",
18.     "ArgumentInvalid",
19.     "AuthFailed",
20.     "ResourceConflict",
21.     "ResourceNotFound",
22.     "ResourceForbidden",
23. ]

复制代码

编写视图函数

代码文件 apis/common/common.py。以下定义了 5 个路由，主要用于测试响应类是否正常返回 JSON 格式。

1. from datetime import datetime
3. from flask import Blueprint
5. from handles import user as user_handle
6. from pkgs.log import logger
7. from responses import Success
9. route = Blueprint("common_apis", __name__, url_prefix="/api")
12. @route.get("/health")
13. def health_check():
14.     # print(g.get("tracking_id", "no-tracking-id"))
15.     logger.info("Health check")
16.     return Success(data="OK")
19. @route.get("/users")
20. def get_users():
21.     users = user_handle.get_users()
22.     return Success(data=users)
25. @route.get("/names")
26. def get_names():
27.     names = ["Alice", "Bob", "Charlie"]
28.     return Success(data=names)
31. @route.get("/item")
32. def get_item():
33.     item = {"id": 101, "name": "Sample Item", "price": 29.99, "now": datetime.now()}
34.     return Success(data=item)
37. @route.get("/error")
38. def get_error():
39.     raise Exception("This is a test exception")

复制代码

GET /api/users 调用了 handles/ 中的代码，模拟查询数据库。handles/user.py 中的代码如下：

1. import time
2. from typing import Any, Dict, List
5. def get_users() -> List[Dict[str, Any]]:
6.     # 模拟查询用户数据
7.     time.sleep(0.1)  # 模拟延迟
8.     users = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
9.     return users

复制代码

代码文件 apis/common/__init__.py 中导入各个蓝图并统一暴露。由于示例代码只定义了一个蓝图，所以这里写得很简单。如果有多个蓝图，可以把蓝图都添加到一个列表中，在 Flask 应用中一次性遍历注册。

1. from .common import route
2. # from .common import route as common_route
4. # routes = [
5. #     common_route,
6. # ]
8. __all__ = ["route"]

复制代码

代码文件 apis/__init__.py 中提供 Flask 应用的工厂函数。

1. import traceback
3. from flask import Flask
5. from apis.common import route as common_route
6. from middlewares import tracking_id_middleware
7. from responses import Fail, ResourceNotFound
8. from pkgs.log import logger
12. # 错误处理器
13. def error_handler_notfound(error):
14.     return ResourceNotFound()
17. def error_handler_generic(error):
18.     logger.error(traceback.format_exc())
19.     return Fail(data=str(error))
23. def create_app() -> Flask:
24.     app = Flask(__name__)
26.     # 注册中间件
27.     app = tracking_id_middleware(app)
29.     # 注册错误处理器
30.     app.errorhandler(Exception)(error_handler_generic)
31.     app.errorhandler(404)(error_handler_notfound)
33.     # 注册蓝图
34.     app.register_blueprint(common_route)
36.     return app
38. __all__ = [
39.     "create_app",
40. ]

复制代码

入口代码文件 main.py

1. from apis import create_app
3. app = create_app()
5. if __name__ == "__main__":
6.     app.run(host="127.0.0.1", port=8000, debug=False)

复制代码

简单运行测试

• 启动应用
  

1. # 方式1, 直接启动, 用于简单测试
2. python main.py
4. # 方式2, 使用 gunicorn, 这是生产环境启动方式. 配置文件默认路径即 ./gunicorn.conf.py
5. gunicorn main:app

复制代码

• curl 请求 /api/health。可以看到响应头中已经有了 X-Tracking-ID 和 X-DateTime
  

1. $ curl -v http://127.0.0.1:8000/api/health
2. *   Trying 127.0.0.1:8000...
3. * Connected to 127.0.0.1 (127.0.0.1) port 8000
4. * using HTTP/1.x
5. > GET /api/health HTTP/1.1
6. > Host: 127.0.0.1:8000
7. > User-Agent: curl/8.14.1
8. > Accept: */*
9. >
10. * Request completely sent off
11. < HTTP/1.1 200 OK
12. < Server: gunicorn
13. < Date: Sat, 17 Jan 2026 08:41:07 GMT
14. < Connection: keep-alive
15. < Content-Type: application/json
16. < X-Tracking-ID: 1f0adb8d-9bee-49d4-873f-31aa1437da60
17. < X-DateTime: 2026-01-17T16:41:07+08:00
18. < Content-Length: 61
19. <
20. * Connection #0 to host 127.0.0.1 left intact
21. {"code": 200, "msg": "GET /api/health success", "data": "OK"}

复制代码

• curl 请求 /api/users。手动指定请求头中的 X-Tracking-ID，响应时也会保持相同的 ID。
  

1. $ curl -v http://127.0.0.1:8000/api/users -H 'X-Tracking-ID:123456'
2. *   Trying 127.0.0.1:8000...
3. * Connected to 127.0.0.1 (127.0.0.1) port 8000
4. * using HTTP/1.x
5. > GET /api/users HTTP/1.1
6. > Host: 127.0.0.1:8000
7. > User-Agent: curl/8.14.1
8. > Accept: */*
9. > X-Tracking-ID:123456
10. >
11. * Request completely sent off
12. < HTTP/1.1 200 OK
13. < Server: gunicorn
14. < Date: Sat, 17 Jan 2026 08:44:37 GMT
15. < Connection: keep-alive
16. < Content-Type: application/json
17. < X-Tracking-ID: 123456
18. < X-DateTime: 2026-01-17T16:44:37+08:00
19. < Content-Length: 110
20. <
21. * Connection #0 to host 127.0.0.1 left intact
22. {"code": 200, "msg": "GET /api/users success", "data": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}

复制代码

编写单元测试

使用 pytest 进行单元测试，这里只是一个简单的示例
配置 pytest

配置文件 pytest.ini

1. [pytest]
2. testpaths = "tests"
3. pythonpath = "."

复制代码

测试代码

代码文件 tests/apis/test_common.py

1. from typing import Generator
2. from unittest.mock import MagicMock, patch
4. import pytest
5. from flask import Flask
6. from flask.testing import FlaskClient
8. from apis.common import route as common_route
11. @pytest.fixture
12. def app() -> Generator[Flask, None, None]:
13.     app = Flask(__name__)
14.     app.config.update(
15.         {
16.             "TESTING": True,
17.             "DEBUG": False,
18.         }
19.     )
20.     app.register_blueprint(common_route)
21.     yield app
24. @pytest.fixture
25. def client(app: Flask) -> FlaskClient:
26.     return app.test_client()
29. class TestGetHealth:
30.     def test_get_health_success(self, client: FlaskClient) -> None:
31.         resp = client.get("/api/health")
32.         assert resp.status_code == 200
34.         resp_headers = resp.headers
35.         assert resp_headers.get("Content-Type") == "application/json"
36.         assert "X-Tracking-ID" in resp_headers
37.         assert "X-DateTime" in resp_headers
39.         resp_body = resp.json
40.         assert resp_body == {
41.             "code": 200,
42.             "msg": "GET /api/health success",
43.             "data": "OK",
44.         }
47. class TestGetUsers:
48.     @patch("apis.common.common.user_handle.get_users")
49.     def test_get_users(self, mock_get_users: MagicMock, client: FlaskClient) -> None:
50.         # mock user.get_users() 的返回值
51.         mock_get_users.return_value = [
52.             {"id": 1, "name": "Alice123"},
53.             {"id": 2, "name": "Bob456"},
54.         ]
56.         # 发送请求
57.         resp = client.get("/api/users")
58.         assert resp.status_code == 200
60.         resp_headers = resp.headers
61.         assert resp_headers.get("Content-Type") == "application/json"
62.         assert "X-Tracking-ID" in resp_headers
63.         assert "X-DateTime" in resp_headers
65.         # resp_body = resp.json
67.         mock_get_users.assert_called_once()

复制代码

执行测试

1. pytest -vv

复制代码

配置 Gunicorn

代码文件 gunicorn.conf.py。简单配置了一些启动参数，以及请求日志的格式。

1. # Gunicorn 配置文件
2. from pathlib import Path
3. from multiprocessing import cpu_count
4. import gunicorn.glogging
5. from datetime import datetime
7. class CustomLogger(gunicorn.glogging.Logger):
8.     def atoms(self, resp, req, environ, request_time):
9.         """
10.         重写 atoms 方法来自定义日志占位符
11.         """
12.         # 获取默认的所有占位符数据
13.         atoms = super().atoms(resp, req, environ, request_time)
14.         
15.         # 自定义 't' (时间戳) 的格式
16.         now = datetime.now().astimezone()
17.         atoms['t'] = now.isoformat(timespec="seconds")
18.         
19.         return atoms
20.    
22. # 预加载应用代码
23. preload_app = True
25. # 工作进程数量：通常是 CPU 核心数的 2 倍加 1
26. # workers = int(cpu_count() * 2 + 1)
27. workers = 2
29. # 使用 gevent 异步 worker 类型，适合 I/O 密集型应用
30. # 注意：gevent worker 不使用 threads 参数，而是使用协程进行并发处理
31. worker_class = "gevent"
33. # 每个 gevent worker 可处理的最大并发连接数
34. worker_connections = 2000
36. # 绑定地址和端口
37. bind = "127.0.0.1:8000"
39. # 进程名称
40. proc_name = "flask-dev"
42. # PID 文件路径
43. pidfile = str(Path(__file__).parent / "tmp" / "gunicorn.pid")
45. logger_class = CustomLogger
46. access_log_format = (
47.     '{"@timestamp": "%(t)s", '
48.     '"remote_addr": "%(h)s", '
49.     '"protocol": "%(H)s", '
50.     '"host": "%({host}i)s", '
51.     '"request_method": "%(m)s", '
52.     '"request_path": "%(U)s", '
53.     '"status_code": %(s)s, '
54.     '"response_length": %(b)s, '
55.     '"referer": "%(f)s", '
56.     '"user_agent": "%(a)s", '
57.     '"x_tracking_id": "%({x-tracking-id}i)s", '
58.     '"request_time": %(L)s}'
59. )
61. # 访问日志路径
62. accesslog = str(Path(__file__).parent / "logs" / "access.log")
64. # 错误日志路径
65. errorlog = str(Path(__file__).parent / "logs" / "error.log")
67. # 日志级别
68. loglevel = "debug"

复制代码

输出的日志格式。可以看到日志格式符合 JSON 规范，便于 Filebeat 收集后在 Kibana 上检索。

1. $ tail -n 1 logs/access.log | python3 -m json.tool
2. {
3.     "@timestamp": "2026-01-17T16:44:37+08:00",
4.     "remote_addr": "127.0.0.1",
5.     "protocol": "HTTP/1.1",
6.     "host": "127.0.0.1:8000",
7.     "request_method": "GET",
8.     "request_path": "/api/users",
9.     "status_code": 200,
10.     "response_length": 110,
11.     "referer": "-",
12.     "user_agent": "curl/8.14.1",
13.     "x_tracking_id": "123456",
14.     "request_time": 0.102042
15. }

复制代码

补充

全局对象 g 的注意事项

• g 不是进程或线程共享的全局变量，请只在请求处理流程中使用 g。
  
• 如果视图函数中启动了后台线程或异步任务，在子线程中直接访问 g 通常会报错或获取不到数据。这时建议显式传递数据。
  
• 不要在 g 中存储大文件或数据对象，否则会占用过高内存。
  
• g 不是 session。
  

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

感谢发布原创作品，程序园因你更精彩

感谢分享，学习下。

这个好，看起来很实用