Python 日志与调试

日志和调试不是“出问题以后才用的技巧”，而是项目从一开始就要设计好的可观测能力。零基础阶段可以用 print() 看变量，但商业项目必须使用 logging、request_id、异常堆栈、耗时日志、敏感信息脱敏和排查流程。

如果没有日志，线上报错时只能靠猜；如果日志没有链路 ID，分布式调用时只能在一堆日志里人工搜索；如果日志记录了密码、Token、患者隐私，排查问题又会变成安全事故。

学习目标

学完本页你应该能回答：

1. print() 和 logging 的区别是什么。
2. Logger、Handler、Formatter、Filter、Level 分别负责什么。
3. 一条日志从代码到文件或日志平台经历哪些步骤。
4. 为什么要记录 request_id、耗时、业务 ID 和错误码。
5. 如何记录异常堆栈，为什么不能只打印异常信息。
6. 如何做日志切割、结构化日志、敏感信息脱敏。
7. 本地 Debug 和线上排查分别怎么做。
8. 接口慢、任务失败、外部服务失败时如何顺着日志定位。

为什么不能只用 print

print() 是把内容输出到标准输出，适合临时练习；logging 是日志系统，适合工程项目。

对比	print()	logging
日志级别	没有	DEBUG、INFO、WARNING、ERROR
输出目标	通常是控制台	控制台、文件、日志平台、队列
格式	手动拼字符串	Formatter 统一格式
异常堆栈	不方便	logger.exception() 自动记录
过滤能力	没有	可按级别、模块、Filter 过滤
线上检索	不适合	适合日志平台检索
长期维护	差	好

练习代码可以 print()，项目代码应该用 logging。否则上线后会遇到这些问题：

1. 不知道日志来自哪个模块。
2. 不知道日志级别，错误和普通信息混在一起。
3. 无法按请求 ID 串联链路。
4. 无法保留异常堆栈。
5. 无法控制日志输出位置和保留策略。

logging 核心组件

Python 标准库 logging 的核心组件：

组件	作用	类比
Logger	代码中调用的日志对象	说话的人
Level	日志级别	事情严重程度
Handler	输出目标	话筒或收件箱
Formatter	日志格式	统一表格格式
Filter	过滤或补充日志字段	安检员
LogRecord	一条日志事件的数据对象	一张记录单

日志流转过程：

flowchart TD
    A["业务代码 logger.info"] --> B["创建 LogRecord"]
    B --> C{"Logger 级别是否允许"}
    C -- "否" --> D["丢弃日志"]
    C -- "是" --> E["进入 Handler"]
    E --> F{"Handler 级别是否允许"}
    F -- "否" --> D
    F -- "是" --> G["Filter 过滤或补充字段"]
    G --> H["Formatter 格式化"]
    H --> I["输出到控制台 文件 日志平台"]

很多人以为 logger.info() 只是打印字符串，实际上它会创建一条 LogRecord，里面包含日志级别、时间、模块名、行号、线程、进程、异常信息等字段。

最小示例

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)

logger = logging.getLogger(__name__)

logger.debug("调试信息")
logger.info("程序启动")
logger.warning("配置缺失，使用默认值")
logger.error("任务执行失败")

因为当前级别是 INFO，所以 DEBUG 不会输出。日志级别从低到高是：

DEBUG < INFO < WARNING < ERROR < CRITICAL

日志级别怎么选

级别	含义	适合记录
DEBUG	调试细节	局部变量、分支、SQL 参数、开发排查
INFO	正常关键流程	服务启动、请求完成、任务开始结束
WARNING	可恢复异常	重试、配置缺省、外部接口偶发失败
ERROR	当前操作失败	请求失败、任务失败、数据库异常
CRITICAL	服务级严重故障	核心依赖不可用、服务无法继续

不要把所有日志都打成 ERROR。如果正常重试也打 ERROR，报警会被噪音淹没；如果真正失败只打 INFO，监控又发现不了。

推荐日志格式

商业项目日志至少包含：

字段	作用
时间	判断问题发生时间
级别	判断严重程度
模块	定位代码位置
request_id	串联一次请求
user_id/task_id	定位用户或任务
message	人能看懂的事件描述
cost_ms	排查慢请求
error_code	统计错误类型

基础格式：

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)

logger = logging.getLogger("asset.import")
logger.info("资产导入任务开始 batch_id=%s", "B20260705001")

不要用 f-string 拼日志参数：

logger.info(f"资产导入任务开始 batch_id={batch_id}")

推荐：

logger.info("资产导入任务开始 batch_id=%s", batch_id)

原因是：当日志级别被过滤时，参数化日志可以避免不必要的字符串拼接，也更符合 logging 的设计。

记录异常堆栈

错误写法：

try:
    1 / 0
except Exception as exc:
    logger.error("执行失败：%s", exc)

这只能看到 division by zero，看不到完整调用栈。

正确写法：

try:
    1 / 0
except Exception:
    logger.exception("执行失败")

logger.exception() 等价于在 ERROR 级别记录日志并带上 exc_info=True。它只能在 except 代码块中使用，因为它依赖当前捕获的异常上下文。

如果不在 except 中：

logger.error("调用失败", exc_info=True)

异常堆栈的价值是：它能告诉你错误从哪个函数一路传到当前日志点，而不是只告诉你最后一行报错。

Handler：输出到控制台和文件

一个 Logger 可以挂多个 Handler，例如同时输出到控制台和文件。

import logging
from logging.handlers import RotatingFileHandler


def setup_logging():
    logger = logging.getLogger("asset-api")
    logger.setLevel(logging.INFO)

    formatter = logging.Formatter(
        "%(asctime)s %(levelname)s [%(name)s] %(message)s"
    )

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)

    file_handler = RotatingFileHandler(
        "app.log",
        maxBytes=10 * 1024 * 1024,
        backupCount=10,
        encoding="utf-8",
    )
    file_handler.setFormatter(formatter)

    if not logger.handlers:
        logger.addHandler(console_handler)
        logger.addHandler(file_handler)

    return logger

为什么要 if not logger.handlers？因为在开发热重载、测试、多次初始化时，如果重复添加 Handler，同一条日志会打印多次。

日志切割

长期运行的服务不能只写一个无限增长的 app.log。日志文件越来越大会导致：

1. 磁盘被打满。
2. 打开和检索日志变慢。
3. 容器或服务器异常。

常见方式：

方式	说明	适合
RotatingFileHandler	按文件大小切割	单机服务、小项目
TimedRotatingFileHandler	按时间切割	每天一个日志文件
容器标准输出	交给 Docker/K8s 日志采集	容器化部署
日志平台	Filebeat、Fluent Bit、ELK 等	生产集中检索

按天切割：

from logging.handlers import TimedRotatingFileHandler

handler = TimedRotatingFileHandler(
    "app.log",
    when="midnight",
    backupCount=14,
    encoding="utf-8",
)

request_id 链路日志

一次 HTTP 请求通常会经过 API、Service、Repository、外部接口。如果没有 request_id，这些日志很难串起来。

flowchart TD
    A["请求进入"] --> B["生成或读取 request_id"]
    B --> C["API 日志带 request_id"]
    C --> D["Service 日志带 request_id"]
    D --> E["Repository 日志带 request_id"]
    D --> F["外部接口日志带 request_id"]
    E --> G["按 request_id 检索完整链路"]
    F --> G

FastAPI 中间件示例：

import logging
import time
import uuid
from contextvars import ContextVar
from fastapi import FastAPI, Request

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
logger = logging.getLogger("asset-api")
app = FastAPI()


@app.middleware("http")
async def request_log_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-Id", str(uuid.uuid4()))
    token = request_id_var.set(request_id)
    start = time.perf_counter()
    try:
        response = await call_next(request)
        cost_ms = int((time.perf_counter() - start) * 1000)
        logger.info(
            "request finished request_id=%s method=%s path=%s status=%s cost_ms=%s",
            request_id,
            request.method,
            request.url.path,
            response.status_code,
            cost_ms,
        )
        response.headers["X-Request-Id"] = request_id
        return response
    finally:
        request_id_var.reset(token)

ContextVar 的意义是：在异步程序中，每个请求都有自己的上下文，避免多个请求之间的 request_id 混乱。

结构化日志

普通文本日志适合人看，结构化日志适合日志平台检索和统计。结构化日志常用 JSON。

import json
import logging
import time


class JsonFormatter(logging.Formatter):
    def format(self, record):
        data = {
            "time": time.strftime("%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "line": record.lineno,
        }
        if record.exc_info:
            data["exception"] = self.formatException(record.exc_info)
        return json.dumps(data, ensure_ascii=False)

配置：

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger = logging.getLogger("asset-api")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

日志平台可以按 level、request_id、error_code、path 聚合，而不是只能模糊搜索文本。

敏感信息脱敏

日志可能被多人查看，也可能进入第三方日志平台。不能记录敏感信息。

不要记录：

1. 密码。
2. Token。
3. Cookie。
4. API Key。
5. 身份证号、手机号明文。
6. 医疗病历、检查报告明文。

脱敏示例：

def mask_phone(phone: str) -> str:
    if len(phone) != 11:
        return "***"
    return phone[:3] + "****" + phone[-4:]


def mask_id_card(id_card: str) -> str:
    if len(id_card) < 8:
        return "***"
    return id_card[:4] + "********" + id_card[-4:]

统一过滤器示例：

import logging
import re


class SensitiveFilter(logging.Filter):
    token_pattern = re.compile(r"(Bearer\s+)[A-Za-z0-9._-]+")

    def filter(self, record):
        message = record.getMessage()
        message = self.token_pattern.sub(r"\1***", message)
        record.msg = message
        record.args = ()
        return True

脱敏要尽量在日志输出前做，而不是等日志已经进入平台后再补救。

本地调试方法

本地调试的目标是复现、缩小范围、验证假设。

flowchart TD
    A["发现问题"] --> B["稳定复现"]
    B --> C["阅读错误堆栈"]
    C --> D["定位文件和行号"]
    D --> E["检查输入数据"]
    E --> F["断点观察变量"]
    F --> G["构造最小用例"]
    G --> H["修复并补测试"]

常用方式：

方法	适合场景
IDE 断点	复杂分支、对象状态变化
breakpoint()	快速进入 pdb
单元测试	固化复现场景
日志	观察多次运行和异步任务
最小复现	排除无关代码

pdb 示例：

def calculate_discount(price: int, vip: bool) -> int:
    breakpoint()
    if vip:
        return int(price * 0.8)
    return price


print(calculate_discount(100, True))

常用命令：

命令	作用
n	下一行
s	进入函数
c	继续运行
p variable	打印变量
l	查看附近代码
q	退出

线上排查方法

线上不能随便断点，主要靠日志、指标和复现。

接口慢排查：

flowchart TD
    A["用户反馈接口慢"] --> B["按 path 查看平均耗时和 P95"]
    B --> C["找一条慢请求 request_id"]
    C --> D["检索完整日志链路"]
    D --> E{"慢在哪里"}
    E -- "数据库" --> F["查 SQL EXPLAIN 锁等待 连接池"]
    E -- "外部接口" --> G["查超时 重试 下游状态"]
    E -- "代码计算" --> H["查 CPU 循环 大对象处理"]
    E -- "日志太多" --> I["降低日志量 异步采集"]

任务失败排查：

flowchart TD
    A["任务失败"] --> B["查 batch_id 或 task_id"]
    B --> C["查看开始日志"]
    C --> D["查看每一步处理数量"]
    D --> E["定位失败记录和异常堆栈"]
    E --> F["判断是否可重试"]
    F --> G["补偿处理并补测试"]

线上排查要有“三件套”：

1. request_id 或 task_id：定位单次链路。
2. 异常堆栈：定位代码位置。
3. 耗时字段：定位慢在哪里。

商业场景：资产导入失败

场景：医疗数据资产平台导入 CSV，用户反馈“导入失败，但不知道哪一行失败”。

不合格日志：

导入失败

合格日志：

2026-07-05 10:15:20 ERROR [asset.import] batch_id=B001 row=37 asset_name=检验报告 error_code=FIELD_INVALID message=字段类型不支持

更好的结构化日志：

{
  "time": "2026-07-05T10:15:20",
  "level": "ERROR",
  "module": "asset.import",
  "batch_id": "B001",
  "row": 37,
  "asset_name": "检验报告",
  "error_code": "FIELD_INVALID",
  "message": "字段类型不支持"
}

这样运维和开发可以直接回答：

1. 哪个批次失败。
2. 哪一行失败。
3. 哪个资产失败。
4. 错误原因是什么。
5. 是否可以修正数据后重试。

常见坑

问题	后果	正确做法
只用 print	线上无法分级、检索和保留	使用 logging
不记录异常堆栈	只知道失败，不知道哪里失败	logger.exception()
没有 request_id	一次请求日志串不起来	中间件生成并透传
记录敏感信息	形成安全事故	脱敏和过滤
日志无限增长	磁盘打满	日志切割或平台采集
重复添加 Handler	一条日志打印多次	初始化时检查 logger.handlers
生产开大量 DEBUG	性能下降、日志爆炸	生产默认 INFO
捕获异常后吞掉	错误被隐藏	记录日志后重新抛出或返回明确错误

面试标准回答

print 和 logging 有什么区别？
print 只是输出文本，适合临时调试；logging 是完整日志系统，支持级别、格式、输出目标、异常堆栈、过滤和线上采集。项目代码应该使用 logging，线上排查依赖结构化、可检索、可串联的日志。

Python logging 的核心组件有哪些？
核心组件包括 Logger、Level、Handler、Formatter、Filter 和 LogRecord。业务代码调用 Logger 生成 LogRecord，经过级别判断后交给 Handler，再由 Filter 过滤或补充字段，最后 Formatter 格式化并输出到控制台、文件或日志平台。

为什么要记录 request_id？
一次请求可能经过 API、Service、Repository 和外部接口。request_id 能把同一次请求的所有日志串起来，排查慢接口、异常和下游调用失败时可以快速定位完整链路。

线上接口慢怎么靠日志排查？
先看接口整体耗时和 P95，再找一条慢请求的 request_id，顺着日志拆分 API、数据库、外部接口、AI 调用和代码计算耗时。数据库慢看 SQL、执行计划、锁等待和连接池；外部接口慢看超时、重试和下游状态。

关联知识点

• Python Web API开发
• Python 测试
• Python 项目实践
• Python 数据库访问
• Python 面试题

小结

日志和调试的本质是让程序“可解释”。零基础先学会 logging、级别、异常堆栈和断点；进阶要学会 request_id、结构化日志、脱敏、日志切割和线上排查流程。商业项目不能只追求代码能跑，还要保证出错时能定位、能复现、能修复、能防止复发。