Flask中Tracking ID請(qǐng)求跟蹤的設(shè)計(jì)實(shí)現(xiàn)

更新時(shí)間：2026年02月05日 08:56:57 作者：花酒鋤作田

本文詳細(xì)介紹如何在 Flask 應(yīng)用中實(shí)現(xiàn)請(qǐng)求跟蹤 ID功能,包括中間件設(shè)計(jì)、日志記錄、響應(yīng)格式化等完整方案,幫助開發(fā)者實(shí)現(xiàn)請(qǐng)求鏈路追蹤,感興趣的可以了解一下

前言

在實(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)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: