本文目录导读:

我来为你展示几种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记录
- 字典配置:使用字典配置日志系统
- 文件操作:处理文件相关异常
- 网络请求:记录网络相关错误
- 定时任务:带重试机制的日志
- 自定义类:封装成可复用的类
- 日志轮转:管理日志文件大小和时间
选择哪种方式取决于你的具体需求!
版权声明:NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播