博客 / 詳情

返回

[python]Flask - Tracking ID的設計

前言

在實際業務中，根據 tracking_id 追溯一條請求的完整處理路徑是比較常見的需求。藉助 Flask 自帶的全局對象 g 以及鈎子函數，可以很容易地為每條請求添加 tracking_id，並在日誌中自動記錄。

主要內容：

  • 如何為每條請求添加 tracking_id
  • 如何為日誌自動添加 tracking_id 記錄
  • 如何自定義響應類，實現統一的響應格式，並在響應頭中添加 tracking_id
  • 視圖函數單元測試示例
  • Gunicorn 配置

項目結構

雖然內容看起來很多，但 tracking_id 的實現其實很簡單。本文按照生產項目的規範組織了代碼，添加了 Gunicorn 配置和單元測試代碼，以及規範了日誌格式和 JSON 響應格式。

├── 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  # 生產環境部署一般依賴這兩個
uv add --dev pytest           # 測試庫

實現添加 tracking_id 的中間件

代碼文件：middlewares/tracking_id.py

from uuid import uuid4

from flask import Flask, Response, g, request


def tracking_id_middleware(app: Flask):
    """
    跟蹤 ID 中間件
    為每個請求生成或獲取跟蹤 ID，用於追蹤請求鏈路
    """
    
    @app.before_request
    def tracking_id_before_request():
        """
        請求前處理函數
        檢查請求頭中是否包含 X-Tracking-ID，如果沒有則生成一個新的 UUID 作為跟蹤 ID
        並將其存儲到 Flask 的全局對象 g 中，供後續處理使用
        """
        # 從請求頭中獲取 X-Tracking-ID
        tracking_id = request.headers.get("X-Tracking-ID")
        if not tracking_id:
            # 如果請求頭中沒有 X-Tracking-ID，則生成一個新的 UUID
            tracking_id = str(uuid4())
        # 將跟蹤 ID 存儲到 Flask 的全局對象 g 中，供後續處理使用
        g.tracking_id = tracking_id

    @app.after_request
    def tracking_id_after_request(response: Response):
        """
        請求後處理函數
        將跟蹤 ID 添加到響應頭中，以便客户端知道本次請求的跟蹤 ID
        """
        # 檢查響應頭中是否已經有 X-Tracking-ID
        tracking_id = response.headers.get("X-Tracking-ID", "")
        if not tracking_id:
            # 如果響應頭中沒有 X-Tracking-ID，則從全局對象 g 中獲取
            tracking_id = g.get("tracking_id", "")
            # 將跟蹤 ID 添加到響應頭中
            response.headers["X-Tracking-ID"] = tracking_id
        return response

    # 返回應用實例
    return app

代碼文件 middlewares/__init__.py，方便其他模塊導入

from .tracking_id import tracking_id_middleware

__all__ = [
    "tracking_id_middleware",
]

日誌模塊 - 自動記錄 tracking_id

實現一個簡單的輸出到控制枱的日誌模塊，日誌格式為 JSON，自動添加 tracking_id 到日誌中，避免手動在 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,  # 如需記錄進程名可取消註釋
            "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:
    """設置控制枱日誌處理器。

    Args:
        level (int): 日誌級別。
    """
    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，實現單例調用。

from .app_log import setup_app_logger

logger = setup_app_logger()

__all__ = ["logger"]

自定義響應類

規範 JSON 類型的響應格式，並在響應頭中添加 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 狀態碼: 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 狀態碼: 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 狀態碼: 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 狀態碼: 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，方便其他模塊調用。

from .json_response import (
    ArgumentInvalid,
    ArgumentNotFound,
    AuthFailed,
    Fail,
    JsonResponse,
    ResourceConflict,
    ResourceForbidden,
    ResourceNotFound,
    Success,
)

__all__ = [
    "JsonResponse",
    "Success",
    "Fail",
    "ArgumentNotFound",
    "ArgumentInvalid",
    "AuthFailed",
    "ResourceConflict",
    "ResourceNotFound",
    "ResourceForbidden",
]

編寫視圖函數

代碼文件 apis/common/common.py。以下定義了 5 個路由，主要用於測試響應類是否正常返回 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 調用了 handles/ 中的代碼，模擬查詢數據庫。handles/user.py 中的代碼如下：

import time
from typing import Any, Dict, List


def get_users() -> List[Dict[str, Any]]:
    # 模擬查詢用户數據
    time.sleep(0.1)  # 模擬延遲
    users = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
    return users

代碼文件 apis/common/__init__.py 中導入各個藍圖並統一暴露。由於示例代碼只定義了一個藍圖，所以這裏寫得很簡單。如果有多個藍圖，可以把藍圖都添加到一個列表中，在 Flask 應用中一次性遍歷註冊。

from .common import route
# from .common import route as common_route

# routes = [
#     common_route,
# ]

__all__ = ["route"]

代碼文件 apis/__init__.py 中提供 Flask 應用的工廠函數。

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



# 錯誤處理器
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__)

    # 註冊中間件
    app = tracking_id_middleware(app)

    # 註冊錯誤處理器
    app.errorhandler(Exception)(error_handler_generic)
    app.errorhandler(404)(error_handler_notfound)

    # 註冊藍圖
    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)

簡單運行測試

  1. 啓動應用
# 方式1, 直接啓動, 用於簡單測試
python main.py

# 方式2, 使用 gunicorn, 這是生產環境啓動方式. 配置文件默認路徑即 ./gunicorn.conf.py
gunicorn main:app
  1. curl 請求 /api/health。可以看到響應頭中已經有了 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"}
  1. curl 請求 /api/users。手動指定請求頭中的 X-Tracking-ID，響應時也會保持相同的 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"}]}

編寫單元測試

使用 pytest 進行單元測試，這裏只是一個簡單的示例

配置 pytest

配置文件 pytest.ini

[pytest]
testpaths = "tests"
pythonpath = "."

測試代碼

代碼文件 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"},
        ]

        # 發送請求
        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()

執行測試

pytest -vv

配置 Gunicorn

代碼文件 gunicorn.conf.py。簡單配置了一些啓動參數，以及請求日誌的格式。

# 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 方法來自定義日誌佔位符
        """
        # 獲取默認的所有佔位符數據
        atoms = super().atoms(resp, req, environ, request_time)
        
        # 自定義 't' (時間戳) 的格式
        now = datetime.now().astimezone()
        atoms['t'] = now.isoformat(timespec="seconds")
        
        return atoms
    

# 預加載應用代碼
preload_app = True

# 工作進程數量：通常是 CPU 核心數的 2 倍加 1
# workers = int(cpu_count() * 2 + 1)
workers = 2

# 使用 gevent 異步 worker 類型，適合 I/O 密集型應用
# 注意：gevent worker 不使用 threads 參數，而是使用協程進行併發處理
worker_class = "gevent"

# 每個 gevent worker 可處理的最大併發連接數
worker_connections = 2000

# 綁定地址和端口
bind = "127.0.0.1:8000"

# 進程名稱
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")

# 錯誤日誌路徑
errorlog = str(Path(__file__).parent / "logs" / "error.log")

# 日誌級別
loglevel = "debug"

輸出的日誌格式。可以看到日誌格式符合 JSON 規範，便於 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
}

補充

全局對象 g 的注意事項

  1. g 不是進程或線程共享的全局變量，請只在請求處理流程中使用 g。
  2. 如果視圖函數中啓動了後台線程或異步任務，在子線程中直接訪問 g 通常會報錯或獲取不到數據。這時建議顯式傳遞數據。
  3. 不要在 g 中存儲大文件或數據對象，否則會佔用過高內存。
  4. g 不是 session。
後端
user avatar
0 位用戶收藏了這個故事！

發佈 評論

Some HTML is okay.