📢 转载信息
原文链接：https://www.kdnuggets.com/the-complete-guide-to-logging-for-python-developers
原文作者：Bala Priya C
Image by Author
# 引言
大多数Python开发者将日志记录视为事后的补救措施。他们在开发过程中到处使用print()语句，也许稍后会切换到基本的日志记录，并认为这就足够了。但当生产环境中出现问题时，他们才会意识到缺少有效诊断问题所需的上下文信息。
恰当的日志记录技术能让你洞察应用程序的行为、性能模式和错误状况。通过正确的方法，你可以在不本地重现问题的情况下，追踪用户操作、识别性能瓶颈并调试问题。良好的日志记录将调试从猜测转变为系统的解决问题过程。
本文将介绍Python开发者可以使用的基本日志记录模式。你将学习如何构建可搜索的日志消息结构、在不丢失上下文的情况下处理异常，以及为不同环境配置日志记录。我们将从基础知识开始，逐步深入到你可以立即在项目中使用的更高级的日志策略。我们将只使用logging 模块。
# 设置你的第一个日志记录器
我们不直接跳到复杂的配置，而是先理解日志记录器（logger）究竟是做什么的。我们将创建一个将日志同时写入控制台和文件的基本日志记录器。
import logging logger = logging.getLogger('my_app') logger.setLevel(logging.DEBUG) console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) file_handler = logging.FileHandler('app.log') file_handler.setLevel(logging.DEBUG) formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) console_handler.setFormatter(formatter) file_handler.setFormatter(formatter) logger.addHandler(console_handler) logger.addHandler(file_handler) logger.debug('This is a debug message') logger.info('Application started') logger.warning('Disk space running low') logger.error('Failed to connect to database') logger.critical('System shutting down')
代码的每一部分的作用如下。
getLogger()函数创建一个带名称的日志记录器实例。可以将其视为为你所有的日志创建一个通道。名称 'my_app' 有助于你在大型应用程序中识别日志来源。
我们将记录器级别设置为DEBUG，这意味着它将处理所有消息。然后我们创建两个处理器（handlers）：一个用于控制台输出，一个用于文件输出。处理器控制日志的去向。
控制台处理器只显示INFO级别及以上的日志，而文件处理器捕获所有内容，包括DEBUG消息。这很有用，因为你希望在文件中保留详细日志，而在屏幕上保持输出整洁。
格式化器（formatter）决定了你的日志消息的外观。格式字符串使用占位符，例如%(asctime)s表示时间戳，%(levelname)s表示严重性。
# 理解日志级别以及何时使用它们
Python的logging 模块有五个标准级别，了解何时使用每个级别对于生成有用的日志至关重要。
这是一个例子：
logger = logging.getLogger('payment_processor') logger.setLevel(logging.DEBUG) handler = logging.StreamHandler() handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s')) logger.addHandler(handler) def process_payment(user_id, amount): logger.debug(f'Starting payment processing for user {user_id}') if amount <= 0: logger.error(f'Invalid payment amount: {amount}') return False logger.info(f'Processing ${amount} payment for user {user_id}') if amount > 10000: logger.warning(f'Large transaction detected: ${amount}') try: # Simulate payment processing success = charge_card(user_id, amount) if success: logger.info(f'Payment successful for user {user_id}') return True else: logger.error(f'Payment failed for user {user_id}') return False except Exception as e: logger.critical(f'Payment system crashed: {e}', exc_info=True) return False def charge_card(user_id, amount): # Simulated payment logic return True process_payment(12345, 150.00) process_payment(12345, 15000.00)
我们来分解一下何时使用每个级别：
- DEBUG 用于开发期间有用的详细信息。你可以用它来记录变量值、循环迭代或分步执行跟踪。这些通常在生产环境中禁用。
- INFO 标记你想要记录的正常操作。启动服务器、完成任务或成功交易都属于此类。它们确认你的应用程序正在按预期工作。
- WARNING 信号表明发生了意料之外但尚未导致程序崩溃的事情。这包括磁盘空间不足、使用已弃用的API或不寻常但已处理的情况。应用程序会继续运行，但应该有人进行调查。
- ERROR 意味着某项操作失败，但应用程序可以继续运行。失败的数据库查询、验证错误或网络超时都属于此类。特定的操作失败了，但应用仍在运行。
- CRITICAL 表示可能导致应用程序崩溃或数据丢失的严重问题。请谨慎使用，用于需要立即关注的灾难性故障。
当你运行上述代码时，你会得到：
DEBUG: Starting payment processing for user 12345 DEBUG:payment_processor:Starting payment processing for user 12345 INFO: Processing $150.0 payment for user 12345 INFO:payment_processor:Processing $150.0 payment for user 12345 INFO: Payment successful for user 12345 INFO:payment_processor:Payment successful for user 12345 DEBUG: Starting payment processing for user 12345 DEBUG:payment_processor:Starting payment processing for user 12345 INFO: Processing $15000.0 payment for user 12345 INFO:payment_processor:Processing $15000.0 payment for user 12345 WARNING: Large transaction detected: $15000.0 WARNING:payment_processor:Large transaction detected: $15000.0 INFO: Payment successful for user 12345 INFO:payment_processor:Payment successful for user 12345 True
接下来，我们继续了解如何记录异常。
# 正确记录异常
当发生异常时，你需要的不只是错误消息；你需要完整的堆栈跟踪（stack trace）。以下是如何有效地捕获异常的方法。
import json logger = logging.getLogger('api_handler') logger.setLevel(logging.DEBUG) handler = logging.FileHandler('errors.log') formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) handler.setFormatter(formatter) logger.addHandler(handler) def fetch_user_data(user_id): logger.info(f'Fetching data for user {user_id}') try: # Simulate API call response = call_external_api(user_id) data = json.loads(response) logger.debug(f'Received data: {data}') return data except json.JSONDecodeError as e: logger.error( f'Failed to parse JSON for user {user_id}: {e}', exc_info=True ) return None except ConnectionError as e: logger.error( f'Network error while fetching user {user_id}', exc_info=True ) return None except Exception as e: logger.critical( f'Unexpected error in fetch_user_data: {e}', exc_info=True ) raise def call_external_api(user_id): # Simulated API response return '{"id": ' + str(user_id) + ', "name": "John"}' fetch_user_data(123)
这里的关键是exc_info=True参数。它告诉日志记录器在日志中包含完整的异常回溯信息。如果没有它，你只能得到错误消息，这通常不足以调试问题。
注意我们先捕获特定的异常，然后有一个通用的Exception处理器。特定的处理器允许我们提供上下文相关的错误消息。通用处理器捕获任何意外情况并重新抛出，因为我们不知道如何安全地处理它。
另外请注意，我们对预期的异常（如网络错误）记录ERROR级别，而对意外的异常记录CRITICAL级别。这种区别有助于你在审查日志时确定优先级。
# 创建可重用的日志记录器配置
在文件中复制日志设置代码既繁琐又容易出错。让我们创建一个可以在项目中任何地方导入的配置函数。
# logger_config.py import logging import os from datetime import datetime def setup_logger(name, log_dir="logs", level=logging.INFO): """ 创建一个已配置的日志记录器实例 参数: name: 日志记录器名称 (通常是调用模块的 __name__) log_dir: 存储日志文件的目录 level: 最低日志级别 返回: 配置好的日志记录器实例 """ # 如果不存在，则创建日志目录 if not os.path.exists(log_dir): os.makedirs(log_dir) logger = logging.getLogger(name) # 如果日志记录器已有处理器，则避免重复添加 if logger.handlers: return logger logger.setLevel(level) # 控制台处理器 - INFO及以上 console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) console_format = logging.Formatter("%(levelname)s - %(name)s - %(message)s") console_handler.setFormatter(console_format) # 文件处理器 - 记录所有信息 log_filename = os.path.join( log_dir, f"{name.replace('.', '_')}_{datetime.now().strftime('%Y%m%d')}.log" ) file_handler = logging.FileHandler(log_filename) file_handler.setLevel(logging.DEBUG) file_format = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s" ) file_handler.setFormatter(file_format) logger.addHandler(console_handler) logger.addHandler(file_handler) return logger
现在你已经设置好了logger_config，可以在Python脚本中这样使用它：
from logger_config import setup_logger logger = setup_logger(__name__) def calculate_discount(price, discount_percent): logger.debug(f'Calculating discount: {price} * {discount_percent}%') if discount_percent < 0 or discount_percent > 100: logger.warning(f'Invalid discount percentage: {discount_percent}') discount_percent = max(0, min(100, discount_percent)) discount = price * (discount_percent / 100) final_price = price - discount logger.info(f'Applied {discount_percent}% discount: ${price} -> ${final_price}') return final_price calculate_discount(100, 20) calculate_discount(100, 150)
这个设置函数处理了几个重要的事情。首先，如果需要，它会创建日志目录，防止因缺少目录而导致的崩溃。
该函数在添加新处理器之前会检查处理器是否已存在。如果没有这个检查，多次调用setup_logger将导致重复的日志条目。
我们自动生成带日期的日志文件名。这可以防止日志文件无限增长，并使查找特定日期的日志变得容易。
文件处理器包含比控制台处理器更多的详细信息，包括函数名和行号。这在调试时非常有价值，但会使控制台输出显得冗余。
使用__name__作为日志记录器名称创建了一个与你的模块结构相匹配的层次结构。这允许你独立控制应用程序特定部分的日志记录。
# 使用上下文构建结构化日志
纯文本日志对简单应用来说还可以，但带有上下文的结构化日志使调试更容易。让我们向日志中添加上下文信息。
import json from datetime import datetime, timezone class ContextLogger: """ 添加上下文信息到所有日志消息的日志包装器 """ def __init__(self, name, context=None): self.logger = logging.getLogger(name) self.context = context or {} handler = logging.StreamHandler() formatter = logging.Formatter('%(message)s') handler.setFormatter(formatter) # 检查处理器是否已存在以避免重复添加 if not any(isinstance(h, logging.StreamHandler) and h.formatter._fmt == '%(message)s' for h in self.logger.handlers): self.logger.addHandler(handler) self.logger.setLevel(logging.DEBUG) def _format_message(self, message, level, extra_context=None): """ 使用上下文以JSON格式化消息 """ log_data = { 'timestamp': datetime.now(timezone.utc).isoformat(), 'level': level, 'message': message, 'context': {**self.context, **(extra_context or {})} } return json.dumps(log_data) def debug(self, message, **kwargs): self.logger.debug(self._format_message(message, 'DEBUG', kwargs)) def info(self, message, **kwargs): self.logger.info(self._format_message(message, 'INFO', kwargs)) def warning(self, message, **kwargs): self.logger.warning(self._format_message(message, 'WARNING', kwargs)) def error(self, message, **kwargs): self.logger.error(self._format_message(message, 'ERROR', kwargs))
你可以这样使用ContextLogger：
def process_order(order_id, user_id): logger = ContextLogger(__name__, context={ 'order_id': order_id, 'user_id': user_id }) logger.info('Order processing started') try: items = fetch_order_items(order_id) logger.info('Items fetched', item_count=len(items)) total = calculate_total(items) logger.info('Total calculated', total=total) if total > 1000: logger.warning('High value order', total=total, flagged=True) return True except Exception as e: logger.error('Order processing failed', error=str(e)) return False def fetch_order_items(order_id): return [{'id': 1, 'price': 50}, {'id': 2, 'price': 75}] def calculate_total(items): return sum(item['price'] for item in items) process_order('ORD-12345', 'USER-789')
这个ContextLogger包装器做了一件很有用的事情：它自动将上下文包含在每个日志消息中。order_id和user_id会被添加到所有日志中，而无需在每次调用日志记录时重复它们。
JSON格式使这些日志易于解析和搜索。
每个日志方法中的**kwargs允许你为特定的日志消息添加额外上下文。这会将全局上下文（order_id、user_id）与本地上下文（item_count、total）自动组合起来。
这种模式在Web应用程序中尤其有用，在你希望请求ID、用户ID或会话ID包含在来自该请求的每个日志消息中时。
# 轮转日志文件以防止磁盘空间问题
生产环境中的日志文件会迅速增长。如果没有轮转，它们最终会填满磁盘。以下是如何实现自动日志轮转的方法。
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler def setup_rotating_logger(name): logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) # 基于大小的轮转：当文件达到10MB大小时轮转 size_handler = RotatingFileHandler( 'app_size_rotation.log', maxBytes=10 * 1024 * 1024, # 10 MB backupCount=5 # 保留5个旧文件 ) size_handler.setLevel(logging.DEBUG) # 基于时间的轮转：每天午夜轮转 time_handler = TimedRotatingFileHandler( 'app_time_rotation.log', when='midnight', interval=1, backupCount=7 # 保留7天 ) time_handler.setLevel(logging.INFO) formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) size_handler.setFormatter(formatter) time_handler.setFormatter(formatter) logger.addHandler(size_handler) logger.addHandler(time_handler) return logger logger = setup_rotating_logger('rotating_app')
现在我们尝试使用日志文件轮转：
for i in range(1000): logger.info(f'Processing record {i}') logger.debug(f'Record {i} details: completed in {i * 0.1}ms')
RotatingFileHandler根据文件大小管理日志。当日志文件达到10MB（以字节为单位指定）时，它会被重命名为app_size_rotation.log.1，并启动一个新的app_size_rotation.log文件。backupCount设置为5意味着你将保留5个旧日志文件，然后最旧的将被删除。
TimedRotatingFileHandler根据时间间隔轮转。'midnight'参数意味着它每天午夜创建一个新日志文件。你也可以使用'H'代表每小时，'D'代表每天（在任何时间），或'W0'代表周一）。
interval参数与when参数协同工作。当when='H'和interval=6时，日志将每6小时轮转一次。
这些处理器对于生产环境至关重要。没有它们，当磁盘被日志填满时，你的应用程序可能会崩溃。
# 在不同环境进行日志记录
你的日志记录需求在开发、暂存（staging）和生产环境中是不同的。以下是如何配置适应每个环境的日志记录。
import logging import os def configure_environment_logger(app_name): """ 根据环境配置日志记录器 """ environment = os.getenv('APP_ENV', 'development') logger = logging.getLogger(app_name) # 清除现有处理器 logger.handlers = [] if environment == 'development': # 开发环境：详细的控制台输出 logger.setLevel(logging.DEBUG) handler = logging.StreamHandler() handler.setLevel(logging.DEBUG) formatter = logging.Formatter( '%(levelname)s - %(name)s - %(funcName)s:%(lineno)d - %(message)s' ) handler.setFormatter(formatter) logger.addHandler(handler) elif environment == 'staging': # 暂存环境：详细的文件日志 + 重要的控制台消息 logger.setLevel(logging.DEBUG) file_handler = logging.FileHandler('staging.log') file_handler.setLevel(logging.DEBUG) file_formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s - %(message)s' ) file_handler.setFormatter(file_formatter) console_handler = logging.StreamHandler() console_handler.setLevel(logging.WARNING) console_formatter = logging.Formatter('%(levelname)s: %(message)s') console_handler.setFormatter(console_formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) elif environment == 'production': # 生产环境：结构化日志，仅错误信息输出到控制台 logger.setLevel(logging.INFO) file_handler = logging.handlers.RotatingFileHandler( 'production.log', maxBytes=50 * 1024 * 1024, # 50 MB backupCount=10 ) file_handler.setLevel(logging.INFO) file_formatter = logging.Formatter( '{"timestamp": "% (asctime)s", "level": "% (levelname)s", ' '"logger": "% (name)s", "message": "% (message)s"}' ) file_handler.setFormatter(file_formatter) console_handler = logging.StreamHandler() console_handler.setLevel(logging.ERROR) console_formatter = logging.Formatter('%(levelname)s: %(message)s') console_handler.setFormatter(console_formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger
这种基于环境的配置对每个阶段的处理方式都不同。开发环境在控制台上显示所有内容，并带有详细信息，包括函数名和行号。这使得调试速度很快。
暂存环境在开发和生产之间取得了平衡。它将详细日志写入文件以供调查，但仅在控制台上显示警告和错误，以避免噪音。
生产环境侧重于性能和结构。它只记录INFO级别及以上信息到文件中，使用JSON格式以便于解析，并实现日志轮转以管理磁盘空间。控制台输出仅限于错误信息。
# 由部署系统设置（通常） os.environ['APP_ENV'] = 'production' logger = configure_environment_logger('my_application') logger.debug('This debug message won\'t appear in production') logger.info('User logged in successfully') logger.error('Failed to process payment')
环境由APP_ENV环境变量决定。你的部署系统（Docker、Kubernetes或其他云平台）会自动设置此变量。
请注意，我们在配置前清除了现有处理器。这可以防止在应用程序生命周期中多次调用该函数时出现重复的处理器。
# 总结
良好的日志记录是快速诊断问题与花费数小时猜测哪里出错之间的区别所在。从使用适当的严重性级别的基本日志记录开始，添加结构化上下文以使日志可搜索，并配置轮转以防止磁盘空间问题。
此处展示的模式适用于任何规模的应用程序。从小处着手，使用基本日志记录，当你需要更好的可搜索性时，再添加结构化日志记录，并在部署到生产环境时实施特定于环境的配置。
祝你日志记录愉快！
Bala Priya C 是来自印度的开发者和技术作家。她喜欢在数学、编程、数据科学和内容创作的交叉点工作。她的兴趣和专业领域包括 DevOps、数据科学和自然语言处理。她喜欢阅读、写作、编码和咖啡！目前，她正致力于通过撰写教程、操作指南、观点文章等方式学习并与开发者社区分享她的知识。Bala 还创建引人入胜的资源概览和编码教程。
🚀 想要体验更好更全面的AI调用？
欢迎使用青云聚合API，约为官网价格的十分之一，支持300+全球最新模型，以及全球各种生图生视频模型，无需翻墙高速稳定，文档丰富，小白也可以简单操作。
评论区