Python脚本如何区分同步临时与永久异常

wen python案例 2026-06-24 271

本文目录导读：

基于异常类型的分类
使用重试库（推荐）
上下文感知的异常处理
异常分类器模式
实际业务场景示例
最佳实践建议

在Python中区分同步临时异常与永久异常，通常需要从异常的类型、上下文、以及业务逻辑层面进行判断,以下是几种常见的区分方法和实践策略：

基于异常类型的分类

临时（可重试）异常

import socket
import requests
from urllib.error import URLError
# 网络相关的临时异常
TEMPORARY_EXCEPTIONS = (
    ConnectionError,      # 连接错误
    TimeoutError,         # 超时
    socket.timeout,       # socket超时
    requests.ConnectionError,  # requests连接错误
    requests.Timeout,     # requests超时
    URLError,             # URL错误（通常是临时性的）
    # IO操作临时异常
    IOError,              # IO错误
    BlockingIOError,      # 阻塞IO错误
    # 数据库临时异常（SQLAlchemy示例）
    # OperationalError,   # 数据库操作错误
)

永久异常

PERMANENT_EXCEPTIONS = (
    ValueError,           # 参数错误
    TypeError,            # 类型错误
    KeyError,             # 键错误
    IndexError,           # 索引错误
    AttributeError,       # 属性错误
    NameError,            # 名称错误
    SyntaxError,          # 语法错误（通常在开发阶段）
    PermissionError,      # 权限错误（通常是永久性的）
    FileNotFoundError,    # 文件不存在
    ImportError,          # 导入错误
)

使用重试库（推荐）

from tenacity import retry, retry_if_exception_type, stop_after_attempt
# 定义临时异常列表
TEMPORARY_EXCEPTIONS = (ConnectionError, TimeoutError, IOError)
@retry(
    retry=retry_if_exception_type(TEMPORARY_EXCEPTIONS),  # 只重试临时异常
    stop=stop_after_attempt(3),  # 最多重试3次
    before_sleep=lambda retry_state: print(f"重试第{retry_state.attempt_number}次")
)
def fetch_data_from_server():
    # 模拟网络请求
    import random
    if random.random() < 0.7:  # 70%概率失败
        raise ConnectionError("网络连接失败")
    return "数据获取成功"
try:
    result = fetch_data_from_server()
    print(result)
except ConnectionError as e:
    print(f"最终失败（临时异常无法恢复）: {e}")
except ValueError as e:
    print(f"永久异常: {e}")

上下文感知的异常处理

class TemporaryException(Exception):
    """自定义临时异常"""
    pass
class PermanentException(Exception):
    """自定义永久异常"""
    pass
def process_data(data, source_type):
    """根据数据来源类型区分异常"""
    try:
        if source_type == "network":
            # 网络数据可能临时错误
            if validate_network_data(data):
                return process_network_data(data)
            else:
                raise TemporaryException("网络数据格式错误")
        elif source_type == "local":
            # 本地数据通常是永久错误
            if validate_local_data(data):
                return process_local_data(data)
            else:
                raise PermanentException("本地数据错误")
    except TemporaryException:
        # 临时异常可重试
        return retry_process(data, source_type)
    except PermanentException:
        # 永久异常需要人工介入
        log_error("永久异常，需要人工处理")
def validate_network_data(data):
    # 模拟网络数据验证
    return True
def validate_local_data(data):
    # 模拟本地数据验证
    return False

异常分类器模式

from typing import Type, Tuple
class ExceptionClassifier:
    """异常分类器"""
    def __init__(self):
        self.temporary_exceptions = []
        self.permanent_exceptions = []
    def register_temporary(self, exception_type: Type[Exception]):
        self.temporary_exceptions.append(exception_type)
    def register_permanent(self, exception_type: Type[Exception]):
        self.permanent_exceptions.append(exception_type)
    def classify(self, exception: Exception) -> str:
        """分类异常为 temporary 或 permanent"""
        exc_type = type(exception)
        if any(issubclass(exc_type, temp) for temp in self.temporary_exceptions):
            return "temporary"
        elif any(issubclass(exc_type, perm) for perm in self.permanent_exceptions):
            return "permanent"
        else:
            return "unknown"
# 使用示例
classifier = ExceptionClassifier()
# 注册临时异常
classifier.register_temporary(ConnectionError)
classifier.register_temporary(TimeoutError)
classifier.register_temporary(IOError)
# 注册永久异常
classifier.register_permanent(ValueError)
classifier.register_permanent(TypeError)
classifier.register_permanent(KeyError)
def handle_exception(exception):
    classification = classifier.classify(exception)
    if classification == "temporary":
        print(f"临时异常（可重试）: {exception}")
        # 执行重试逻辑
    elif classification == "permanent":
        print(f"永久异常（不可重试）: {exception}")
        # 记录日志，通知管理员
    else:
        print(f"未知异常类型: {exception}")
        # 保守处理
# 测试
try:
    1 / 0  # ZeroDivisionError
except Exception as e:
    handle_exception(e)  # 输出：未知异常类型

实际业务场景示例

import time
from functools import wraps
def retry_on_temporary(max_retries=3, delay=1):
    """装饰器：只重试临时异常"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (ConnectionError, TimeoutError, IOError) as e:
                    # 临时异常，进行重试
                    print(f"临时异常（尝试{attempt + 1}/{max_retries}）: {e}")
                    last_exception = e
                    time.sleep(delay * (attempt + 1))  # 指数退避
                except (ValueError, TypeError, KeyError) as e:
                    # 永久异常，立即失败
                    raise PermanentError(f"永久异常: {e}")
            # 所有重试都失败
            raise TemporaryError(f"重试{max_retries}次后失败: {last_exception}")
        return wrapper
    return decorator
# 使用示例
@retry_on_temporary(max_retries=3, delay=0.5)
def fetch_user_data(user_id):
    """获取用户数据"""
    import random
    if random.random() < 0.6:
        raise ConnectionError("数据库连接失败")
    if user_id <= 0:
        raise ValueError("用户ID必须大于0")
    return {"id": user_id, "name": f"User_{user_id}"}
# 测试
try:
    data = fetch_user_data(-1)  # 永久异常
except PermanentError as e:
    print(f"永久异常: {e}")
try:
    data = fetch_user_data(1)  # 可能临时异常和重试
except TemporaryError as e:
    print(f"临时异常最终失败: {e}")