日志系统：logging模块、Handler、Formatter与最佳实践
日志系统：logging模块、Handler、Formatter与最佳实践
日志是程序运行时的重要记录，帮助我们调试问题、监控系统状态、追踪用户行为。Python的logging模块提供了灵活且强大的日志功能。本文将全面介绍日志系统的使用。
为什么需要日志
- 调试：定位代码中的问题
- 监控：了解系统运行状态
- 审计：追踪用户操作和系统事件
- 分析：通过日志统计用户行为
logging基础
import logging
# 基础配置
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 获取logger
logger = logging.getLogger(__name__)
# 不同级别的日志
logger.debug("调试信息")
logger.info("一般信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.critical("严重错误")
日志级别
import logging
# 日志级别从低到高
# DEBUG < INFO < WARNING < ERROR < CRITICAL
logger = logging.getLogger('example')
logger.setLevel(logging.DEBUG)
# 临时改变级别
logger.setLevel(logging.WARNING)
logger.info("这条不会显示") # 被过滤
logger.warning("这条会显示") # 会显示
# 检查级别
print(f"当前级别: {logger.level}")
print(f"DEBUG是否启用: {logger.isEnabledFor(logging.DEBUG)}")
Formatter格式化
import logging
# 自定义格式
formatter = logging.Formatter(
fmt='%(asctime)s | %(name)s | %(levelname)s | %(filename)s:%(lineno)d | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 常用格式变量
# %(name)s - Logger名称
# %(levelname)s - 日志级别
# %(asctime)s - 时间戳
# %(filename)s - 文件名
# %(lineno)d - 行号
# %(funcName)s - 函数名
# %(message)s - 日志消息
# %(thread)d - 线程ID
# %(process)d - 进程ID
# 创建handler并设置格式
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger = logging.getLogger('formatted')
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)
logger.info("格式化日志消息")
Handler处理器
Handler决定日志输出到哪里：
import logging
import sys
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
# 创建logger
logger = logging.getLogger('app')
logger.setLevel(logging.DEBUG)
# 1. 控制台Handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
))
# 2. 文件Handler
file_handler = logging.FileHandler('app.log', encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
# 3. 滚动文件Handler（按大小）
rotating_handler = RotatingFileHandler(
'app_rotating.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5
)
# 4. 定时滚动文件Handler
timed_handler = TimedRotatingFileHandler(
'app_timed.log',
when='midnight',
interval=1,
backupCount=30
)
# 添加handlers
logger.addHandler(console_handler)
logger.addHandler(file_handler)
logger.addHandler(rotating_handler)
# 使用
logger.info("应用启动")
logger.debug("详细调试信息")
logger.error("发生错误")
多模块日志配置
# config/logging_config.py
import logging
import logging.config
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'standard': {
'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
},
'detailed': {
'format': '%(asctime)s [%(levelname)s] %(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': 'app.log',
'maxBytes': 10485760,
'backupCount': 5,
'encoding': 'utf-8'
}
},
'root': {
'level': 'DEBUG',
'handlers': ['console', 'file']
},
'loggers': {
'app': {
'level': 'DEBUG',
'handlers': ['console', 'file'],
'propagate': False
},
'app.db': {
'level': 'WARNING',
'handlers': ['file'],
'propagate': False
}
}
}
def setup_logging():
logging.config.dictConfig(LOGGING_CONFIG)
使用logging.config
import logging
import logging.config
import yaml
# 从YAML文件加载配置
def setup_logging_from_yaml(config_path):
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
logging.config.dictConfig(config)
# logging.yaml 示例内容
"""
version: 1
disable_existing_loggers: false
formatters:
standard:
format: '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
handlers:
console:
class: logging.StreamHandler
level: INFO
formatter: standard
stream: ext://sys.stdout
root:
level: INFO
handlers:
- console
"""
# 从字典配置
config = {
'version': 1,
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'INFO'
}
},
'root': {
'level': 'INFO',
'handlers': ['console']
}
}
logging.config.dictConfig(config)
高级用法
import logging
import sys
from functools import wraps
# 1. 添加额外字段
class ContextFilter(logging.Filter):
def __init__(self, context):
super().__init__()
self.context = context
def filter(self, record):
record.context = self.context
return True
logger = logging.getLogger('context')
context_filter = ContextFilter('user_id=123')
logger.addFilter(context_filter)
# 2. 日志装饰器
def log_execution(logger):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
logger.info(f"开始执行 {func.__name__}")
try:
result = func(*args, **kwargs)
logger.info(f"{func.__name__} 执行成功")
return result
except Exception as e:
logger.error(f"{func.__name__} 执行失败: {e}")
raise
return wrapper
return decorator
# 3. 异常日志
def process_data(data):
try:
result = data / 0
except Exception:
logger.exception("处理数据时发生异常") # 自动记录堆栈
raise
# 4. 性能计时日志
import time
from contextlib import contextmanager
@contextmanager
def log_time(logger, operation):
start = time.time()
yield
elapsed = time.time() - start
logger.info(f"{operation} 耗时: {elapsed:.2f}秒")
实战：Web应用日志
import logging
import logging.config
from datetime import datetime
class WebLogger:
def __init__(self, name):
self.logger = logging.getLogger(name)
self.setup_logger()
def setup_logger(self):
self.logger.setLevel(logging.DEBUG)
# 格式
formatter = logging.Formatter(
'%(asctime)s | %(levelname)s | %(name)s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 控制台
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
# 文件
file_handler = logging.FileHandler(
f'app_{datetime.now().strftime("%Y%m%d")}.log',
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
self.logger.addHandler(console)
self.logger.addHandler(file_handler)
def request(self, method, path, status):
self.logger.info(f"{method} {path} - {status}")
def error(self, message, exc_info=False):
self.logger.error(message, exc_info=exc_info)
def user_action(self, user_id, action):
self.logger.info(f"User {user_id}: {action}")
# 使用示例
web_log = WebLogger('webapp')
web_log.request('GET', '/api/users', 200)
web_log.user_action(123, 'login')
最佳实践
import logging
import sys
# 1. 使用模块名作为logger名称
logger = logging.getLogger(__name__)
# 2. 合理使用日志级别
logger.debug("处理用户请求: %s", request_id) # 详细调试
logger.info("用户登录成功: %s", username) # 正常操作
logger.warning("磁盘空间不足: %d%%", space) # 潜在问题
logger.error("数据库连接失败: %s", error) # 错误
logger.critical("系统崩溃，请立即处理") # 严重错误
# 3. 使用延迟格式化避免字符串拼接开销
# 错误方式（即使日志被过滤也会拼接字符串）
# logger.info("Processing " + str(data))
# 正确方式（使用%格式化）
logger.info("Processing %s", data)
# 4. 记录异常信息
try:
risky_operation()
except Exception:
logger.exception("操作失败") # 自动记录堆栈
# 5. 结构化日志
logger.info("Order created", extra={
'order_id': 12345,
'amount': 99.99,
'user_id': 67890
})
# 6. 敏感信息过滤
class SensitiveFilter(logging.Filter):
PATTERNS = ['password', 'token', 'secret', 'api_key']
def filter(self, record):
msg = record.getMessage().lower()
return not any(p in msg for p in self.PATTERNS)
生产环境配置示例
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '%(levelname)s %(asctime)s %(module)s %(process)d %(thread)d %(message)s'
},
'simple': {
'format': '%(levelname)s %(message)s'
},
},
'filters': {
'require_debug_true': {
'()': 'django.utils.log.RequireDebugTrue',
},
},
'handlers': {
'console': {
'level': 'INFO',
'class': 'logging.StreamHandler',
'formatter': 'simple'
},
'file': {
'level': 'WARNING',
'class': 'logging.handlers.RotatingFileHandler',
'filename': '/var/log/app/warning.log',
'maxBytes': 1024*1024*5, # 5MB
'backupCount': 5,
'formatter': 'verbose',
},
},
'loggers': {
'django': {
'handlers': ['console'],
'level': 'INFO',
'propagate': True,
},
'app': {
'handlers': ['console', 'file'],
'level': 'DEBUG',
'propagate': True,
},
}
}
总结
合理的日志策略是生产系统的基石。记住：选择合适的日志级别、结构化日志消息、配置多个handler、在生产环境使用WARNING以上级别。好的日志能让你在问题发生时快速定位和解决。