FastAPI 實戰秘籍：從零構建高性能 API-日志篇

作者：老朱2000 2025-09-08 11:00:00

開發

?日志（Logging）在軟件開發和系統運維中扮演著至關重要的角色，它記錄了系統運行時的各種事件、狀態和錯誤信息。

接上文《FastAPI 實戰秘籍：從零構建高性能 API -配置篇》

日志（Logging）在軟件開發和系統運維中扮演著至關重要的角色，它記錄了系統運行時的各種事件、狀態和錯誤信息。

作用：

問題排查與調試
監控系統健康狀態
安全審計與合規
性能分析與優化
用戶行為分析
數據恢復與回溯

日志級別和應用場景：

DEBUG 開發調試細節（變量值、流程分支）
INFO 常規運行狀態（服務啟動、關鍵業務事件）
WARN 非預期但可恢復的問題（低磁盤空間、降級操作）
ERROR 需要干預的錯誤（外部API失敗、數據庫連接中斷）
FATAL 導致服務崩潰的嚴重錯誤（無法恢復的異常）

在我們web開發過程中一般通過app.log記錄全部日志，通過error.log記錄ERROR級別以上的日志，能夠快速定位、排查問題。

APP日志

在通過fastapi日志模塊中，我們使用標準模塊logging。通過配置文件配置日志文件，等級、文件大小輪轉、備份日志文件數目等。

# config/config.yml

logging:
level:"INFO"
log_path:"logs/app.log"
error_path:"logs/error.log"
rotate_size:10# 日志文件輪轉大小 單位M
back_files:5# 多保留n個備份

在配置解析腳本中config.py：

# config/config.py

class LoggingConfig(BaseModel):
    level: str = "INFO"
    log_path: str = "logs/app.log"
    error_path: str = "logs/error.log"
    rotate_size: int = 10
    back_files: int = 5

修改主配置模型：

# config/config.py

# 主配置模型中嵌套Logging
class Settings(BaseSettings):
    ...
    logging: LoggingConfig

在core目錄下創建logger.py模塊，定義日志句柄，格式，處理方式等。

import logging
import os
import queue
from enum import Enum
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler, QueueHandler, QueueListener
from typing import Dict, Callable, Optional

from config.config import BASE_DIR, get_settings

# 初始化配置
settings = get_settings()
logger = logging.getLogger(settings.app.name)

# 常量定義
LOG_FORMAT = "%(asctime)s - %(module)s - %(funcName)s - line:%(lineno)d - %(levelname)s - %(message)s"
LOG_ACCESS_FILE = os.path.join(BASE_DIR, settings.logging.log_path)
LOG_PATH = os.path.dirname(LOG_ACCESS_FILE)
HANDLER_TYPE = "rotate_file"

# 確保日志目錄存在
os.makedirs(LOG_PATH, exist_ok=True)


class LoggerLevel(Enum):
    """日志級別枚舉類"""
    DEBUG = ("DEBUG", logging.DEBUG)
    INFO = ("INFO", logging.INFO)
    WARNING = ("WARNING", logging.WARNING)
    ERROR = ("ERROR", logging.ERROR)
    CRITICAL = ("CRITICAL", logging.CRITICAL)

    @property
    def level(self) -> int:
        """獲取日志級別數值"""
        return self.value[1]

    @classmethod
    def get_by_config(cls, level_str: str) -> Optional['LoggerLevel']:
        """根據配置字符串獲取日志級別枚舉"""
        return next((log for log in cls if log.value[0] == level_str), None)


class LoggerFactory:
    """日志工廠類，集中管理日志處理器創建"""

    HANDLERS: Dict[str, Callable] = {
        "default": "_create_default_handler",
        "queue": "_create_queue_handler",
        "rotate_file": "_create_rotate_file_handler",
        "timed_rotate_file": "_create_timed_rotate_file_handler"
    }

    @classmethod
    def get_logger(cls) -> logging.Logger:
        """根據配置獲取適當的日志處理器"""
        handler_method = getattr(cls, cls.HANDLERS.get(HANDLER_TYPE, "_create_default_handler"))
        handler = handler_method()
        return cls._setup_logger(handler)

    @classmethod
    def _setup_logger(cls, handler: logging.Handler) -> logging.Logger:
        """配置日志記錄器"""
        # 清除現有處理器
        for h in logger.handlers[:]:
            logger.removeHandler(h)
            h.close()

        # 設置日志級別
        logger_level = LoggerLevel.get_by_config(settings.logging.level)
        logger.setLevel(logger_level.level if logger_level else logging.DEBUG)

        # 添加處理器
        handler.setFormatter(logging.Formatter(LOG_FORMAT))
        logger.addHandler(handler)
        return logger

    @classmethod
    def _create_default_handler(cls) -> logging.FileHandler:
        """創建默認文件處理器"""
        print("使用默認文件日志處理器")
        return logging.FileHandler(LOG_ACCESS_FILE, delay=True, encoding="utf-8")

    @classmethod
    def _create_queue_handler(cls) -> QueueHandler:
        """創建隊列日志處理器"""
        print("使用異步隊列日志處理器")
        log_queue = queue.Queue(-1)  # 無限大小隊列
        file_handler = logging.FileHandler(LOG_ACCESS_FILE, delay=True, encoding="utf-8")

        # 創建并啟動隊列監聽器
        listener = QueueListener(
            log_queue,
            file_handler,
            respect_handler_level=True
        )
        listener.start()

        return QueueHandler(log_queue)

    @classmethod
    def _create_rotate_file_handler(cls) -> RotatingFileHandler:
        """創建按文件大小輪轉的處理器"""
        print("使用文件大小輪轉日志處理器")
        return RotatingFileHandler(
            LOG_ACCESS_FILE,
            maxBytes=settings.logging.rotate_size * 1024 * 1024,
            backupCount=settings.logging.back_files,
            encoding="utf-8"
        )

    @classmethod
    def _create_timed_rotate_file_handler(cls) -> TimedRotatingFileHandler:
        """創建按時間輪轉的處理器"""
        print("使用時間輪轉日志處理器")
        return TimedRotatingFileHandler(
            LOG_ACCESS_FILE,
            when='midnight',
            interval=1,
            backupCount=settings.logging.back_files,
            encoding="utf-8"
        )


def get_logger() -> logging.Logger:
    """獲取配置好的日志記錄器"""
    return LoggerFactory.get_logger()

上面腳本中HANDLER_TYPE沒設置在配置文件中，僅開發人員能夠修改。 logging中大概有四種處理器：

FileHandler 阻塞寫入日志文件
QueueHandler 非阻塞隊列寫入日志
RotatingFileHandler 文件大小輪轉
TimedRotatingFileHandler 時間間隔輪轉

在__init__.py定義暴露方法：

from core.logger import get_logger
__all__ = [get_logger]

測試日志：

# main.py

def create_app():
    """啟動項目"""

    ...

    @app.get("/{form_id}")
    async def root(form_id: int = Path(..., gt=0)):
        get_logger().info(" ===aaa===")
        return {"message": form_id}

    return app

啟動服務，我們可以看到輸出的日志內容。

效果如下：

ERROR日志

為了更好的處理線上問題，排查問題。除了項目日志app.log外，系統異常報錯的日志記錄到error.log中。

Fastapi中可以通過裝飾器@app.exception_handler(Exception)記錄全局異常。

文檔地址：https://fastapi.tiangolo.com/zh/tutorial/handling-errors/#_4

在core包中，我們創建模塊exception.py。在其中通過依賴注入的方式設置全局異常。

# core/exception.py

...
def register_exception(app: FastAPI):
    """
    異常捕捉
    """

    @app.exception_handler(CustomException)
    asyncdef custom_exception_handler(request: Request, exc: CustomException):
        """
        自定義異常
        """
        _logger.error(
            f"Path: {request.url.path}\n"
            f"Method: {request.method}\n"
            f"Client: {request.client.host if request.client else 'unknown'}\n"
            f"Exception: {type(exc).__name__}",
            exc_info=(type(exc), exc, exc.__traceback__)
        )

        return JSONResponse(
            status_code=exc.status_code,
            cnotallow={"message": exc.msg, "code": exc.code},
        )

    @app.exception_handler(HTTPException)
    asyncdef unicorn_exception_handler(request: Request, exc: HTTPException):
        """
        重寫HTTPException異常處理器
        """
        _logger.error(
            f"Path: {request.url.path}\n"
            f"Method: {request.method}\n"
            f"Client: {request.client.host if request.client else 'unknown'}\n"
            f"Exception: {type(exc).__name__}",
            exc_info=(type(exc), exc, exc.__traceback__)
        )
        return JSONResponse(
            status_code=exc.status_code,
            cnotallow={
                "code": exc.status_code,
                "message": exc.detail,
            }
        )
      ...

代碼倉庫 https://github.com/pyzxs/zadmin。

在__init__.py中設置暴露函數：

from core.exception import register_exception
from core.logger import get_logger

__all__ = [get_logger,register_exception]

在main.py的啟動腳本中設置：

# main.py

from core import get_logger, register_exception

def create_app():
    ...
    register_exception(app)
    ...

測試異常處理：

@app.get("/{form_id}")
    async def root(form_id: int = Path(..., gt=0)):
        # raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
        # raise CustomException("CustomException")
        # raise ValueError("ValueError")
        
        return {"message": form_id}

    return app

運行測試, 該路由地址中定義{form_id:int}是整形，如果輸入字符串就會拋出RequestValidationError,在error.log中會記錄異常的追溯。

效果如下：

現在項目的日志模塊就定義好了。你可以在調試的代碼中通過get_logger().debug()后info()調試代碼了。同時系統exception報錯也會記錄到error.log中。幫助你盡快定位異常。

責任編輯：趙寧寧來源：程序員老朱