如何用Python案例演示日志模块的配置

访客 python案例 2026-06-04 18:22:07 2

本文目录导读：

基础配置案例
完整配置文件示例
实际应用案例 - Web服务
自定义日志过滤器
上下文日志示例
性能比较：不同配置方式

我来为你展示Python日志模块(logging)的详细配置案例：

基础配置案例

import logging
# 基础配置
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
# 创建日志记录器
logger = logging.getLogger(__name__)
# 测试不同级别的日志
logger.debug("这是DEBUG消息")
logger.info("这是INFO消息")
logger.warning("这是WARNING消息")
logger.error("这是ERROR消息")
logger.critical("这是CRITICAL消息")

完整配置文件示例

import logging
import logging.config
# 配置文件方式
LOGGING_CONFIG = {
    'version': 1,
    'formatters': {
        'default': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            'datefmt': '%Y-%m-%d %H:%M:%S'
        },
        'detailed': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(module)s - %(funcName)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'default',
            'stream': 'ext://sys.stdout'
        },
        'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'DEBUG',
            'formatter': 'detailed',
            'filename': 'app.log',
            'maxBytes': 1024*1024,  # 1MB
            'backupCount': 3,
            'encoding': 'utf8'
        },
        'error_file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'ERROR',
            'formatter': 'detailed',
            'filename': 'error.log',
            'maxBytes': 1024*1024,
            'backupCount': 3,
            'encoding': 'utf8'
        }
    },
    'loggers': {
        '': {  # root logger
            'handlers': ['console', 'file'],
            'level': 'DEBUG'
        },
        'myapp': {
            'handlers': ['console', 'file', 'error_file'],
            'level': 'DEBUG',
            'propagate': False
        },
        'myapp.database': {
            'handlers': ['console', 'error_file'],
            'level': 'INFO',
            'propagate': False
        }
    }
}
# 应用配置
logging.config.dictConfig(LOGGING_CONFIG)
# 创建不同模块的日志记录器
root_logger = logging.getLogger()
app_logger = logging.getLogger('myapp')
db_logger = logging.getLogger('myapp.database')

实际应用案例 - Web服务

import logging
import time
import random
from logging.handlers import TimedRotatingFileHandler
class WebAppLogger:
    """Web应用日志管理器"""
    def __init__(self, app_name='webapp'):
        self.app_name = app_name
        self.logger = logging.getLogger(app_name)
        self._setup_logger()
    def _setup_logger(self):
        """配置日志记录器"""
        self.logger.setLevel(logging.DEBUG)
        # 控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.INFO)
        console_format = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        console_handler.setFormatter(console_format)
        # 文件处理器（按天轮转）
        file_handler = TimedRotatingFileHandler(
            f'{self.app_name}.log',
            when='midnight',
            interval=1,
            backupCount=7,
            encoding='utf-8'
        )
        file_handler.setLevel(logging.DEBUG)
        file_format = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - '
            '%(module)s:%(lineno)d - %(message)s'
        )
        file_handler.setFormatter(file_format)
        # 错误日志处理器
        error_handler = TimedRotatingFileHandler(
            f'{self.app_name}_error.log',
            when='midnight',
            interval=1,
            backupCount=30,
            encoding='utf-8'
        )
        error_handler.setLevel(logging.ERROR)
        error_handler.setFormatter(file_format)
        # 添加处理器
        self.logger.addHandler(console_handler)
        self.logger.addHandler(file_handler)
        self.logger.addHandler(error_handler)
    def log_request(self, request_id, method, path, status_code, duration):
        """记录请求日志"""
        self.logger.info(
            f"Request {request_id}: {method} {path} - "
            f"Status: {status_code} - Duration: {duration:.3f}s"
        )
    def log_error(self, request_id, error, traceback_info):
        """记录错误日志"""
        self.logger.error(
            f"Error in request {request_id}: {str(error)}\n"
            f"Traceback: {traceback_info}"
        )
# 模拟Web应用运行
def simulate_web_app():
    logger = WebAppLogger('mywebapp')
    # 模拟请求处理
    for i in range(20):
        request_id = f"REQ-{i:04d}"
        method = random.choice(['GET', 'POST', 'PUT', 'DELETE'])
        path = random.choice(['/api/users', '/api/products', '/api/orders'])
        # 模拟一些正常的请求
        if random.random() < 0.1:  # 10%的概率产生错误
            status_code = random.choice([500, 502, 503])
            duration = random.uniform(0.5, 2.0)
            try:
                raise ValueError(f"模拟数据库连接失败")
            except ValueError as e:
                import traceback
                logger.log_error(request_id, e, traceback.format_exc())
        else:
            status_code = random.choice([200, 201, 204, 304])
            duration = random.uniform(0.01, 0.5)
            logger.log_request(request_id, method, path, status_code, duration)
        time.sleep(0.1)  # 模拟请求间隔
if __name__ == '__main__':
    # 运行模拟应用
    print("开始模拟Web应用日志...")
    simulate_web_app()
    print("日志模拟完成！")

自定义日志过滤器

import logging
import re
class SensitiveDataFilter(logging.Filter):
    """过滤敏感数据"""
    def __init__(self, patterns=None):
        super().__init__()
        self.patterns = patterns or {
            'password': r'password[=:]\s*\S+',
            'credit_card': r'\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}',
            'email': r'[\w\.-]+@[\w\.-]+\.\w+'
        }
    def filter(self, record):
        if hasattr(record, 'msg') and isinstance(record.msg, str):
            for name, pattern in self.patterns.items():
                record.msg = re.sub(
                    pattern,
                    f'[{name}_REDACTED]',
                    record.msg,
                    flags=re.IGNORECASE
                )
        return True
# 使用过滤器
class SecureLogger:
    def __init__(self):
        self.logger = logging.getLogger('secure_app')
        self.logger.setLevel(logging.DEBUG)
        # 添加敏感数据过滤器
        sensitive_filter = SensitiveDataFilter()
        self.logger.addFilter(sensitive_filter)
        # 添加处理器
        handler = logging.StreamHandler()
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
    def log_sensitive(self):
        """模拟包含敏感数据的日志"""
        self.logger.info("用户登录: email=user@example.com, password=secret123")
        self.logger.info("信用卡信息: 1234-5678-9012-3456")
        self.logger.info("用户信息: user@company.com, password=my_password")
# 测试过滤功能
secure_logger = SecureLogger()
print("测试敏感数据过滤:")
secure_logger.log_sensitive()

上下文日志示例

import logging
import threading
import uuid
class RequestContext:
    """请求上下文管理器"""
    _local = threading.local()
    @classmethod
    def get_request_id(cls):
        if not hasattr(cls._local, 'request_id'):
            cls._local.request_id = str(uuid.uuid4())[:8]
        return cls._local.request_id
    @classmethod
    def set_request_id(cls, request_id):
        cls._local.request_id = request_id
class ContextFormatter(logging.Formatter):
    """包含上下文信息的格式化器"""
    def format(self, record):
        record.request_id = RequestContext.get_request_id()
        return super().format(record)
# 创建上下文日志记录器
def setup_context_logger():
    logger = logging.getLogger('context_app')
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    formatter = ContextFormatter(
        '%(asctime)s [%(request_id)s] %(levelname)s: %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    return logger
# 模拟并发请求
def process_request(request_id):
    RequestContext.set_request_id(request_id)
    logger = setup_context_logger()
    logger.info("开始处理请求")
    logger.debug("执行数据库查询")
    logger.warning("资源使用率较高")
    logger.info("请求处理完成")
# 测试并发日志
threads = []
for i in range(3):
    thread = threading.Thread(
        target=process_request,
        args=(f"REQ-{chr(65+i)}",)
    )
    threads.append(thread)
    thread.start()
for thread in threads:
    thread.join()

性能比较：不同配置方式

import logging
import time
class LoggingPerformanceTest:
    """日志性能测试"""
    def __init__(self):
        self.setup_loggers()
    def setup_loggers(self):
        # 方式1: 无日志
        self.logger_none = logging.getLogger('none')
        self.logger_none.disabled = True
        # 方式2: 基本控制台日志
        self.logger_console = logging.getLogger('console')
        self.logger_console.setLevel(logging.INFO)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger_console.addHandler(handler)
        # 方式3: 详细日志
        self.logger_detailed = logging.getLogger('detailed')
        self.logger_detailed.setLevel(logging.DEBUG)
        handler = logging.StreamHandler()
        handler.setFormatter(
            logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            )
        )
        self.logger_detailed.addHandler(handler)
    def test_performance(self, iterations=10000):
        """测试不同配置的性能"""
        # 测试无日志
        start = time.time()
        for i in range(iterations):
            self.logger_none.info(f"测试消息 {i}")
        none_time = time.time() - start
        # 测试基本日志
        start = time.time()
        for i in range(iterations):
            self.logger_console.info(f"测试消息 {i}")
        console_time = time.time() - start
        # 测试详细日志
        start = time.time()
        for i in range(iterations):
            self.logger_detailed.info(f"测试消息 {i}")
        detailed_time = time.time() - start
        # 输出结果
        print(f"\n性能测试结果 ({iterations} 次迭代):")
        print(f"无日志: {none_time:.3f}s")
        print(f"基本日志: {console_time:.3f}s")
        print(f"详细日志: {detailed_time:.3f}s")
        print(f"基本日志 vs 无日志: {console_time/none_time:.2f}x 慢")
        print(f"详细日志 vs 基本日志: {detailed_time/console_time:.2f}x 慢")
# 运行性能测试
perf_test = LoggingPerformanceTest()
perf_test.test_performance(1000)  # 减少迭代次数以避免大量输出

这些示例展示了：