Python Logging 详解

import logging

# 设置日志等级和输出格式
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)

logging.debug("调试信息")
logging.info("普通信息")
logging.warning("警告")
logging.error("错误")
logging.critical("严重错误")

Logger（日志记录器）     ← 你在代码中调用的接口
  ↓
Handler（日志处理器）    ← 决定日志输出到哪里（终端、文件等）
  ↓
Formatter（格式器）      ← 决定日志内容怎么显示
  ↓
Filter（过滤器）         ← 控制哪些日志可以通过

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

from logging.handlers import RotatingFileHandler

rotating = RotatingFileHandler(
    filename="app.log",
    maxBytes=5*1024*1024,  # 每个文件最大5MB
    backupCount=3,         # 最多保留3个旧日志
    encoding="utf-8"
)
rotating.setLevel(logging.INFO)

from logging.handlers import TimedRotatingFileHandler

timed = TimedRotatingFileHandler(
    filename="app.log",
    when="midnight",       # 每天0点切分
    backupCount=7,         # 保留7天日志
    encoding="utf-8"
)
timed.setLevel(logging.INFO)

from logging.handlers import QueueHandler, QueueListener
from multiprocessing import Queue

log_queue = Queue()
queue_handler = QueueHandler(log_queue)

listener = QueueListener(log_queue, timed)
listener.start()

logger.addHandler(queue_handler)

logger = logging.getLogger("myapp")

# 控制台显示所有日志
console = logging.StreamHandler()
console.setLevel(logging.DEBUG)

# 文件仅记录错误
file = logging.FileHandler("error.log")
file.setLevel(logging.ERROR)

logger.addHandler(console)
logger.addHandler(file)
logger.setLevel(logging.DEBUG)

from logging import Formatter

formatter = Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

handler.setFormatter(formatter)

Formatter('%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')

Formatter('%(asctime)s', datefmt='%Y/%m/%d %H:%M:%S')
# 输出: 2025/07/14 10:30:10

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

2025-07-14 10:30:10 [INFO] app.user.login_user:42 - 用户登录成功

# 自定义 filter 注入字段
class TraceIdFilter(logging.Filter):
    def filter(self, record):
        record.trace_id = "abc123xyz"
        return True

'%(asctime)s [%(trace_id)s] %(message)s'

class MyFormatter(logging.Formatter):
    def format(self, record):
        record.msg = f"[MyPrefix] {record.msg}"
        return super().format(record)

import json

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log = {
            "time": self.formatTime(record),
            "level": record.levelname,
            "msg": record.getMessage(),
            "trace_id": getattr(record, "trace_id", "-")
        }
        return json.dumps(log, ensure_ascii=False)

console_handler.setFormatter(color_formatter)
file_handler.setFormatter(json_formatter)

import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "formatters": {
        "default": {
            "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "default",
            "level": "DEBUG"
        },
        "file": {
            "class": "logging.FileHandler",
            "formatter": "default",
            "filename": "logfile.log",
            "level": "INFO"
        }
    },
    "root": {
        "handlers": ["console", "file"],
        "level": "DEBUG"
    }
}

logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger(__name__)