Python logging 库全面指南

Python logging 库全面指南

flyfish

一、概述

logging 是 Python 的标准库之一，用于记录程序运行时的信息。它提供了灵活的日志级别设置、日志格式定制以及多种输出方式（如控制台、文件等），便于开发者监控和调试程序。在软件开发过程中，良好的日志记录能帮助开发者追踪程序执行流程、定位错误以及分析系统性能。

二、基本组件

Logger（日志记录器）

负责收集日志消息，并将其发送给与之关联的 Handler。可以通过 logging.getLogger() 函数获取 Logger 实例，获取时可指定名称，若名称为 None，则返回根记录器。不同模块或功能可创建不同的 Logger，方便组织和管理日志。

import logging

# 获取名为 my_logger 的 Logger 实例
logger = logging.getLogger('my_logger')

Handler（处理器）

决定日志的输出位置，常见的有 StreamHandler（输出到控制台）和 FileHandler（输出到文件）。每个 Logger 可以关联多个 Handler，从而实现将日志消息发送到多个不同的目的地。

import logging

# 创建 StreamHandler
stream_handler = logging.StreamHandler()
# 创建 FileHandler
file_handler = logging.FileHandler('app.log')

Formatter（格式化器）

用于定义日志输出的格式，包括日志级别、时间、消息内容等信息的显示方式。开发者可以使用 Formatter 类创建格式化器实例，并通过自定义格式字符串满足不同的日志格式需求。

import logging

# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

Filter（过滤器）

根据条件过滤日志消息，只有满足条件的日志消息才会被处理。通过创建自定义的 Filter 类，可以实现更复杂的过滤逻辑。

import logging

class ModuleFilter(logging.Filter):
    def __init__(self, module_name):
        self.module_name = module_name

    def filter(self, record):
        return record.name == self.module_name

# 创建过滤器实例
filter = ModuleFilter('specific_module')

三、基本使用

简单配置和使用

使用 basicConfig 函数可以进行简单的日志配置，设置日志级别、格式和输出位置等。

import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logging.info('This is an info message')
logging.warning('This is a warning message')

使用 Logger 实例

import logging

# 获取 Logger 实例
logger = logging.getLogger('my_logger')
logger.setLevel(logging.DEBUG)

# 创建 StreamHandler
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.INFO)

# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)

# 将 Handler 添加到 Logger
logger.addHandler(stream_handler)

# 记录日志
logger.debug('This is a debug message')
logger.info('This is an info message')

四、高级特性

日志级别

logging 库定义了多个日志级别，从低到高依次为 DEBUG（调试信息）、INFO（普通信息）、WARNING（警告）、ERROR（错误）和 CRITICAL（严重错误）。通过设置 Logger 和 Handler 的日志级别，只有日志级别大于等于设置级别的日志消息才会被处理。

import logging

# 获取 Logger 实例
logger = logging.getLogger('my_logger')
# 设置 Logger 级别为 DEBUG
logger.setLevel(logging.DEBUG)

# 创建 StreamHandler 并设置级别为 WARNING
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.WARNING)

# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)

# 将 Handler 添加到 Logger
logger.addHandler(stream_handler)

# 记录不同级别的日志
logger.debug('This is a debug message')  # 不会输出，因为 Handler 级别为 WARNING
logger.warning('This is a warning message')

日志格式定制

Formatter 支持丰富的格式化字段，除了基本的时间、级别和消息内容外，还有 %(module)s（模块名）、%(lineno)d（行号）等。通过组合这些字段，可以定义出非常详细和个性化的日志格式。

import logging

# 创建格式化器，包含模块名和行号
formatter = logging.Formatter('%(asctime)s - %(module)s:%(lineno)d - %(levelname)s - %(message)s')

# 获取 Logger 实例
logger = logging.getLogger('my_logger')
logger.setLevel(logging.DEBUG)

# 创建 StreamHandler
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)
stream_handler.setFormatter(formatter)

# 将 Handler 添加到 Logger
logger.addHandler(stream_handler)

# 记录日志
logger.info('This is a detailed info message')

多 Handler 和多 Logger

一个 Logger 可以添加多个不同类型的 Handler，例如同时将日志输出到控制台和文件。在大型项目中，还可以为不同的模块或功能创建多个 Logger，每个 Logger 可以有自己独立的配置和处理方式。

import logging

# 创建两个不同的 Logger
logger1 = logging.getLogger('logger1')
logger2 = logging.getLogger('logger2')

# 设置 Logger 级别
logger1.setLevel(logging.DEBUG)
logger2.setLevel(logging.INFO)

# 创建 StreamHandler 和 FileHandler
stream_handler = logging.StreamHandler()
file_handler = logging.FileHandler('app.log')

# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)

# 将 Handler 添加到 Logger1
logger1.addHandler(stream_handler)
logger1.addHandler(file_handler)

# 将 StreamHandler 添加到 Logger2
logger2.addHandler(stream_handler)

# 记录日志
logger1.debug('Debug message from logger1')
logger2.info('Info message from logger2')

日志过滤

通过创建自定义的 Filter 类，可以实现更复杂的日志过滤功能。

import logging

class ModuleFilter(logging.Filter):
    def __init__(self, module_name):
        self.module_name = module_name

    def filter(self, record):
        return record.name == self.module_name

# 获取 Logger 实例
logger = logging.getLogger('specific_module')
logger.setLevel(logging.DEBUG)

# 创建 StreamHandler
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)

# 创建过滤器并添加到 Handler
filter = ModuleFilter('specific_module')
stream_handler.addFilter(filter)

# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)

# 将 Handler 添加到 Logger
logger.addHandler(stream_handler)

# 记录日志
logger.debug('This is a filtered debug message')

五、实际项目中的应用

Web 应用中的日志记录（Flask 示例）

在 Flask 应用中，logging 库可用于记录用户的请求信息、服务器的响应状态等。

from flask import Flask
import logging

app = Flask(__name__)

# 配置日志
app.logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)
app.logger.addHandler(stream_handler)

@app.route('/')
def home():
    app.logger.info('Processing default request')
    return "Hello, World!"

if __name__ == '__main__':
    app.run()

数据处理任务中的日志记录

在进行数据处理任务时，logging 库可记录数据的读取、清洗、转换等过程中的关键信息。

import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def process_data(data):
    logging.info("Starting data processing.")
    # 模拟数据处理逻辑
    try:
        processed_data = [i * 2 for i in data]
        logging.info("Data processing completed successfully.")
        return processed_data
    except Exception as e:
        logging.error(f"Data processing failed: {e}")
        return None

if __name__ == "__main__":
    sample_data = [1, 2, 3, 4, 5]
    result = process_data(sample_data)
    if result:
        logging.info(f"Processed data: {result}")

分布式系统中的日志记录

在分布式系统中，logging 库可以与分布式日志管理工具（如 Elasticsearch、Kibana 等）结合使用，将各个节点的日志集中收集和管理。以下是一个简单的示例，模拟将日志发送到 Elasticsearch。

import logging
from elasticsearch import Elasticsearch

# 连接到 Elasticsearch
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])

class ElasticsearchHandler(logging.Handler):
    def __init__(self, index_name):
        super().__init__()
        self.index_name = index_name

    def emit(self, record):
        try:
            log_entry = {
                'timestamp': record.created,
                'level': record.levelname,
                'message': self.format(record)
            }
            es.index(index=self.index_name, body=log_entry)
        except Exception as e:
            self.handleError(record)

# 获取 Logger 实例
logger = logging.getLogger('distributed_logger')
logger.setLevel(logging.DEBUG)

# 创建 ElasticsearchHandler
es_handler = ElasticsearchHandler('my_log_index')
es_handler.setLevel(logging.DEBUG)

# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
es_handler.setFormatter(formatter)

# 将 Handler 添加到 Logger
logger.addHandler(es_handler)

# 记录日志
logger.debug('This is a debug message for distributed system')

六、配置方式

代码内配置

直接在代码中通过调用 logging 库的函数和方法进行配置，如前面示例中使用的 basicConfig 和创建 Logger、Handler 等实例并进行配置的方式。

import logging

# 代码内配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logging.info('This is an info message from code configuration')

配置文件方式

logging 库支持通过配置文件来进行日志配置，常用的配置文件格式有 JSON、YAML 等。通过配置文件可以将日志配置与代码分离，方便在不修改代码的情况下对日志进行调整和优化。

示例 YAML 配置文件（logging_config.yaml）

version: 1
disable_existing_loggers: False
formatters:
    simple:
        format: '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
handlers:
    console:
        class: logging.StreamHandler
        level: DEBUG
        formatter: simple
        stream: ext://sys.stdout
loggers:
    my_module:
        level: ERROR
        handlers: [console]
        propagate: no
root:
    level: DEBUG
    handlers: [console]

加载该配置文件的代码

import logging.config
import yaml

with open('logging_config.yaml', 'r') as f:
    config = yaml.safe_load(f)
    logging.config.dictConfig(config)

logger = logging.getLogger('my_module')
logger.error('This is an error message from my_module')

七、与其他库的集成

与 Web 框架集成（Django 示例）

在 Django 项目中，可以在 settings.py 中配置 logging 库。

# settings.py
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'verbose': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        },
    },
    'handlers': {
        'console': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
            'formatter': 'verbose'
        },
    },
    'loggers': {
        'django': {
            'handlers': ['console'],
            'level': 'INFO',
            'propagate': True,
        },
        'my_app': {
            'handlers': ['console'],
            'level': 'DEBUG',
            'propagate': True,
        },
    },
}

在 Django 视图中使用日志：

# views.py
import logging

logger = logging.getLogger('my_app')

def my_view(request):
    logger.info('This is an info message from Django view')
    return HttpResponse('Hello, Django!')

与数据库操作库集成（SQLAlchemy 示例）

当使用 SQLAlchemy 进行数据库操作时，可以将 logging 库与数据库操作结合起来，记录数据库的连接、查询、更新等操作的日志。

import logging
from sqlalchemy import create_engine, text

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger('sqlalchemy')
logger.setLevel(logging.INFO)

# 创建数据库引擎
engine = create_engine('sqlite:///example.db')

try:
    # 记录数据库连接信息
    logger.info('Connecting to the database...')
    with engine.connect() as conn:
        # 执行查询
        result = conn.execute(text('SELECT 1'))
        logger.info('Query executed successfully')
        for row in result:
            logger.info(f'Query result: {row}')
except Exception as e:
    logger.error(f'Database operation failed: {e}')

八、常用函数和方法

配置相关

basicConfig(**kwargs)：简单地配置日志系统，常用于快速设置日志级别、格式和输出位置。

import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

dictConfig(config)：通过字典配置日志系统，适合从文件（如 JSON、YAML）加载配置。

import logging.config
import yaml

with open('logging_config.yaml', 'r') as f:
    config = yaml.safe_load(f)
    logging.config.dictConfig(config)

fileConfig(fname, defaults=None)：使用 INI 文件格式配置日志系统。

Logger 相关

getLogger(name=None)：获取一个指定名称的日志记录器，若 name 为 None，则返回根记录器。

import logging

logger = logging.getLogger('my_logger')

setLevel(level)：设置日志记录器的最低严重级别。

logger.setLevel(logging.DEBUG)

addHandler(hdlr) 和 removeHandler(hdlr)：添加或移除处理器到日志记录器。

stream_handler = logging.StreamHandler()
logger.addHandler(stream_handler)

addFilter(filter) 和 removeFilter(filter)：添加或移除过滤器到日志记录器。

Handler 相关

StreamHandler() 和 FileHandler(filename)：分别创建控制台和文件处理器。

stream_handler = logging.StreamHandler()
file_handler = logging.FileHandler('app.log')

setFormatter(fmt)：设置处理器的格式化器。

formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)

setLevel(level)：设置处理器的最低严重级别。

stream_handler.setLevel(logging.WARNING)

Formatter 相关

Formatter(fmt=None, datefmt=None)：创建一个格式化器，可以自定义日志消息的格式和日期格式。

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

日志记录方法

debug(msg, *args, **kwargs)：记录调试级别的日志消息。

logger.debug('This is a debug message')

info(msg, *args, **kwargs)：记录信息级别的日志消息。

logger.info('This is an info message')

warning(msg, *args, **kwargs)：记录警告级别的日志消息。

logger.warning('This is a warning message')

error(msg, *args, **kwargs)：记录错误级别的日志消息。

logger.error('This is an error message')

critical(msg, *args, **kwargs)：记录严重
错误级别的日志消息。

logger.critical('This is a critical message')

exception(msg, *args, **kwargs)：记录异常信息，通常在捕获异常时使用。

try:
    result = 1 / 0
except ZeroDivisionError:
    logger.exception("Division by zero occurred!")

其他有用的方法

log(level, msg, *args, **kwargs)：根据给定的日志级别记录消息。

logger.log(logging.INFO, 'This message is logged using the log method.')

isEnabledFor(level)：检查是否启用了指定级别的日志记录。

if logger.isEnabledFor(logging.DEBUG):
    logger.debug('This debug message will be logged only if DEBUG level is enabled.')

九、实际应用案例分享

金融交易系统日志记录

在金融交易系统中，精确且全面的日志记录至关重要，它有助于监控交易流程、合规审计以及故障排查。以下是一个简化的示例：

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='financial_transaction.log'
)

logger = logging.getLogger('financial_transaction')

class Transaction:
    def __init__(self, transaction_id, amount, account_from, account_to):
        self.transaction_id = transaction_id
        self.amount = amount
        self.account_from = account_from
        self.account_to = account_to

    def execute(self):
        try:
            logger.info(f"Starting transaction {self.transaction_id}: Transferring {self.amount} from {self.account_from} to {self.account_to}")
            # 模拟交易执行
            # 这里可以添加实际的数据库操作、网络请求等
            if self.amount <= 0:
                raise ValueError("Transaction amount must be greater than zero.")
            logger.info(f"Transaction {self.transaction_id} completed successfully.")
        except Exception as e:
            logger.error(f"Transaction {self.transaction_id} failed: {e}", exc_info=True)


# 模拟创建交易
transaction = Transaction("T001", 1000, "A123", "B456")
transaction.execute()

在这个案例中，日志记录了交易的开始、执行过程中的异常以及成功完成的信息。通过将日志写入文件，方便后续的审计和分析。

物联网设备监控日志

对于物联网（IoT）设备监控系统，日志记录可以帮助管理员实时了解设备的状态、故障信息等。以下是一个简单的 IoT 设备模拟示例：

import logging
import random
import time

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger('iot_device_monitor')

class IoTDevice:
    def __init__(self, device_id):
        self.device_id = device_id
        self.temperature = 25  # 初始温度

    def read_temperature(self):
        try:
            # 模拟温度波动
            self.temperature += random.uniform(-1, 1)
            logger.info(f"Device {self.device_id} temperature: {self.temperature}°C")
            if self.temperature > 40:
                logger.warning(f"Device {self.device_id} temperature is too high!")
        except Exception as e:
            logger.error(f"Error reading temperature from device {self.device_id}: {e}")


# 模拟多个 IoT 设备
devices = [IoTDevice(f"D{i}") for i in range(5)]

while True:
    for device in devices:
        device.read_temperature()
    time.sleep(5)

在这个案例中，系统定期读取设备的温度信息并记录日志。当温度超过阈值时，会记录警告信息，方便管理员及时采取措施。

机器学习模型训练日志

在机器学习模型训练过程中，日志记录可以帮助开发者监控训练进度、评估模型性能以及发现潜在问题。以下是一个使用 scikit - learn 进行简单线性回归模型训练的示例：

import logging
from sklearn.datasets import make_regression
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger('ml_model_training')

# 生成数据集
X, y = make_regression(n_samples=1000, n_features=10, noise=0.1)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 创建线性回归模型
model = LinearRegression()

try:
    logger.info("Starting model training...")
    model.fit(X_train, y_train)
    logger.info("Model training completed.")

    # 进行预测并评估模型
    y_pred = model.predict(X_test)
    mse = mean_squared_error(y_test, y_pred)
    logger.info(f"Model evaluation: Mean Squared Error = {mse}")
except Exception as e:
    logger.error(f"Model training failed: {e}", exc_info=True)

作者：二分掌柜的