Skip to content

64. FastAPI 日志功能 ​

日志是应用运行时的"值班记录"。请求成功了要留痕,出错了要能追溯到是哪个模块、哪一步、发生了什么。本篇用一个完整项目,从"全局日志"到"分级存储"再到"自动清理",一步步搭起来。

案例目标与目录结构 ​

最终项目结构:

text
project/
├── main.py                    # 入口 + lifespan 初始化日志
├── deps.py                    # 依赖层,也使用日志
├── core/
│   ├── __init__.py
│   └── logger.py              # 全局日志配置
├── routers/
│   ├── __init__.py
│   └── items.py               # 路由层
├── services/
│   ├── __init__.py
│   └── item_service.py        # 服务层
└── logs/                      # 运行后自动生成
    ├── app.log                # INFO 及以上日志
    └── error.log              # 只装 ERROR 及以上

其中 core/__init__.py、routers/__init__.py、services/__init__.py 都是空文件,用来把目录标记为包。

一、全局日志配置 ​

日志配置模块 core/logger.py ​

把日志的格式、输出位置、级别集中在这里,整个项目只配置一次。

python
import logging
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path

LOG_DIR = Path(__file__).resolve().parent.parent / "logs"
_initialized = False


def setup_logging() -> None:
    global _initialized
    if _initialized:
        return
    _initialized = True

    LOG_DIR.mkdir(exist_ok=True)

    fmt = logging.Formatter(
        "%(asctime)s [%(levelname)s] %(name)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    # INFO 及以上的日志(含 WARNING、ERROR)都进 app.log
    info_handler = TimedRotatingFileHandler(
        LOG_DIR / "app.log",
        when="midnight",
        interval=1,
        backupCount=7,
        encoding="utf-8",
    )
    info_handler.setLevel(logging.INFO)
    info_handler.setFormatter(fmt)

    # 只有 ERROR 及以上的日志进 error.log
    error_handler = TimedRotatingFileHandler(
        LOG_DIR / "error.log",
        when="midnight",
        interval=1,
        backupCount=7,
        encoding="utf-8",
    )
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(fmt)

    root = logging.getLogger()
    root.setLevel(logging.INFO)
    root.addHandler(info_handler)
    root.addHandler(error_handler)


def get_logger(name: str | None = None) -> logging.Logger:
    return logging.getLogger(name or "app")

要点:

  • setup_logging() 把两个 handler 挂到根 logger 上,并设置全局级别为 INFO。
  • _initialized 防止重复初始化(--reload 重启进程时避免同一进程内重复加 handler)。
  • get_logger(name) 只是 logging.getLogger 的薄封装,方便统一入口。

应用启动时初始化 ​

用 lifespan 在应用启动时调用 setup_logging(),而不是在模块导入时调用。这样初始化时机可控,也避免 import 副作用。

main.py:

python
from contextlib import asynccontextmanager

from fastapi import FastAPI

from core.logger import get_logger, setup_logging
from routers import items

logger = get_logger("app")


@asynccontextmanager
async def lifespan(app: FastAPI):
    setup_logging()
    logger.info("应用启动,日志初始化完成")
    yield
    logger.info("应用关闭")


app = FastAPI(lifespan=lifespan)
app.include_router(items.router)

跨模块使用同一个日志实例 ​

路由、服务、依赖各自拿到 logger 后直接调用即可。它们最终都通过根 logger 上的同一套 handler 写进文件,%(name)s 又能区分是哪个模块打的。

依赖层 deps.py:

python
from core.logger import get_logger

logger = get_logger(__name__)


def log_context():
    logger.info("依赖执行:记录本次请求上下文")
    return {"trace_id": "abc123"}

服务层 services/item_service.py:

python
from core.logger import get_logger

logger = get_logger(__name__)

ITEMS = {
    1: {"id": 1, "name": "键盘"},
    2: {"id": 2, "name": "鼠标"},
}


def get_item(item_id: int):
    item = ITEMS.get(item_id)
    if item is None:
        logger.error("商品不存在: item_id=%s", item_id)
        return None
    logger.info("查询商品成功: item_id=%s", item_id)
    return item


def create_item(name: str):
    logger.info("创建商品: %s", name)
    return {"id": 3, "name": name}

路由层 routers/items.py:

python
from fastapi import APIRouter, Depends, HTTPException

from core.logger import get_logger
from deps import log_context
from services import item_service

logger = get_logger(__name__)
router = APIRouter(prefix="/items", tags=["商品"])


@router.get("/{item_id}")
def read_item(item_id: int, ctx: dict = Depends(log_context)):
    logger.info("收到查询请求: item_id=%s", item_id)
    item = item_service.get_item(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="商品不存在")
    return item


@router.post("")
def create_item(name: str):
    logger.info("收到创建请求: name=%s", name)
    return item_service.create_item(name)

三层都调用了 get_logger(__name__),得到的是以各自模块命名的 logger,但它们都向根 logger 传递日志,共用同一套 handler 和格式。这样"全局一处配置,处处可用",又不丢失"这条日志来自哪个模块"的信息。

二、日志分级存储 ​

两个 handler 各管各的:

文件级别要求装什么
app.logINFO 及以上正常日志 + 警告 + 错误(全量)
error.logERROR 及以上只有错误(是 app.log 的子集)

所以一条 ERROR 会同时进 app.log 和 error.log;一条 INFO 只进 app.log。

启动服务:

text
uv run uvicorn main:app --reload

触发正常请求 ​

text
curl.exe http://127.0.0.1:8000/items/1

logs/app.log 新增:

text
2026-10-25 10:00:01 [INFO] app - 应用启动,日志初始化完成
2026-10-25 10:00:03 [INFO] deps - 依赖执行:记录本次请求上下文
2026-10-25 10:00:03 [INFO] routers.items - 收到查询请求: item_id=1
2026-10-25 10:00:03 [INFO] services.item_service - 查询商品成功: item_id=1

logs/error.log 没有新增(没有 ERROR)。

触发异常请求 ​

text
curl.exe http://127.0.0.1:8000/items/999

logs/app.log 新增:

text
2026-10-25 10:00:10 [INFO] deps - 依赖执行:记录本次请求上下文
2026-10-25 10:00:10 [INFO] routers.items - 收到查询请求: item_id=999
2026-10-25 10:00:10 [ERROR] services.item_service - 商品不存在: item_id=999

logs/error.log 新增(只有错误):

text
2026-10-25 10:00:10 [ERROR] services.item_service - 商品不存在: item_id=999

对比可见:正常请求只写 app.log;异常请求的 ERROR 同时进了 app.log 和 error.log。日常排查翻 app.log 看完整过程,报警监控只盯 error.log 看有没有错误。

三、日志自动清理 ​

TimedRotatingFileHandler 既负责按时间轮转,也负责删除过期文件。上面 logger.py 里已经配好:

python
TimedRotatingFileHandler(
    LOG_DIR / "app.log",
    when="midnight",   # 每天午夜轮转一次
    interval=1,        # 间隔 1 天
    backupCount=7,     # 最多保留 7 份历史文件
    encoding="utf-8",
)

含义:

  • when="midnight" + interval=1:每天午夜把当前 app.log 改名为带日期的历史文件(如 app.log.2026-10-24),再开一个新的 app.log。
  • backupCount=7:轮转后最多保留 7 份历史文件,最早的那份会被自动删除。

注意:backupCount=7 保留的是"7 份历史 + 1 份当前",即目录里最多 8 个 app.log* 文件。如果想让目录里严格只保留 7 个文件,就把 backupCount 设为 6。

快速验证轮转与清理 ​

等 7 天才能看到效果不现实,可以临时把轮转周期改小来观察机制。写一个 demo_rotation.py:

python
import logging
import time
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path

log_dir = Path("logs")
log_dir.mkdir(exist_ok=True)

handler = TimedRotatingFileHandler(
    log_dir / "demo.log",
    when="S",       # 按秒轮转,方便快速观察
    interval=2,     # 每 2 秒轮转一次
    backupCount=2,  # 只保留 2 份历史文件
    encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))

logger = logging.getLogger("demo")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

for i in range(1, 9):
    logger.info("第 %s 条日志", i)
    time.sleep(1.1)

运行 python demo_rotation.py(约 9 秒),结束后 logs/ 目录里只会看到 demo.log 和最多 2 个历史备份,例如:

text
logs/
├── demo.log                          # 当前文件
├── demo.log.2026-10-25_10-00-04      # 历史备份 1
└── demo.log.2026-10-25_10-00-06      # 历史备份 2

随着继续写入,新的轮转会生成新的历史文件,同时把最早的备份删掉,始终保持"当前文件 + 最多 2 份历史"。把 when 换回 midnight、backupCount 换回 7,就是生产上"保留最近 7 天"的效果。

小结与练习 ​

  • 用 setup_logging() 在 lifespan 里统一初始化,全局只配一次;模块通过 get_logger(__name__) 复用同一套 handler。
  • 用两个 handler 分级:app.log 记 INFO 及以上,error.log 只记 ERROR。
  • 用 TimedRotatingFileHandler 按天轮转 + backupCount 控制保留数量,自动清理过期日志。

练习:给服务层再加一个会抛异常的分支,用全局异常处理器 logger.error 记录完整堆栈,验证它同时出现在 app.log 和 error.log;再把 backupCount 改成 3,运行快速轮转脚本观察保留文件数量的变化。