Python 中的日志记录：生产环境配置

1. 概述

print() 在编写小脚本时就足够了。但在生产环境就不同了：你需要按级别过滤日志、将日志汇总到中央日志系统、追踪跨多条日志行的单个请求、并且日志写入不能使应用程序变慢。

本文介绍了如何为生产环境配置 Python 的标准 logging 模块。完成后，你将建立一个包含模块级 logger、dictConfig、JSON 格式日志、自动附加 request_id、使用 QueueHandler 进行异步日志写入以及使用 pytest 进行日志测试的配置。所有示例使用 Python 3.12 并仅需标准库。测试部分需要额外的 pytest 库。

2. 为什么 print 和 basicConfig 不够

看一个按“快速完成”风格编写的订单处理服务：

import logging

logging.basicConfig(level=logging.INFO)

def place_order(order_id: str, amount: float) -> None:
    print(f"Placing order {order_id}")
    logging.info(f"Order {order_id} amount={amount}")

place_order("ORD-1001", 250_000)

Placing order ORD-1001
INFO:root:Order ORD-1001 amount=250000

这段代码有三个问题：

• print 没有日志级别、没有时间戳，并且无法通过配置关闭。
• logging.info(...) 写入 根 logger。你不知道日志来自哪个模块，也无法为不同部分单独调整级别。
• basicConfig 只在第一次调用时有效，且仅当根 logger 还没有 handler 时。如果任何库在你之前调用了它，你的配置将被静默忽略。

3. 使用 getLogger(name) 按模块创建 Logger

第一条规则：每个模块创建自己的 logger，按 __name__ 命名。

# shop/orders.py
import logging

logger = logging.getLogger(__name__)  # logger 名称: "shop.orders"

class OrderService:
    def place_order(self, order_id: str, quantity: int) -> None:
        logger.info("Placing order %s with quantity %d", order_id, quantity)
        if quantity > 50:
            logger.warning("Large order %s: quantity=%d", order_id, quantity)

logger 名称形成一个按句点划分的层级树。shop.orders 是 shop 的子级，shop 是 root 的子级。一条记录从子 logger 传播到父 logger（称为 propagation）。因此，你只需在 root 或 shop 处附加 handler，然后为每个分支单独调整级别。例如，你可以将 shop.payments 设置为 DEBUG，其余部分保持为 INFO。

你也会注意到参数传递的方式：使用 %s 和参数，而不是 f-string。当级别被关闭时，字符串不会被格式化，所以你不会为永远不会被打印的日志付出代价。此外，日志聚合工具可以按相同的消息模板对行进行分组。

注意： 在库（你为他人发布的包）中，不要配置 handler。只调用 getLogger(__name__)，让应用程序来决定。如果需要，添加 logging.getLogger("your_lib").addHandler(logging.NullHandler())。

4. 使用 dictConfig 进行集中配置

在生产环境中，你应该将所有配置集中在一个地方，在应用程序启动时只调用一次。logging.config.dictConfig 适合用于此目的。

# shop/logging_config.py
import logging.config
import os

def setup_logging() -> None:
    level = os.getenv("LOG_LEVEL", "INFO")
    logging.config.dictConfig({
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "plain": {
                "format": "%(asctime)s %(levelname)-8s %(name)s - %(message)s",
            },
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
                "formatter": "plain",
            },
        },
        "loggers": {
            "shop": {"level": level},
            "urllib3": {"level": "WARNING"},
        },
        "root": {"level": "WARNING", "handlers": ["console"]},
    })

配置中的关键点：

• "disable_existing_loggers": False 在实践中是必需的。默认值 True 会禁用在 dictConfig 运行之前创建的所有 logger，包括你导入的模块开头的 logger。结果是日志会消失，没有任何错误。
• Handler 只附加到 root。Logger shop 只设置级别，记录会传播到 root 进行输出。
• Root 设置为 WARNING 是为了阻止来自第三方库的嘈杂 INFO 日志。你的代码在 shop 分支中仍然会以 INFO 级别记录日志。
• 级别从环境变量 LOG_LEVEL 读取。在调查问题时，你可以启用 DEBUG 而无需修改代码。

在入口点调用函数：

# main.py
from shop.logging_config import setup_logging
from shop.orders import OrderService

setup_logging()
OrderService().place_order("ORD-1001", quantity=80)

2026-10-18 09:15:02,114 INFO     shop.orders - Placing order ORD-1001 with quantity 80
2026-10-18 09:15:02,114 WARNING  shop.orders - Large order ORD-1001: quantity=80

在容器中运行时，写入 stdout 是正确的选择。Docker 或 Kubernetes 将从这个流中收集日志，所以你不需要自己管理文件和文件轮换。

5. 结构化 JSON 日志

像 Loki、Elasticsearch 或 CloudWatch 这样的日志聚合系统在 JSON 上的搜索效果比自由文本好得多。你可以自己编写 formatter，无需外部库：

# shop/json_formatter.py
import json
import logging
from datetime import datetime, timezone

_RESERVED = set(vars(logging.LogRecord("", 0, "", 0, "", None, None))) | {"message", "asctime"}

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        for key, value in record.__dict__.items():
            if key not in _RESERVED and not key.startswith("_"):
                payload[key] = value
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False, default=str)

工作方式：

• _RESERVED 包含 LogRecord 的默认属性，例如 pathname、lineno、args。
• 所有其他属性都被视为自定义字段并包含在 JSON 中。这些字段通常来自 extra= 参数或 filter（参见第 6 节）。
• default=str 帮助 json.dumps 处理 Decimal 或 UUID 等类型而不出错。

在 dictConfig 中使用特殊键 "()" 声明 formatter：

"formatters": {
    "json": {"()": "shop.json_formatter.JsonFormatter"},
},
"handlers": {
    "console": {
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stdout",
        "formatter": "json",
    },
},

使用 extra 添加业务字段：

logger.info("Payment captured", extra={"order_id": "ORD-1001", "amount": 250000})

{"ts": "2026-10-18T02:15:02.114+00:00", "level": "INFO", "logger": "shop.payments", "message": "Payment captured", "order_id": "ORD-1001", "amount": 250000}

注意： extra 中的 key 不能与 LogRecord 的现有属性名相同。例如，extra={"name": "..."} 或 extra={"message": "..."} 将抛出 KeyError: "Attempt to overwrite 'name' in LogRecord"。

6. 使用 contextvars 自动附加 request_id

向每个日志调用传递 extra={"request_id": ...} 既冗长又容易忘记。更好的方法是将 request_id 保存到 ContextVar，然后使用 Filter 将其附加到每条记录。

# shop/request_context.py
import logging
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  # 不阻止任何记录，只是补充数据

ContextVar 对 thread 和 asyncio 都是安全的。每个 task 或 request 都持有自己的值，所以不会发生一个 request 的 ID 被另一个 request “借用” 的情况。

将 filter 附加到 handler，而不是 logger。附加到 logger 的 filter 只对直接由该 logger 创建的记录运行，不对从子 logger 传播上来的记录运行。

"filters": {
    "request_id": {"()": "shop.request_context.RequestIdFilter"},
},
"handlers": {
    "console": {
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stdout",
        "formatter": "json",
        "filters": ["request_id"],
    },
},

在每个请求的开始设置值，例如在中间件中：

import uuid
from shop.request_context import request_id_var

def handle_request(order_id: str) -> None:
    token = request_id_var.set(uuid.uuid4().hex[:12])
    try:
        OrderService().place_order(order_id, quantity=3)
    finally:
        request_id_var.reset(token)

{"ts": "...", "level": "INFO", "logger": "shop.orders", "message": "Placing order ORD-1002 with quantity 3", "request_id": "9f2c1ab07d3e"}

现在你只需按 request_id 过滤就能看到一个请求的整个旅程。

7. 正确记录异常日志

logger = logging.getLogger("shop.payments")

class PaymentService:
    def charge(self, order_id: str, amount: int) -> bool:
        try:
            raise ConnectionError("gateway timeout")
        except ConnectionError:
            logger.exception("Charge failed for order %s", order_id)
            return False

logger.exception(...) 以 ERROR 级别记录并自动附加堆栈跟踪。你只能在 except 块内调用它。如果要在不同级别记录但仍保持堆栈跟踪，请使用 logger.warning("...", exc_info=True)。

两个常见的错误：

• logger.error(f"Charge failed: {e}") 会失去堆栈跟踪。调查时，你只会看到一行 “gateway timeout”，不知道它发生在哪里。
• 在每个层级都既记录又重新抛出异常。同一个异常会在日志中出现五次。在异常处理层记录，即决定不再抛出异常的地方。

8. 不让日志记录阻塞应用：QueueHandler

StreamHandler 写入 stdout 通常足够快。但通过网络发送（syslog、HTTP、SMTP）或写入磁盘的 handler 会阻塞处理请求的线程。为了避免这种情况，将记录放入队列中，由后台线程写入实际目标：

import atexit
import logging
import logging.handlers
import queue

def setup_async_logging(target: logging.Handler) -> None:
    log_queue: queue.Queue = queue.Queue(-1)
    queue_handler = logging.handlers.QueueHandler(log_queue)

    listener = logging.handlers.QueueListener(
        log_queue, target, respect_handler_level=True
    )
    listener.start()
    atexit.register(listener.stop)  # 关闭时刷新剩余部分

    root = logging.getLogger()
    root.handlers = [queue_handler]

处理流程如下：

• 应用线程只是将记录 put 到队列中。此操作几乎是即时的。
• QueueListener 运行在单独的线程上，从队列中取出记录并传递给 target，例如 SysLogHandler。
• atexit.register(listener.stop) 确保最后的日志在进程退出前被写入。

注意： 像 RequestIdFilter 这样的 Filter 必须附加到 QueueHandler，而不是目标 handler。目标 handler 运行在 listener 的线程上，而该线程中的 ContextVar 不会携带原始请求的值。

9. 使用 pytest caplog 测试日志

业务警告日志，如大量订单，是值得测试的行为。将 pytest 添加到开发依赖：

pip install pytest

# tests/test_orders.py
import logging
from shop.orders import OrderService

def test_large_order_logs_warning(caplog):
    with caplog.at_level(logging.INFO, logger="shop.orders"):
        OrderService().place_order("ORD-2001", quantity=80)

    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
    assert len(warnings) == 1
    assert warnings[0].name == "shop.orders"
    assert warnings[0].getMessage() == "Large order ORD-2001: quantity=80"

def test_small_order_does_not_warn(caplog):
    with caplog.at_level(logging.INFO, logger="shop.orders"):
        OrderService().place_order("ORD-2002", quantity=2)

    assert all(r.levelno < logging.WARNING for r in caplog.records)

tests/test_orders.py ..                                  [100%]
2 passed in 0.03s

caplog.at_level 在 with 块的范围内暂时降低 logger shop.orders 的级别。caplog.records 包含原始 LogRecord，所以你可以检查级别、logger 名称和任何 extra 字段。

注意： caplog 将 handler 附加到根 logger。如果你的配置为 logger shop 设置了 "propagate": False，测试将无法捕获任何记录。

10. 常见错误和不应该做什么

• 在模块级别配置日志（在导入时调用 basicConfig 或 dictConfig）。仅在入口点配置一次。
• Handler 被附加了两次，例如同时附加到 shop 和 root 而 propagate 仍为 True。每行日志会被打印两次。你应该只在一个级别附加 handler。
• 记录敏感数据：密码、token、卡号。不要记录整个请求对象。如果必须，请编写一个 Filter 来隐藏（mask）这些字段。
• 在多进程应用中使用 RotatingFileHandler（使用多个 worker 的 Gunicorn）。多个进程轮换同一个文件会导致日志丢失。应该写入 stdout，或让每个 worker 通过 QueueHandler 推送到单个写入进程。
• 在热循环中过度记录 INFO 级别的日志。存储成本和噪音会迅速增加。这样的日志应该设置为 DEBUG 级别。
• 在本地开发时使用 JSON。 JSON 很难用眼睛读取，所以应该根据环境变量选择 formatter：dev 使用 plain，production 使用 json。

11. 结论

一个良好的生产日志配置并不复杂，只需一致。在所有模块中使用 getLogger(__name__)。将配置集中在一次 dictConfig 调用中，设置 disable_existing_loggers: False。将 JSON 写入 stdout，通过 ContextVar 和 Filter 附加 request_id，并使用 logger.exception 保留堆栈跟踪。当目标 handler 较慢时，切换到 QueueHandler 和 QueueListener。

最后，将日志视为系统行为的一部分，并使用 caplog 进行测试。当凌晨 2 点出现问题时，这些结构化日志行将帮助你更快地找到原因。