在 FastAPI 專案建立客製化日誌模組

FastAPI
, ,

實作日誌模組

建立 logger.py 檔。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import logging
import sys

logging_config = {
    "version": 1,
    "formatters": {"simple": {"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"}},
    "handlers": {
        "console": {
            "level": "DEBUG",
            "class": "logging.StreamHandler",
            "formatter": "simple",
            "stream": sys.stderr,
        }
    },
    "root": {"level": "DEBUG", "handlers": ["console"], "propagate": True},
}

logging.config.dictConfig(logging_config)

logger = logging.getLogger()

使用如下：

1
2
3
from logger import logger

logger.info("Hello, World!")

輸出如下：

1
2024-08-29 23:25:25,665 - root - INFO - Hello, World!

實作中介層

建立 middleware/logging.py 中介層。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import logging
from typing import Callable
from uuid import uuid4

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response


class LoggingMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: Callable, *, logger: logging.Logger) -> None:
        self._logger = logger
        super().__init__(app)

    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        request_id: str = str(uuid4())
        request_info = await self._log_request(request)

        try:
            response = await call_next(request)
            response.headers["X-REQUEST-ID"] = request_id
            log_entry = {
                "request_id": request_id,
                "request": request_info,
                "response": await self._log_response(response),
            }
            self._logger.info(log_entry)
        except Exception as e:
            log_entry = {"request_id": request_id, "request": request_info, "error": str(e)}
            self._logger.error(log_entry)
            raise

        return response

    async def _log_request(self, request: Request) -> dict:
        headers = {k: v for k, v in request.headers.items() if k.lower() != "authorization"}
        body = await request.body()
        request_body = body.decode("utf-8") if body else ""
        request_info = {
            "method": request.method,
            "url": str(request.url),
            "headers": headers,
            "body": request_body,
        }

        return request_info

    async def _log_response(self, response: Response) -> dict:
        response_info = {
            "status_code": response.status_code,
            "headers": dict(response.headers),
        }

        return response_info

修改 main.py 檔，添加 logging 中介層。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from handlers import book_handler, user_handler
from logger import logger
from middleware.logging import LoggingMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    allow_credentials=True,
)
app.add_middleware(LoggingMiddleware, logger=logger)

app.include_router(user_handler.router)

啟動網頁伺服器。

1
uvicorn main:app --reload --port 8000

呼叫 API 端點。

1
curl http://localhost:8000/api

輸出如下：

1
2024-08-29 23:28:40,989 - root - INFO - {'request_id': 'f2688c87-51b6-4edb-90f2-2d6e67c02bee', 'request': {'method': 'GET', 'url': 'http://localhost:8000/api', 'headers': {'host': 'localhost:8000', 'user-agent': 'curl/8.4.0', 'accept': '*/*'}, 'body': ''}, 'response': {'status_code': 200, 'headers': {'content-length': '15', 'content-type': 'application/json', 'x-request-id': 'f2688c87-51b6-4edb-90f2-2d6e67c02bee'}}}

參考資料