Pythonlogging库全面指南

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)