Flask中Tracking ID請(qǐng)求跟蹤的設(shè)計(jì)實(shí)現(xiàn)
前言
在實(shí)際業(yè)務(wù)中,根據(jù) tracking_id 追溯一條請(qǐng)求的完整處理路徑是比較常見的需求。借助 Flask 自帶的全局對(duì)象 g 以及鉤子函數(shù),可以很容易地為每條請(qǐng)求添加 tracking_id,并在日志中自動(dòng)記錄。
主要內(nèi)容:
- 如何為每條請(qǐng)求添加
tracking_id - 如何為日志自動(dòng)添加
tracking_id記錄 - 如何自定義響應(yīng)類,實(shí)現(xiàn)統(tǒng)一的響應(yīng)格式,并在響應(yīng)頭中添加
tracking_id - 視圖函數(shù)單元測(cè)試示例
- Gunicorn 配置
項(xiàng)目結(jié)構(gòu)
雖然內(nèi)容看起來(lái)很多,但 tracking_id 的實(shí)現(xiàn)其實(shí)很簡(jiǎn)單。本文按照生產(chǎn)項(xiàng)目的規(guī)范組織了代碼,添加了 Gunicorn 配置和單元測(cè)試代碼,以及規(guī)范了日志格式和 JSON 響應(yīng)格式。
├── apis │ ├── common │ │ ├── common.py │ │ └── __init__.py │ └── __init__.py ├── gunicorn.conf.py ├── handles │ └── user.py ├── logs │ ├── access.log │ └── error.log ├── main.py ├── middlewares │ ├── __init__.py │ └── tracking_id.py ├── pkgs │ └── log │ ├── app_log.py │ └── __init__.py ├── pyproject.toml ├── pytest.ini ├── README.md ├── responses │ ├── __init__.py │ └── json_response.py ├── tests │ └── apis │ └── test_common.py ├── tmp │ └── gunicorn.pid └── uv.lock
安裝依賴
uv add flask uv add gunicorn gevent # 生產(chǎn)環(huán)境部署一般依賴這兩個(gè) uv add --dev pytest # 測(cè)試庫(kù)
實(shí)現(xiàn)添加 tracking_id 的中間件
代碼文件:middlewares/tracking_id.py
from uuid import uuid4
from flask import Flask, Response, g, request
def tracking_id_middleware(app: Flask):
"""
跟蹤 ID 中間件
為每個(gè)請(qǐng)求生成或獲取跟蹤 ID,用于追蹤請(qǐng)求鏈路
"""
@app.before_request
def tracking_id_before_request():
"""
請(qǐng)求前處理函數(shù)
檢查請(qǐng)求頭中是否包含 X-Tracking-ID,如果沒有則生成一個(gè)新的 UUID 作為跟蹤 ID
并將其存儲(chǔ)到 Flask 的全局對(duì)象 g 中,供后續(xù)處理使用
"""
# 從請(qǐng)求頭中獲取 X-Tracking-ID
tracking_id = request.headers.get("X-Tracking-ID")
if not tracking_id:
# 如果請(qǐng)求頭中沒有 X-Tracking-ID,則生成一個(gè)新的 UUID
tracking_id = str(uuid4())
# 將跟蹤 ID 存儲(chǔ)到 Flask 的全局對(duì)象 g 中,供后續(xù)處理使用
g.tracking_id = tracking_id
@app.after_request
def tracking_id_after_request(response: Response):
"""
請(qǐng)求后處理函數(shù)
將跟蹤 ID 添加到響應(yīng)頭中,以便客戶端知道本次請(qǐng)求的跟蹤 ID
"""
# 檢查響應(yīng)頭中是否已經(jīng)有 X-Tracking-ID
tracking_id = response.headers.get("X-Tracking-ID", "")
if not tracking_id:
# 如果響應(yīng)頭中沒有 X-Tracking-ID,則從全局對(duì)象 g 中獲取
tracking_id = g.get("tracking_id", "")
# 將跟蹤 ID 添加到響應(yīng)頭中
response.headers["X-Tracking-ID"] = tracking_id
return response
# 返回應(yīng)用實(shí)例
return app
代碼文件 middlewares/__init__.py,方便其他模塊導(dǎo)入
from .tracking_id import tracking_id_middleware
__all__ = [
"tracking_id_middleware",
]
日志模塊 - 自動(dòng)記錄 tracking_id
實(shí)現(xiàn)一個(gè)簡(jiǎn)單的輸出到控制臺(tái)的日志模塊,日志格式為 JSON,自動(dòng)添加 tracking_id 到日志中,避免手動(dòng)在 logger.info() 這類方法中傳入 tracking_id。
代碼文件 pkgs/log/app_log.py
import json
import logging
import sys
from flask import g
class JSONFormatter(logging.Formatter):
"""日志格式化器,輸出 JSON 格式的日志。"""
def format(self, record: logging.LogRecord) -> str:
log_record = {
"@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
"level": record.levelname,
"name": record.name,
# "processName": record.processName, # 如需記錄進(jìn)程名可取消注釋
"tracking_id": getattr(record, "tracking_id", None),
"loc": "%s:%d" % (record.filename, record.lineno),
"func": record.funcName,
"message": record.getMessage(),
}
return json.dumps(log_record, ensure_ascii=False, default=str)
class TrackingIDFilter(logging.Filter):
"""日志過濾器,為日志記錄添加 tracking_id。"""
def filter(self, record):
record.tracking_id = g.get("tracking_id", None)
return True
def _setup_console_handler(level: int) -> logging.StreamHandler:
"""設(shè)置控制臺(tái)日志處理器。
Args:
level (int): 日志級(jí)別。
"""
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(level)
handler.setFormatter(JSONFormatter())
return handler
def setup_app_logger(level: int = logging.INFO, name: str = "app") -> logging.Logger:
logger = logging.getLogger(name)
if logger.hasHandlers():
return logger
logger.setLevel(level)
logger.propagate = False
logger.addHandler(_setup_console_handler(level))
logger.addFilter(TrackingIDFilter())
return logger
在 pkgs/log/__init__.py 中初始化 logger,實(shí)現(xiàn)單例調(diào)用。
from .app_log import setup_app_logger logger = setup_app_logger() __all__ = ["logger"]
自定義響應(yīng)類
規(guī)范 JSON 類型的響應(yīng)格式,并在響應(yīng)頭中添加 X-Tracking-ID 和 X-DateTime。
代碼文件 responses/json_response.py
import json
from datetime import datetime
from http import HTTPStatus
from typing import Any
from flask import Response, g, request
class JsonResponse(Response):
def __init__(
self,
data: Any = None,
code: HTTPStatus = HTTPStatus.OK,
msg: str = "this is a json response",
):
x_tracking_id = g.get("tracking_id", "")
x_datetime = datetime.now().astimezone().isoformat(timespec="seconds")
resp_headers = {
"Content-Type": "application/json",
"X-Tracking-ID": x_tracking_id,
"X-DateTime": x_datetime,
}
try:
resp = json.dumps(
{
"code": code.value,
"msg": msg,
"data": data,
},
ensure_ascii=False,
default=str,
)
except Exception as e:
resp = json.dumps(
{
"code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
"msg": f"Response serialization error: {str(e)}",
"data": None,
}
)
super().__init__(response=resp, status=code.value, headers=resp_headers)
class Success(JsonResponse):
def __init__(self, data: Any = None, msg: str = ""):
if not msg:
msg = f"{request.method} {request.path} success"
super().__init__(data=data, code=HTTPStatus.OK, msg=msg)
class Fail(JsonResponse):
def __init__(self, msg: str = "", data: Any = None):
if not msg:
msg = f"{request.method} {request.path} failed"
super().__init__(data=data, code=HTTPStatus.INTERNAL_SERVER_ERROR, msg=msg)
class ArgumentNotFound(JsonResponse):
def __init__(self, msg: str = "", data: Any = None):
if not msg:
msg = f"{request.method} {request.path} argument not found"
super().__init__(data=data, code=HTTPStatus.BAD_REQUEST, msg=msg)
class ArgumentInvalid(JsonResponse):
def __init__(self, msg: str = "", data: Any = None):
if not msg:
msg = f"{request.method} {request.path} argument invalid"
super().__init__(data=data, code=HTTPStatus.BAD_REQUEST, msg=msg)
class AuthFailed(JsonResponse):
"""HTTP 狀態(tài)碼: 401"""
def __init__(self, msg: str = "", data: Any = None):
if not msg:
msg = f"{request.method} {request.path} auth failed"
super().__init__(data=data, code=HTTPStatus.UNAUTHORIZED, msg=msg)
class ResourceConflict(JsonResponse):
"""HTTP 狀態(tài)碼: 409"""
def __init__(self, msg: str = "", data: Any = None):
if not msg:
msg = f"{request.method} {request.path} resource conflict"
super().__init__(data=data, code=HTTPStatus.CONFLICT, msg=msg)
class ResourceNotFound(JsonResponse):
"""HTTP 狀態(tài)碼: 404"""
def __init__(self, msg: str = "", data: Any = None):
if not msg:
msg = f"{request.method} {request.path} resource not found"
super().__init__(data=data, code=HTTPStatus.NOT_FOUND, msg=msg)
class ResourceForbidden(JsonResponse):
"""HTTP 狀態(tài)碼: 403"""
def __init__(self, msg: str = "", data: Any = None):
if not msg:
msg = f"{request.method} {request.path} resource forbidden"
super().__init__(data=data, code=HTTPStatus.FORBIDDEN, msg=msg)
代碼文件 responses/__init__.py,方便其他模塊調(diào)用。
from .json_response import (
ArgumentInvalid,
ArgumentNotFound,
AuthFailed,
Fail,
JsonResponse,
ResourceConflict,
ResourceForbidden,
ResourceNotFound,
Success,
)
__all__ = [
"JsonResponse",
"Success",
"Fail",
"ArgumentNotFound",
"ArgumentInvalid",
"AuthFailed",
"ResourceConflict",
"ResourceNotFound",
"ResourceForbidden",
]
編寫視圖函數(shù)
代碼文件 apis/common/common.py。以下定義了 5 個(gè)路由,主要用于測(cè)試響應(yīng)類是否正常返回 JSON 格式。
from datetime import datetime
from flask import Blueprint
from handles import user as user_handle
from pkgs.log import logger
from responses import Success
route = Blueprint("common_apis", __name__, url_prefix="/api")
@route.get("/health")
def health_check():
# print(g.get("tracking_id", "no-tracking-id"))
logger.info("Health check")
return Success(data="OK")
@route.get("/users")
def get_users():
users = user_handle.get_users()
return Success(data=users)
@route.get("/names")
def get_names():
names = ["Alice", "Bob", "Charlie"]
return Success(data=names)
@route.get("/item")
def get_item():
item = {"id": 101, "name": "Sample Item", "price": 29.99, "now": datetime.now()}
return Success(data=item)
@route.get("/error")
def get_error():
raise Exception("This is a test exception")
GET /api/users 調(diào)用了 handles/ 中的代碼,模擬查詢數(shù)據(jù)庫(kù)。handles/user.py 中的代碼如下:
import time
from typing import Any, Dict, List
def get_users() -> List[Dict[str, Any]]:
# 模擬查詢用戶數(shù)據(jù)
time.sleep(0.1) # 模擬延遲
users = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
return users
代碼文件 apis/common/__init__.py 中導(dǎo)入各個(gè)藍(lán)圖并統(tǒng)一暴露。由于示例代碼只定義了一個(gè)藍(lán)圖,所以這里寫得很簡(jiǎn)單。如果有多個(gè)藍(lán)圖,可以把藍(lán)圖都添加到一個(gè)列表中,在 Flask 應(yīng)用中一次性遍歷注冊(cè)。
from .common import route # from .common import route as common_route # routes = [ # common_route, # ] __all__ = ["route"]
代碼文件 apis/__init__.py 中提供 Flask 應(yīng)用的工廠函數(shù)。
import traceback
from flask import Flask
from apis.common import route as common_route
from middlewares import tracking_id_middleware
from responses import Fail, ResourceNotFound
from pkgs.log import logger
# 錯(cuò)誤處理器
def error_handler_notfound(error):
return ResourceNotFound()
def error_handler_generic(error):
logger.error(traceback.format_exc())
return Fail(data=str(error))
def create_app() -> Flask:
app = Flask(__name__)
# 注冊(cè)中間件
app = tracking_id_middleware(app)
# 注冊(cè)錯(cuò)誤處理器
app.errorhandler(Exception)(error_handler_generic)
app.errorhandler(404)(error_handler_notfound)
# 注冊(cè)藍(lán)圖
app.register_blueprint(common_route)
return app
__all__ = [
"create_app",
]
入口代碼文件 main.py
from apis import create_app
app = create_app()
if __name__ == "__main__":
app.run(host="127.0.0.1", port=8000, debug=False)
簡(jiǎn)單運(yùn)行測(cè)試
- 啟動(dòng)應(yīng)用
# 方式1, 直接啟動(dòng), 用于簡(jiǎn)單測(cè)試 python main.py # 方式2, 使用 gunicorn, 這是生產(chǎn)環(huán)境啟動(dòng)方式. 配置文件默認(rèn)路徑即 ./gunicorn.conf.py gunicorn main:app
- curl 請(qǐng)求
/api/health??梢钥吹巾憫?yīng)頭中已經(jīng)有了X-Tracking-ID和X-DateTime
$ curl -v http://127.0.0.1:8000/api/health
* Trying 127.0.0.1:8000...
* Connected to 127.0.0.1 (127.0.0.1) port 8000
* using HTTP/1.x
> GET /api/health HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/8.14.1
> Accept: */*
>
* Request completely sent off
< HTTP/1.1 200 OK
< Server: gunicorn
< Date: Sat, 17 Jan 2026 08:41:07 GMT
< Connection: keep-alive
< Content-Type: application/json
< X-Tracking-ID: 1f0adb8d-9bee-49d4-873f-31aa1437da60
< X-DateTime: 2026-01-17T16:41:07+08:00
< Content-Length: 61
<
* Connection #0 to host 127.0.0.1 left intact
{"code": 200, "msg": "GET /api/health success", "data": "OK"}
- curl 請(qǐng)求
/api/users。手動(dòng)指定請(qǐng)求頭中的X-Tracking-ID,響應(yīng)時(shí)也會(huì)保持相同的 ID。
$ curl -v http://127.0.0.1:8000/api/users -H 'X-Tracking-ID:123456'
* Trying 127.0.0.1:8000...
* Connected to 127.0.0.1 (127.0.0.1) port 8000
* using HTTP/1.x
> GET /api/users HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/8.14.1
> Accept: */*
> X-Tracking-ID:123456
>
* Request completely sent off
< HTTP/1.1 200 OK
< Server: gunicorn
< Date: Sat, 17 Jan 2026 08:44:37 GMT
< Connection: keep-alive
< Content-Type: application/json
< X-Tracking-ID: 123456
< X-DateTime: 2026-01-17T16:44:37+08:00
< Content-Length: 110
<
* Connection #0 to host 127.0.0.1 left intact
{"code": 200, "msg": "GET /api/users success", "data": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}
編寫單元測(cè)試
使用 pytest 進(jìn)行單元測(cè)試,這里只是一個(gè)簡(jiǎn)單的示例
配置 pytest
配置文件 pytest.ini
[pytest] testpaths = "tests" pythonpath = "."
測(cè)試代碼
代碼文件 tests/apis/test_common.py
from typing import Generator
from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
from flask.testing import FlaskClient
from apis.common import route as common_route
@pytest.fixture
def app() -> Generator[Flask, None, None]:
app = Flask(__name__)
app.config.update(
{
"TESTING": True,
"DEBUG": False,
}
)
app.register_blueprint(common_route)
yield app
@pytest.fixture
def client(app: Flask) -> FlaskClient:
return app.test_client()
class TestGetHealth:
def test_get_health_success(self, client: FlaskClient) -> None:
resp = client.get("/api/health")
assert resp.status_code == 200
resp_headers = resp.headers
assert resp_headers.get("Content-Type") == "application/json"
assert "X-Tracking-ID" in resp_headers
assert "X-DateTime" in resp_headers
resp_body = resp.json
assert resp_body == {
"code": 200,
"msg": "GET /api/health success",
"data": "OK",
}
class TestGetUsers:
@patch("apis.common.common.user_handle.get_users")
def test_get_users(self, mock_get_users: MagicMock, client: FlaskClient) -> None:
# mock user.get_users() 的返回值
mock_get_users.return_value = [
{"id": 1, "name": "Alice123"},
{"id": 2, "name": "Bob456"},
]
# 發(fā)送請(qǐng)求
resp = client.get("/api/users")
assert resp.status_code == 200
resp_headers = resp.headers
assert resp_headers.get("Content-Type") == "application/json"
assert "X-Tracking-ID" in resp_headers
assert "X-DateTime" in resp_headers
# resp_body = resp.json
mock_get_users.assert_called_once()
執(zhí)行測(cè)試
pytest -vv
配置 Gunicorn
代碼文件 gunicorn.conf.py。簡(jiǎn)單配置了一些啟動(dòng)參數(shù),以及請(qǐng)求日志的格式。
# Gunicorn 配置文件
from pathlib import Path
from multiprocessing import cpu_count
import gunicorn.glogging
from datetime import datetime
class CustomLogger(gunicorn.glogging.Logger):
def atoms(self, resp, req, environ, request_time):
"""
重寫 atoms 方法來(lái)自定義日志占位符
"""
# 獲取默認(rèn)的所有占位符數(shù)據(jù)
atoms = super().atoms(resp, req, environ, request_time)
# 自定義 't' (時(shí)間戳) 的格式
now = datetime.now().astimezone()
atoms['t'] = now.isoformat(timespec="seconds")
return atoms
# 預(yù)加載應(yīng)用代碼
preload_app = True
# 工作進(jìn)程數(shù)量:通常是 CPU 核心數(shù)的 2 倍加 1
# workers = int(cpu_count() * 2 + 1)
workers = 2
# 使用 gevent 異步 worker 類型,適合 I/O 密集型應(yīng)用
# 注意:gevent worker 不使用 threads 參數(shù),而是使用協(xié)程進(jìn)行并發(fā)處理
worker_class = "gevent"
# 每個(gè) gevent worker 可處理的最大并發(fā)連接數(shù)
worker_connections = 2000
# 綁定地址和端口
bind = "127.0.0.1:8000"
# 進(jìn)程名稱
proc_name = "flask-dev"
# PID 文件路徑
pidfile = str(Path(__file__).parent / "tmp" / "gunicorn.pid")
logger_class = CustomLogger
access_log_format = (
'{"@timestamp": "%(t)s", '
'"remote_addr": "%(h)s", '
'"protocol": "%(H)s", '
'"host": "%({host}i)s", '
'"request_method": "%(m)s", '
'"request_path": "%(U)s", '
'"status_code": %(s)s, '
'"response_length": %(b)s, '
'"referer": "%(f)s", '
'"user_agent": "%(a)s", '
'"x_tracking_id": "%({x-tracking-id}i)s", '
'"request_time": %(L)s}'
)
# 訪問日志路徑
accesslog = str(Path(__file__).parent / "logs" / "access.log")
# 錯(cuò)誤日志路徑
errorlog = str(Path(__file__).parent / "logs" / "error.log")
# 日志級(jí)別
loglevel = "debug"
輸出的日志格式。可以看到日志格式符合 JSON 規(guī)范,便于 Filebeat 收集后在 Kibana 上檢索。
$ tail -n 1 logs/access.log | python3 -m json.tool
{
"@timestamp": "2026-01-17T16:44:37+08:00",
"remote_addr": "127.0.0.1",
"protocol": "HTTP/1.1",
"host": "127.0.0.1:8000",
"request_method": "GET",
"request_path": "/api/users",
"status_code": 200,
"response_length": 110,
"referer": "-",
"user_agent": "curl/8.14.1",
"x_tracking_id": "123456",
"request_time": 0.102042
}
補(bǔ)充
全局對(duì)象 g 的注意事項(xiàng)
g不是進(jìn)程或線程共享的全局變量,請(qǐng)只在請(qǐng)求處理流程中使用g。- 如果視圖函數(shù)中啟動(dòng)了后臺(tái)線程或異步任務(wù),在子線程中直接訪問
g通常會(huì)報(bào)錯(cuò)或獲取不到數(shù)據(jù)。這時(shí)建議顯式傳遞數(shù)據(jù)。 - 不要在
g中存儲(chǔ)大文件或數(shù)據(jù)對(duì)象,否則會(huì)占用過高內(nèi)存。 g不是session。
到此這篇關(guān)于Flask中Tracking ID請(qǐng)求跟蹤的設(shè)計(jì)實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Flask Tracking ID內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Pytorch神經(jīng)網(wǎng)絡(luò)參數(shù)管理方法詳細(xì)講解
這篇文章主要介紹了Pytorch神經(jīng)網(wǎng)絡(luò)參數(shù)管理方法,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧2023-05-05
基于TensorBoard中g(shù)raph模塊圖結(jié)構(gòu)分析
今天小編就為大家分享一篇基于TensorBoard中g(shù)raph模塊圖結(jié)構(gòu)分析,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來(lái)看看吧2020-02-02
淺析Python中壓縮zipfile與解壓縮tarfile模塊的使用
Python?提供了兩個(gè)標(biāo)準(zhǔn)庫(kù)模塊來(lái)處理文件的壓縮和解壓縮操作:zipfile和tarfile,本文將分享?這兩個(gè)模塊的使用方法,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2023-10-10
使用Pandas將inf, nan轉(zhuǎn)化成特定的值
今天小編就為大家分享一篇使用Pandas將inf, nan轉(zhuǎn)化成特定的值,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來(lái)看看吧2019-12-12
解決python中遇到字典里key值為None的情況,取不出來(lái)的問題
今天小編就為大家分享一篇解決python中遇到字典里key值為None的情況,取不出來(lái)的問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來(lái)看看吧2018-10-10
Python?numpy中np.random.seed()的詳細(xì)用法實(shí)例
在學(xué)習(xí)人工智能時(shí),大量的使用了np.random.seed(),利用隨機(jī)數(shù)種子,使得每次生成的隨機(jī)數(shù)相同,下面這篇文章主要給大家介紹了關(guān)于Python?numpy中np.random.seed()的詳細(xì)用法,需要的朋友可以參考下2022-08-08
Python基于pip實(shí)現(xiàn)離線打包過程詳解
這篇文章主要介紹了Python基于pip實(shí)現(xiàn)離線打包過程詳解,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-05-05
python中有關(guān)時(shí)間日期格式轉(zhuǎn)換問題
這篇文章主要介紹了python中有關(guān)時(shí)間日期格式轉(zhuǎn)換問題,本文給大家介紹的非常詳細(xì),具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2019-12-12

