├── 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ù)

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

from .tracking_id import tracking_id_middleware

__all__ = [
    "tracking_id_middleware",
]

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

from .app_log import setup_app_logger

logger = setup_app_logger()

__all__ = ["logger"]

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)

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

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

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")

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

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

# routes = [
#     common_route,
# ]

__all__ = ["route"]

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",
]

from apis import create_app

app = create_app()

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8000, debug=False)

# 方式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 -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 -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]
testpaths = "tests"
pythonpath = "."

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()

pytest -vv

# 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"

$ 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
}