本文目录导读:

我来为你展示几种Python异常日志记录的完整案例:
基础异常日志记录
import logging
import traceback
# 配置日志
logging.basicConfig(
level=logging.ERROR,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('error.log', encoding='utf-8'),
logging.StreamHandler() # 同时输出到控制台
]
)
def divide_numbers(a, b):
try:
result = a / b
return result
except ZeroDivisionError as e:
# 记录错误信息和堆栈
logging.error(f"除零错误: {e}")
logging.error(f"堆栈信息: {traceback.format_exc()}")
return None
except Exception as e:
logging.error(f"未知错误: {e}", exc_info=True)
return None
# 测试
print(divide_numbers(10, 0))
print(divide_numbers(10, 2))
使用日志配置字典
import logging
import logging.config
import traceback
LOGGING_CONFIG = {
'version': 1,
'formatters': {
'detailed': {
'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
},
'simple': {
'format': '%(levelname)s: %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'INFO',
'formatter': 'simple'
},
'file': {
'class': 'logging.FileHandler',
'filename': 'app_errors.log',
'level': 'ERROR',
'formatter': 'detailed',
'encoding': 'utf-8'
}
},
'loggers': {
'myapp': {
'level': 'DEBUG',
'handlers': ['console', 'file']
}
}
}
logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger('myapp')
class DatabaseConnectionError(Exception):
pass
def connect_to_database():
try:
# 模拟数据库连接错误
raise DatabaseConnectionError("无法连接到数据库服务器")
except DatabaseConnectionError as e:
logger.error(f"数据库连接失败: {e}")
logger.error(traceback.format_exc())
return False
except Exception as e:
logger.critical(f"严重错误: {e}", exc_info=True)
return False
connect_to_database()
文件操作异常日志
import logging
import os
from datetime import datetime
logging.basicConfig(
level=logging.ERROR,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('file_operations.log')
]
)
logger = logging.getLogger(__name__)
def read_config_file(filename):
try:
if not os.path.exists(filename):
raise FileNotFoundError(f"配置文件 {filename} 不存在")
with open(filename, 'r', encoding='utf-8') as f:
content = f.read()
return content
except FileNotFoundError as e:
logger.error(f"文件未找到: {e}")
return None
except PermissionError as e:
logger.error(f"权限不足: {e}")
return None
except Exception as e:
logger.error(f"读取文件时发生错误: {e}", exc_info=True)
return None
def write_log(data, logfile="app.log"):
try:
with open(logfile, 'a', encoding='utf-8') as f:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
f.write(f"[{timestamp}] {data}\n")
except Exception as e:
logger.error(f"写入日志失败: {e}")
# 测试
content = read_config_file("config.ini")
write_log("应用程序启动")
网络请求异常日志
import logging
import requests
import json
from requests.exceptions import RequestException
logger = logging.getLogger('network')
logger.setLevel(logging.DEBUG)
# 创建文件处理器
fh = logging.FileHandler('network_errors.log')
fh.setLevel(logging.ERROR)
# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)
logger.addHandler(fh)
def fetch_api_data(url, timeout=10):
try:
response = requests.get(url, timeout=timeout)
response.raise_for_status() # 抛出HTTP错误
data = response.json()
return data
except requests.ConnectionError as e:
logger.error(f"网络连接错误: {url} - {e}")
return None
except requests.Timeout as e:
logger.error(f"请求超时: {url} - {e}")
return None
except requests.HTTPError as e:
logger.error(f"HTTP错误 {response.status_code}: {url} - {e}")
return None
except json.JSONDecodeError as e:
logger.error(f"JSON解析错误: {e}")
return None
except RequestException as e:
logger.error(f"请求异常: {e}", exc_info=True)
return None
# 测试
result = fetch_api_data("https://api.example.com/data")
if result is None:
logger.warning("API请求失败,使用缓存数据")
定时任务异常日志
import logging
import time
import functools
from datetime import datetime
logger = logging.getLogger('scheduler')
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('scheduler.log'),
logging.StreamHandler()
]
)
def retry(max_attempts=3, delay=1):
"""重试装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempts += 1
logger.warning(f"{func.__name__} 执行失败 (尝试 {attempts}/{max_attempts}): {e}")
if attempts == max_attempts:
logger.error(f"{func.__name__} 最终失败: {e}", exc_info=True)
raise
time.sleep(delay * attempts)
return None
return wrapper
return decorator
@retry(max_attempts=3, delay=2)
def scheduled_task():
"""模拟定时任务"""
current_time = datetime.now()
# 模拟随机失败
if current_time.second % 3 == 0:
raise RuntimeError("任务执行错误")
logger.info(f"任务执行成功: {current_time}")
return True
# 模拟定时执行
for i in range(5):
try:
scheduled_task()
except Exception as e:
logger.critical(f"任务最终失败: {e}")
time.sleep(1)
自定义异常日志类
import logging
import traceback
from functools import wraps
class ExceptionLogger:
def __init__(self, logger_name=None, log_file='custom_errors.log'):
self.logger = logging.getLogger(logger_name or __name__)
# 设置处理器
handler = logging.FileHandler(log_file, encoding='utf-8')
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s\n'
'Exception Type: %(exc_type)s\n'
'File: %(filename)s\n'
'Line: %(lineno)s\n'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def log_exception(self, func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
extra = {
'exc_type': type(e).__name__,
'func_name': func.__name__,
'filename': traceback.extract_tb(exc_traceback)[-1][0],
'lineno': traceback.extract_tb(exc_traceback)[-1][1]
}
self.logger.error(
f"Error in {func.__name__}: {str(e)}",
extra=extra,
exc_info=True
)
raise
return wrapper
# 使用示例
import sys
error_logger = ExceptionLogger('myapp')
@error_logger.log_exception
def risky_operation():
result = 10 / 0 # 故意制造错误
return result
try:
risky_operation()
except:
print("错误已记录到日志文件")
日志文件轮转
import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
logger = logging.getLogger('rotating_logger')
# 按大小轮转日志
size_handler = RotatingFileHandler(
'size_based.log',
maxBytes=1024, # 1KB
backupCount=3,
encoding='utf-8'
)
# 按时间轮转日志
time_handler = TimedRotatingFileHandler(
'time_based.log',
when='midnight', # 每天午夜轮转
interval=1,
backupCount=7, # 保留7天的日志
encoding='utf-8',
atTime=None
)
formatter = logging.Formatter('%(asctime)s - %(message)s')
size_handler.setFormatter(formatter)
time_handler.setFormatter(formatter)
logger.addHandler(size_handler)
logger.addHandler(time_handler)
logger.setLevel(logging.INFO)
# 测试日志轮转
for i in range(1000):
try:
if i % 50 == 0:
raise ValueError(f"测试错误 #{i}")
logger.info(f"正常日志 #{i}")
except ValueError as e:
logger.error(f"发生错误: {e}")
这些案例覆盖了:
- 基础异常日志:简单的try-catch记录
- 字典配置:使用字典配置日志系统
- 文件操作:处理文件相关异常
- 网络请求:记录网络相关错误
- 定时任务:带重试机制的日志
- 自定义类:封装成可复用的类
- 日志轮转:管理日志文件大小和时间
选择哪种方式取决于你的具体需求!