logging 完全指南

官方文档：https://docs.python.org/zh-cn/3/library/logging.html
适用版本：Python 3.12（2026-05-07 核实）

相关文档：FastAPI完全指南 Celery完全指南 asyncio异步编程完全指南

1. 基础概念

日志系统的四个核心组件

日志级别

Logger 层级关系

Logger 名称以 . 分隔形成树形结构，子 Logger 会将日志向上传播给父 Logger：

root logger
  └── myapp               (logging.getLogger("myapp"))
        ├── myapp.web     (logging.getLogger("myapp.web"))
        └── myapp.db      (logging.getLogger("myapp.db"))

2. 快速开始

basicConfig — 简单配置

import logging

# 基础配置（只能调用一次，在 root logger 尚未配置时有效）
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logging.debug("调试信息")
logging.info("运行正常")
logging.warning("注意：磁盘空间不足")
logging.error("发生错误")
logging.critical("严重错误，服务停止")

按模块获取 Logger（推荐）

# 每个模块独立获取 logger，便于追踪日志来源
import logging

logger = logging.getLogger(__name__)  # __name__ = 模块的完整路径，如 myapp.web.views

def create_user(name: str):
    logger.info("创建用户: %s", name)   # 使用 % 格式化，比 f-string 高效（惰性求值）
    try:
        user = db.create(name)
        logger.debug("用户创建成功，id=%d", user.id)
        return user
    except Exception as e:
        logger.error("创建用户失败: %s", name, exc_info=True)  # exc_info=True 附加堆栈
        raise

3. Handler — 输出目标

StreamHandler — 输出到控制台

import logging

handler = logging.StreamHandler()  # 默认输出到 sys.stderr
handler.setLevel(logging.DEBUG)

FileHandler — 输出到文件

handler = logging.FileHandler("app.log", encoding="utf-8")
handler.setLevel(logging.INFO)

RotatingFileHandler — 按文件大小轮转

from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    "app.log",
    maxBytes=10 * 1024 * 1024,  # 单个文件最大 10MB
    backupCount=5,              # 保留最近 5 个备份（app.log.1 ~ app.log.5）
    encoding="utf-8",
)

TimedRotatingFileHandler — 按时间轮转

from logging.handlers import TimedRotatingFileHandler

handler = TimedRotatingFileHandler(
    "app.log",
    when="midnight",   # 每天午夜轮转（可选：S/M/H/D/W0-W6/midnight）
    interval=1,
    backupCount=30,    # 保留最近 30 天
    encoding="utf-8",
)
handler.suffix = "%Y-%m-%d"  # 备份文件名后缀格式

4. Formatter — 日志格式

常用格式字段

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

5. 标准配置模式

代码配置（适合小型项目）

import logging
from logging.handlers import RotatingFileHandler

def setup_logging(level: str = "INFO"):
    root_logger = logging.getLogger()
    root_logger.setLevel(logging.DEBUG)  # root logger 设最低级别，由 handler 各自过滤

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

    # 控制台：INFO 及以上
    console = logging.StreamHandler()
    console.setLevel(getattr(logging, level.upper()))
    console.setFormatter(formatter)

    # 文件：DEBUG 及以上，按大小轮转
    file_handler = RotatingFileHandler(
        "logs/app.log",
        maxBytes=10 * 1024 * 1024,
        backupCount=10,
        encoding="utf-8",
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

    # 错误单独记录到一个文件
    error_handler = RotatingFileHandler(
        "logs/error.log",
        maxBytes=5 * 1024 * 1024,
        backupCount=5,
        encoding="utf-8",
    )
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(formatter)

    root_logger.addHandler(console)
    root_logger.addHandler(file_handler)
    root_logger.addHandler(error_handler)

字典配置（推荐用于生产）

import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,  # 不禁用已有的 logger
    "formatters": {
        "standard": {
            "format": "%(asctime)s [%(levelname)-8s] %(name)s: %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
        "detailed": {
            "format": "%(asctime)s [%(levelname)-8s] %(name)s:%(lineno)d %(funcName)s() - %(message)s",
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "standard",
            "stream": "ext://sys.stdout",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "DEBUG",
            "formatter": "detailed",
            "filename": "logs/app.log",
            "maxBytes": 10485760,  # 10MB
            "backupCount": 10,
            "encoding": "utf-8",
        },
        "error_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "ERROR",
            "formatter": "detailed",
            "filename": "logs/error.log",
            "maxBytes": 5242880,
            "backupCount": 5,
            "encoding": "utf-8",
        },
    },
    "loggers": {
        # 第三方库静默（只输出 WARNING 及以上）
        "httpx": {"level": "WARNING"},
        "sqlalchemy.engine": {"level": "WARNING"},
        "uvicorn.access": {"level": "WARNING"},
    },
    "root": {
        "level": "DEBUG",
        "handlers": ["console", "file", "error_file"],
    },
}

logging.config.dictConfig(LOGGING_CONFIG)

6. 结构化日志（生产推荐）

标准 logging 输出纯文本，难以被日志系统（ELK/Loki）解析。推荐使用 structlog 或 loguru 输出 JSON 格式。

loguru — 简洁易用

pip install loguru

from loguru import logger
import sys

# 移除默认 handler，重新配置
logger.remove()

# 控制台（彩色输出）
logger.add(
    sys.stdout,
    level="INFO",
    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
)

# 文件（JSON 格式，按天轮转）
logger.add(
    "logs/app_{time:YYYY-MM-DD}.log",
    level="DEBUG",
    rotation="00:00",       # 每天午夜轮转
    retention="30 days",    # 保留 30 天
    compression="gz",       # 自动压缩旧文件
    serialize=True,         # 输出 JSON 格式
    encoding="utf-8",
)

logger.info("服务启动", port=8000)
logger.error("请求失败", user_id=42, url="/api/users")

与标准 logging 兼容

# 将标准库 logging 的日志路由到 loguru
import logging
from loguru import logger

class InterceptHandler(logging.Handler):
    def emit(self, record: logging.LogRecord):
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1
        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

structlog — 与标准库深度集成

pip install structlog

import structlog
import logging

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),  # 输出 JSON
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    logger_factory=structlog.PrintLoggerFactory(),
)

log = structlog.get_logger()
log.info("用户登录", user_id=42, ip="192.168.1.1")
# 输出：{"event": "用户登录", "user_id": 42, "ip": "192.168.1.1", "level": "info", "timestamp": "..."}

7. 与 FastAPI 集成

# src/core/logging.py
import logging
import logging.config
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import time

LOGGING_CONFIG = { ... }  # 同上字典配置

def setup_logging():
    logging.config.dictConfig(LOGGING_CONFIG)

@asynccontextmanager
async def lifespan(app: FastAPI):
    setup_logging()
    logger = logging.getLogger("app")
    logger.info("服务启动")
    yield
    logger.info("服务关闭")

app = FastAPI(lifespan=lifespan)

# 请求日志中间件
@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger = logging.getLogger("app.access")
    start = time.perf_counter()
    response = await call_next(request)
    duration = (time.perf_counter() - start) * 1000
    logger.info(
        "%s %s %d %.1fms",
        request.method,
        request.url.path,
        response.status_code,
        duration,
    )
    return response

8. 常用代码段

捕获异常并记录堆栈

import logging

logger = logging.getLogger(__name__)

try:
    risky_operation()
except Exception:
    logger.exception("操作失败")  # 等同于 logger.error(..., exc_info=True)

临时提升日志级别（调试用）

import logging
from contextlib import contextmanager

@contextmanager
def debug_logging(logger_name: str):
    log = logging.getLogger(logger_name)
    old_level = log.level
    log.setLevel(logging.DEBUG)
    try:
        yield
    finally:
        log.setLevel(old_level)

with debug_logging("sqlalchemy.engine"):
    result = db.execute(...)  # 此段会打印 SQL 语句

给日志添加请求 ID（contextvars）

import logging
import uuid
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True

# 在 Formatter 中使用 %(request_id)s
# 在中间件中设置
async def log_requests(request, call_next):
    token = request_id_var.set(str(uuid.uuid4())[:8])
    try:
        return await call_next(request)
    finally:
        request_id_var.reset(token)

9. 最佳实践

用 %s 格式化而非 f-string

# 推荐：惰性求值，level 不满足时不做字符串拼接
logger.debug("用户数据：%s", user_data)

# 不推荐：无论级别是否满足，都会执行 f-string 求值
logger.debug(f"用户数据：{user_data}")

第三方库日志降级

# 防止 httpx、sqlalchemy 等库的 DEBUG 日志淹没业务日志
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)

不要在库代码中配置 logging

库代码只应获取 logger 并记录，不应调用 basicConfig 或添加 Handler，配置权交给使用者：

# 库代码（正确）
import logging
logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())  # 防止"No handlers could be found"警告

最佳实践

每个模块用 logging.getLogger(__name__) 而不是 root logger：直接用 logging.info() 等调用 root logger，无法区分日志来源模块，也无法对特定模块单独设置日志级别。__name__ 自动形成层级（myapp.utils、myapp.models），可按层级配置。

# 正确：每个模块独立 logger
import logging
logger = logging.getLogger(__name__)
logger.info("module-level log")

# 低效：无法区分来源
import logging
logging.info("can't tell where this comes from")

日志配置用 dictConfig，不用 basicConfig：basicConfig 只适合脚本快速调试，功能有限。dictConfig 支持多 handler、多 formatter、按模块设置级别，是生产环境标准做法。

import logging.config
logging.config.dictConfig({
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {"json": {"()": "pythonjsonlogger.jsonlogger.JsonFormatter"}},
    "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "json"}},
    "root": {"level": "INFO", "handlers": ["console"]},
})

生产环境输出结构化 JSON 日志：纯文本日志难以被 ELK / Loki 等日志系统解析和检索，JSON 格式天然支持字段查询和聚合。

pip install python-json-logger

formatter = pythonjsonlogger.jsonlogger.JsonFormatter(
    "%(asctime)s %(name)s %(levelname)s %(message)s"
)

库代码只添加 NullHandler，不配置 logging：库不应该为用户配置 logging，否则会影响用户应用的日志行为。只添加 NullHandler 作为占位，让用户决定如何处理库的日志。

# 库代码
logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())

常见陷阱

陷阱：basicConfig 调用后静默失效

现象： 调用 basicConfig(level=DEBUG) 后日志级别没有变化，INFO 以下的日志仍然不显示。

原因： 如果在 basicConfig 之前已经有代码（包括第三方库的 import）调用了任何 logging.* 函数，root logger 就已被初始化，之后的 basicConfig 调用会被静默忽略。

解决： 在程序入口最先配置 logging，或使用 force=True（Python 3.8+）强制重新配置。

# 错误：在 basicConfig 之前已触发初始化
import logging
logging.warning("early log")       # root logger 已初始化
logging.basicConfig(level=logging.DEBUG)  # 无效！

# 正确
logging.basicConfig(level=logging.DEBUG, force=True)  # 强制覆盖

陷阱：dictConfig 的 disable_existing_loggers 导致第三方库静默

现象： 配置 dictConfig 后，第三方库（如 SQLAlchemy、httpx）的日志全部消失。

原因： disable_existing_loggers 默认值为 True，会在 dictConfig 执行时禁用所有在此之前已创建的 Logger（import 库时就已创建）。

解决： 始终将 disable_existing_loggers 设为 False。

logging.config.dictConfig({
    "version": 1,
    "disable_existing_loggers": False,   # 必须设为 False
    ...
})

陷阱：多进程写同一日志文件造成日志混乱

现象： gunicorn 多 worker 模式下，日志文件中出现行混合、截断、乱序等问题。

原因： FileHandler 不是进程安全的，多进程同时写同一文件会造成写入竞争。

解决： 方案一：用 logging.handlers.QueueHandler + QueueListener 将日志集中到单一进程写入；方案二：用 SocketHandler 发送到独立日志收集进程；方案三：每个 worker 写独立文件，再用 logrotate 合并。

from logging.handlers import QueueHandler, QueueListener
import queue

log_queue = queue.Queue()
queue_handler = QueueHandler(log_queue)
file_handler = logging.FileHandler("app.log")
listener = QueueListener(log_queue, file_handler, respect_handler_level=True)
listener.start()

参见

• FastAPI完全指南
• Celery完全指南
• asyncio异步编程完全指南
• loguru完全指南