python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python logging模块

Python logging模块详细教程与最佳实践

作者:cuber膜拜

本文详细介绍了Python的logging模块,从基础概念到高级用法,包括日志级别、格式、处理器、记录器等通过实例代码给大家介绍的非常详细,感兴趣的朋友跟随小编一起看看吧

Python logging模块完整教程

Python的logging模块是标准库中强大而灵活的日志记录工具,它能够帮助开发者记录应用程序运行过程中的各种事件和信息。通过本教程,您将从基础概念开始,逐步学习如何配置、使用和优化日志系统,最终掌握在实际项目中应用logging模块的各种技巧。本教程涵盖了从简单的日志输出到高级的日志配置和性能优化的完整内容。

什么是 Python logging

Python logging模块是Python标准库的一部分,提供了一个灵活的框架来记录应用程序运行时的各种事件和信息。

核心特点

快速开始

学习建议
建议先跟着这个例子动手实践,理解基本流程后再深入学习细节。

Hello World 示例

import logging
# 基本的日志记录
logging.debug("这是一条调试信息")
logging.info("这是一条信息")
logging.warning("这是一条警告")
logging.error("这是一条错误")
logging.critical("这是一条严重错误")

代码说明

  1. import logging: 导入logging模块
  2. logging.debug(): 记录调试级别信息
  3. logging.info(): 记录普通信息
  4. logging.warning(): 记录警告信息
  5. logging.error(): 记录错误信息
  6. logging.critical(): 记录严重错误信息

运行结果

WARNING:root:这是一条警告
ERROR:root:这是一条错误
CRITICAL:root:这是一条严重错误

代码解析

[!question] 为什么要这样写?
logging模块的默认配置相对保守,只显示WARNING级别的日志。这是因为过多的日志信息可能会干扰程序的主要输出,同时日志系统的开销也应该被控制。

[!bug] 如果没有看到DEBUG和INFO日志
这是正常的!logging默认只显示WARNING及以上级别的日志。要显示所有级别的日志,需要配置logging级别。

核心概念

概念1:Logger(记录器)

定义: Logger是日志系统的核心组件,用于创建和记录日志消息。

为什么重要: Logger为应用程序提供了记录日志的入口点,它决定了哪些级别的消息会被记录,以及消息应该发送到哪里。

工作原理:
Logger创建后会被分配一个名称和日志级别。当调用logger的日志方法时,会检查消息的级别是否高于或等于logger的级别,如果是则处理该消息。

示例

import logging
# 创建logger
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)
# 记录日志
logger.debug("调试消息")
logger.info("信息消息")
logger.warning("警告消息")

[!note] 深入理解

  • Logger名称通常采用点分格式,如’my_app.database’
  • 同一个名称的logger返回的是同一个实例
  • 如果不设置级别,会继承父logger的级别

概念2:Handler(处理器)

定义: Handler负责将日志消息发送到目标位置,如控制台、文件、网络等。

为什么重要: Handler决定了日志的最终去向,是实现日志系统灵活性的关键。

工作原理:
每个Logger可以附加多个Handler。当Logger记录消息时,会将消息传递给所有附加的Handler,每个Handler根据其配置独立处理消息。

示例

import logging
# 创建logger
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)
# 控制台handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 文件handler
file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.DEBUG)
# 添加handler
logger.addHandler(console_handler)
logger.addHandler(file_handler)
# 记录日志
logger.debug("只记录到文件")
logger.info("同时记录到控制台和文件")

[!note] 深入理解

  • 一个Logger可以关联多个Handler
  • 每个Handler可以有自己的日志级别和格式
  • 可以通过Handler实现日志过滤、分发等功能

概念3:Formatter(格式化器)

定义: Formatter定义了日志消息的输出格式。

为什么重要: 好的日志格式能够使日志信息更加清晰、易读,便于问题排查。

工作原理:
Formatter将日志事件对象转换为字符串。可以自定义包含时间戳、日志级别、logger名称、消息等字段。

示例

import logging
# 创建formatter
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 创建handler并设置formatter
handler = logging.StreamHandler()
handler.setFormatter(formatter)
# 使用handler
logger = logging.getLogger('my_app')
logger.addHandler(handler)
# 记录日志
logger.info("格式化的日志消息")

输出

2024-01-01 12:00:00,123 - my_app - INFO - 格式化的日志消息

[!note] 深入理解

概念之间的关系

Logger作为入口点,通过日志级别控制哪些消息会被记录。Logger可以关联多个Handler,每个Handler可以有自己的Formatter来格式化输出。这样实现了日志系统的灵活配置。

基础功能详解

功能1:日志级别

用途: 控制哪些信息应该被记录,帮助过滤日志信息。

基本语法

logging.debug("消息")      # 调试级别
logging.info("消息")       # 信息级别
logging.warning("消息")     # 警告级别
logging.error("消息")       # 错误级别
logging.critical("消息")   # 严重错误级别

参数说明

简单示例

import logging
# 设置日志级别
logging.basicConfig(level=logging.DEBUG)
logging.debug("调试信息:程序开始执行")
logging.info("信息:加载配置文件")
logging.warning("警告:配置文件版本可能过时")
logging.error("错误:无法连接到数据库")
logging.critical("严重错误:系统即将崩溃")

常见用法

import logging
# 获取logger
logger = logging.getLogger('app')
# 设置日志级别
logger.setLevel(logging.INFO)  # 只记录INFO及以上级别
# 针对不同级别有不同处理
if some_critical_condition:
    logger.critical("关键问题发生,需要立即处理")
elif some_error_condition:
    logger.error("发生错误,影响功能正常运行")
else:
    logger.info("系统运行正常")

[!tip] 最佳实践

[!warning] 常见陷阱

功能2:基本配置

用途: 快速配置logging系统的基本参数。

基本语法

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='app.log',
    filemode='a'
)

参数说明

简单示例

import logging
# 基本配置
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='my_app.log',
    filemode='a'
)
# 现在所有日志都会写入文件
logging.debug("调试信息")
logging.info("信息")

常见用法

import logging
# 生产环境配置
logging.basicConfig(
    level=logging.WARNING,
    format='%(asctime)s [%(levelname)s] %(message)s',
    filename='production.log'
)
# 开发环境配置
import sys
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s %(name)s %(levelname)s: %(message)s',
    stream=sys.stdout  # 输出到控制台
)

[!tip] 最佳实践

[[文件操作]]

功能3:Logger创建和使用

用途: 创建和管理应用程序的logger实例。

基本语法

logger = logging.getLogger('logger_name')
logger.debug("消息")

参数说明

简单示例

import logging
# 创建不同模块的logger
db_logger = logging.getLogger('my_app.database')
web_logger = logging.getLogger('my_app.web')
user_logger = logging.getLogger('my_app.user')
# 配置logger级别
db_logger.setLevel(logging.DEBUG)
web_logger.setLevel(logging.INFO)
user_logger.setLevel(logging.WARNING)
# 记录日志
db_logger.debug("数据库连接建立")
db_logger.info("执行SQL查询")
web_logger.info("收到HTTP请求")
user_logger.warning("用户尝试访问无权限的页面")

常见用法

import logging
import os
# 基于模块名创建logger
logger = logging.getLogger(__name__)
# 如果logger还没有配置,进行基本配置
if not logger.handlers:
    handler = logging.StreamHandler()
    formatter = logging.Formatter(
        '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
# 使用logger
logger.debug("Debug message")
logger.info("App started")
logger.error("Failed to process request")

[!tip] 最佳实践

功能4:Handler配置

用途: 配置日志输出的目标位置和方式。

基本语法

# StreamHandler - 输出到控制台
console_handler = logging.StreamHandler()
# FileHandler - 输出到文件
file_handler = logging.FileHandler('app.log')
# RotatingFileHandler - 轮转文件
from logging.handlers import RotatingFileHandler
rotating_handler = RotatingFileHandler(
    'app.log',
    maxBytes=1024*1024,  # 1MB
    backupCount=5
)
# TimedRotatingFileHandler - 按时间轮转
from logging.handlers import TimedRotatingFileHandler
timed_handler = TimedRotatingFileHandler(
    'app.log',
    when='midnight',
    backupCount=7
)

参数说明

简单示例

import logging
from logging.handlers import RotatingFileHandler
# 创建logger
logger = logging.getLogger('app')
logger.setLevel(logging.DEBUG)
# 控制台handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 文件handler
file_handler = RotatingFileHandler(
    'app.log',
    maxBytes=1024*1024,  # 1MB
    backupCount=3,
    encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
# 添加handler
logger.addHandler(console_handler)
logger.addHandler(file_handler)
# 测试日志
logger.debug("这是调试信息,只记录到文件")
logger.info("这是信息,同时记录到控制台和文件")

常见用法

import logging
import sys
from logging.handlers import SMTPHandler
# 配置多输出
logger = logging.getLogger('my_app')
# 控制台输出
console = logging.StreamHandler(sys.stdout)
console.setLevel(logging.INFO)
console_format = logging.Formatter('%(levelname)s: %(message)s')
console.setFormatter(console_format)
# 文件输出
file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.DEBUG)
file_format = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(file_format)
# 邮件handler(仅用于严重错误)
mail_handler = SMTPHandler(
    mailhost=('smtp.example.com', 587),
    fromaddr='app@example.com',
    toaddrs=['admin@example.com'],
    subject='Application Error',
    credentials=('username', 'password'),
    secure=()
)
mail_handler.setLevel(logging.CRITICAL)
# 添加handler
logger.addHandler(console)
logger.addHandler(file_handler)
logger.addHandler(mail_handler)

[!tip] 最佳实践

[[异常处理]]
[[Python基础语法]]

实践项目

[!example] 项目目标
通过构建一个简单的Web服务器日志系统,综合运用前面学到的logging知识。

项目需求

项目结构

web_logger/
├── main.py
├── logger_config.py
├── requirements.txt
└── logs/
    └── app.log

实现步骤

步骤1:创建日志配置模块

目标:创建一个可复用的日志配置模块。

代码

# logger_config.py
import logging
import logging.handlers
import os
from datetime import datetime
class WebLoggerConfig:
    def __init__(self, log_dir='logs'):
        self.log_dir = log_dir
        self.logger = logging.getLogger('web_server')
        self._setup_logger()
    def _setup_logger(self):
        # 确保日志目录存在
        os.makedirs(self.log_dir, exist_ok=True)
        # 设置logger级别
        self.logger.setLevel(logging.DEBUG)
        # 清除现有的handler
        self.logger.handlers.clear()
        # 创建formatter
        detailed_formatter = logging.Formatter(
            '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
        )
        # 控制台handler
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.INFO)
        console_handler.setFormatter(detailed_formatter)
        # 文件handler - 主日志
        file_handler = logging.handlers.RotatingFileHandler(
            os.path.join(self.log_dir, 'app.log'),
            maxBytes=10*1024*1024,  # 10MB
            backupCount=5,
            encoding='utf-8'
        )
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(detailed_formatter)
        # 错误日志handler
        error_handler = logging.handlers.RotatingFileHandler(
            os.path.join(self.log_dir, 'error.log'),
            maxBytes=5*1024*1024,  # 5MB
            backupCount=3,
            encoding='utf-8'
        )
        error_handler.setLevel(logging.ERROR)
        error_handler.setFormatter(detailed_formatter)
        # 添加所有handler
        self.logger.addHandler(console_handler)
        self.logger.addHandler(file_handler)
        self.logger.addHandler(error_handler)
    def get_logger(self):
        return self.logger

解释

步骤2:创建主应用程序

目标:模拟Web服务器处理HTTP请求并记录日志。

代码

# main.py
from logger_config import WebLoggerConfig
import random
import time
from datetime import datetime
# 初始化日志系统
logger_config = WebLoggerConfig()
logger = logger_config.get_logger()
def simulate_web_request(request_id):
    """模拟处理HTTP请求"""
    logger.info(f"收到请求 #{request_id}")
    # 模拟请求处理时间
    process_time = random.uniform(0.1, 2.0)
    time.sleep(process_time)
    # 模拟不同的响应状态
    status_code = random.choices(
        [200, 404, 500],
        weights=[0.8, 0.15, 0.05]
    )[0]
    # 记录处理结果
    if status_code == 200:
        logger.debug(f"请求 #{request_id} 处理成功,耗时: {process_time:.2f}s")
        logger.info(f"响应 #{request_id}: {status_code} OK")
        return "OK"
    elif status_code == 404:
        logger.warning(f"请求 #{request_id}: 404 Not Found - 资源不存在")
        return "Not Found"
    else:
        logger.error(f"请求 #{request_id} 处理失败,错误代码: {status_code}")
        logger.error(f"详细信息: 服务器内部错误")
        return "Internal Server Error"
def main():
    """主程序"""
    logger.info("Web服务器启动")
    logger.info("开始处理请求...")
    try:
        # 模拟处理100个请求
        for i in range(1, 101):
            result = simulate_web_request(i)
            # 每10个请求输出统计信息
            if i % 10 == 0:
                logger.info(f"已处理 {i} 个请求")
    except KeyboardInterrupt:
        logger.warning("收到中断信号,正在关闭服务器...")
    except Exception as e:
        logger.critical(f"服务器发生严重错误: {str(e)}", exc_info=True)
    finally:
        logger.info("Web服务器关闭")
if __name__ == "__main__":
    main()

解释

完整代码

项目的完整代码包含两个文件:

  1. logger_config.py - 日志配置模块
  2. main.py - 主应用程序

运行和测试

# 安装依赖(这个项目不需要额外依赖)
pip install -r requirements.txt
# 运行程序
python main.py

预期输出

2024-01-01 10:00:00,123 [INFO] web_server: Web服务器启动
2024-01-01 10:00:00,124 [INFO] web_server: 开始处理请求...
2024-01-01 10:00:00,125 [INFO] web_server: 收到请求 #1
2024-01-01 10:00:01,321 [INFO] web_server: 响应 #1: 200 OK
2024-01-01 10:00:01,322 [INFO] web_server: 收到请求 #2
2024-01-01 10:00:02,456 [WARNING] web_server: 请求 #2: 404 Not Found - 资源不存在
...
2024-01-01 10:00:15,789 [INFO] web_server: 已处理 10 个请求
...

检查logs目录下的日志文件,可以看到:

进阶主题

主题1:自定义日志级别

什么时候需要: 当内置的日志级别不足以满足业务需求时,如需要区分不同类型的警告。

深入理解:
Python logging允许创建自定义的日志级别,这样可以根据业务需求定义更细粒度的日志分类。自定义级别需要定义级别数值和级别名称。

高级示例

import logging
# 定义自定义级别
logging.addLevelName(35, "TRACE")  # 比DEBUG更低的级别
logging.addLevelName(45, "NOTICE")  # 介于INFO和WARNING之间
def trace(self, message, *args, **kwargs):
    """TRACE级别方法"""
    if self.isEnabledFor(35):
        self._log(35, message, args, **kwargs)
def notice(self, message, *args, **kwargs):
    """NOTICE级别方法"""
    if self.isEnabledFor(45):
        self._log(45, message, args, **kwargs)
# 添加方法到Logger类
logging.Logger.trace = trace
logging.Logger.notice = notice
# 使用自定义级别
logger = logging.getLogger('app')
logger.trace("详细的跟踪信息")
logger.debug("调试信息")
logger.notice("需要注意的事项")
logger.info("常规信息")

性能考虑:

主题2:上下文感知的日志

什么时候需要: 需要在日志中包含请求ID、用户ID等上下文信息时。

深入理解:
通过使用logger的adapter功能,可以轻松地为所有日志消息添加上下文信息,这样在分布式系统中可以追踪完整的请求链路。

高级示例

import logging
class ContextAdapter(logging.LoggerAdapter):
    """带上下文的Logger适配器"""
    def __init__(self, logger, extra=None):
        super().__init__(logger, extra or {})
    def process(self, msg, kwargs):
        # 为每条消息添加上下文
        kwargs['extra'] = kwargs.get('extra', {})
        kwargs['extra'].update(self.extra)
        return msg, kwargs
# 使用示例
logger = logging.getLogger('web_service')
adapter = ContextAdapter(logger, {
    'request_id': 'req-12345',
    'user_id': 'user-67890',
    'ip_address': '192.168.1.1'
})
# 记录日志
adapter.info("用户登录成功")
adapter.error("数据库连接失败")
# 输出示例:
# 2024-01-01 12:00:00 [INFO] web_service: 用户登录成功
#   request_id=req-12345 user_id=user-67890 ip_address=192.168.1.1

性能考虑:

主题3:异步日志

什么时候需要: 在高并发环境下,同步日志记录可能成为性能瓶颈时。

深入理解:
异步日志通过后台线程处理日志写入,避免阻塞主线程。Python 3.7+的logging模块内置了QueueHandler和QueueListener来实现异步日志。

高级示例

import logging
import logging.handlers
import queue
import time
from threading import Thread
class AsyncLogging:
    def __init__(self):
        self.log_queue = queue.Queue(-1)  # 无限队列
        self.setup_logger()
    def setup_logger(self):
        # 配置logger
        logger = logging.getLogger('async_app')
        logger.setLevel(logging.DEBUG)
        # 创建QueueHandler
        queue_handler = logging.handlers.QueueHandler(self.log_queue)
        logger.addHandler(queue_handler)
        # 创建QueueListener
        file_handler = logging.FileHandler('async_app.log')
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        ))
        # 启动监听器线程
        self.listener = logging.handlers.QueueListener(
            self.log_queue,
            file_handler,
            respect_handler_level=True
        )
        self.listener.start()
        # 保存logger
        self.logger = logger
    def stop(self):
        """停止异步日志"""
        self.logger.info("正在关闭异步日志系统")
        self.listener.stop()
    def get_logger(self):
        return self.logger
# 使用异步日志
async_logger = AsyncLogging()
logger = async_logger.get_logger()
# 模拟高并发场景
def worker(worker_id):
    for i in range(10):
        logger.info(f"Worker {worker_id} 处理任务 {i}")
        time.sleep(0.1)
# 创建多个线程
threads = []
for i in range(5):
    t = Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()
# 等待所有线程完成
for t in threads:
    t.join()
# 停止异步日志
async_logger.stop()

性能考虑:

主题4:结构化日志

什么时候需要: 需要将日志数据发送到日志分析系统(如ELK Stack)时。

深入理解:
结构化日志使用JSON等格式记录日志,便于日志解析和分析。Python 3.9+的logging模块支持直接输出JSON格式的日志。

高级示例

import logging
import json
from datetime import datetime
class JSONFormatter(logging.Formatter):
    """JSON格式的日志格式化器"""
    def format(self, record):
        log_data = {
            'timestamp': datetime.fromtimestamp(record.created).isoformat(),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno,
            'thread': record.thread,
            'process': record.process
        }
        # 添加异常信息
        if record.exc_info:
            log_data['exception'] = self.formatException(record.exc_info)
        # 添加额外的字段
        if hasattr(record, 'extra_data'):
            log_data.update(record.extra_data)
        return json.dumps(log_data, ensure_ascii=False)
# 配置结构化日志
logger = logging.getLogger('structured')
logger.setLevel(logging.DEBUG)
# 文件handler
file_handler = logging.FileHandler('structured.log')
file_handler.setFormatter(JSONFormatter())
# 控制台handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter(
    '%(levelname)s: %(message)s'
))
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# 记录结构化日志
logger.info("用户登录", extra={
    'user_id': '12345',
    'ip': '192.168.1.1',
    'device': 'mobile'
})
logger.error("数据库操作失败", exc_info=True)

输出示例(JSON)

{
  "timestamp": "2024-01-01T12:00:00.123456",
  "level": "INFO",
  "logger": "structured",
  "message": "用户登录",
  "module": "main",
  "function": "main",
  "line": 42,
  "thread": 12345,
  "process": 67890,
  "user_id": "12345",
  "ip": "192.168.1.1",
  "device": "mobile"
}

到此这篇关于Python logging模块详细教程与最佳实践的文章就介绍到这了,更多相关Python logging模块内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文