Python异常日志案例如何记录错误

wen python案例 24

本文目录导读:

Python异常日志案例如何记录错误

  1. 基础异常日志记录
  2. 使用日志配置字典
  3. 文件操作异常日志
  4. 网络请求异常日志
  5. 定时任务异常日志
  6. 自定义异常日志类
  7. 日志文件轮转

我来为你展示几种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记录
  • 字典配置:使用字典配置日志系统
  • 文件操作:处理文件相关异常
  • 网络请求:记录网络相关错误
  • 定时任务:带重试机制的日志
  • 自定义类:封装成可复用的类
  • 日志轮转:管理日志文件大小和时间

选择哪种方式取决于你的具体需求!

抱歉,评论功能暂时关闭!