Python脚本如何标准化同步异常处理

wen python案例 2026-06-25 253

本文目录导读：

Python脚本如何标准化同步异常处理

基础异常处理框架
装饰器实现
上下文管理器
完整的异常处理框架
使用示例
最佳实践建议

在Python中标准化同步异常处理可以通过以下几种方式实现：

基础异常处理框架

import logging
from datetime import datetime
from typing import Callable, Any
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class SyncExceptionHandler:
    """同步操作异常处理基类"""
    def __init__(self, max_retries: int = 3, retry_delay: float = 1.0):
        self.max_retries = max_retries
        self.retry_delay = retry_delay
    def handle_exception(self, func: Callable, *args, **kwargs) -> Any:
        """统一异常处理方法"""
        for attempt in range(self.max_retries):
            try:
                return func(*args, **kwargs)
            except ConnectionError as e:
                logger.error(f"连接异常 (尝试 {attempt + 1}/{self.max_retries}): {e}")
                if attempt < self.max_retries - 1:
                    self._wait_before_retry()
                else:
                    raise
            except TimeoutError as e:
                logger.error(f"超时异常 (尝试 {attempt + 1}/{self.max_retries}): {e}")
                if attempt < self.max_retries - 1:
                    self._wait_before_retry()
                else:
                    raise
            except ValueError as e:
                logger.error(f"值异常: {e}")
                raise  # 值异常不需要重试
            except Exception as e:
                logger.error(f"未知异常: {e}", exc_info=True)
                raise
    def _wait_before_retry(self):
        """重试前等待"""
        import time
        logger.info(f"等待 {self.retry_delay} 秒后重试...")
        time.sleep(self.retry_delay)

装饰器实现

from functools import wraps
import time
import logging
logger = logging.getLogger(__name__)
def sync_retry(max_retries: int = 3, retry_delay: float = 1.0):
    """同步重试装饰器"""
    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) as e:
                    last_exception = e
                    logger.warning(
                        f"重试 {attempt + 1}/{max_retries}: {func.__name__}, 错误: {e}"
                    )
                    if attempt < max_retries - 1:
                        time.sleep(retry_delay)
                except Exception as e:
                    logger.error(f"不可重试的错误: {e}")
                    raise
            raise last_exception
        return wrapper
    return decorator
# 使用示例
@sync_retry(max_retries=3, retry_delay=2.0)
def fetch_data_from_api():
    """模拟API调用"""
    import random
    if random.random() < 0.7:
        raise ConnectionError("API连接失败")
    return {"data": "success"}

上下文管理器

from contextlib import contextmanager
import logging
logger = logging.getLogger(__name__)
@contextmanager
def sync_operation_context(operation_name: str = "同步操作"):
    """同步操作上下文管理器"""
    try:
        logger.info(f"开始 {operation_name}")
        yield
        logger.info(f"{operation_name} 成功完成")
    except Exception as e:
        logger.error(f"{operation_name} 失败: {e}", exc_info=True)
        raise
    finally:
        logger.info(f"{operation_name} 结束")
# 使用示例
def process_data():
    with sync_operation_context("数据处理"):
        # 业务逻辑
        data = read_file("data.txt")
        processed = transform_data(data)
        save_result(processed)

完整的异常处理框架

import logging
from typing import Optional, Type
from enum import Enum
class ErrorSeverity(Enum):
    """错误严重性级别"""
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4
class SyncOperationResult:
    """同步操作结果封装"""
    def __init__(self, success: bool, data=None, error: Optional[Exception] = None):
        self.success = success
        self.data = data
        self.error = error
        self.timestamp = datetime.now()
    def is_success(self):
        return self.success
    def get_or_raise(self):
        if not self.success and self.error:
            raise self.error
        return self.data
class StandardSyncHandler:
    """标准化同步异常处理器"""
    def __init__(self, service_name: str = "SyncService"):
        self.service_name = service_name
        self.logger = logging.getLogger(service_name)
        self.setup_logging()
    def setup_logging(self):
        """配置日志系统"""
        handler = logging.StreamHandler()
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
    def execute_with_standard_handling(
        self,
        func: Callable,
        error_severity: ErrorSeverity = ErrorSeverity.MEDIUM,
        *args, **kwargs
    ) -> SyncOperationResult:
        """标准化的异常处理执行"""
        try:
            self.logger.info(f"执行操作: {func.__name__}")
            result = func(*args, **kwargs)
            return SyncOperationResult(success=True, data=result)
        except ValueError as e:
            self._log_error(f"参数错误: {e}", ErrorSeverity.LOW)
            return SyncOperationResult(success=False, error=e)
        except ConnectionError as e:
            self._log_error(f"连接错误: {e}", ErrorSeverity.HIGH)
            # 尝试重连
            if self._try_reconnect():
                return self.execute_with_standard_handling(
                    func, error_severity, *args, **kwargs
                )
            return SyncOperationResult(success=False, error=e)
        except TimeoutError as e:
            self._log_error(f"超时错误: {e}", ErrorSeverity.MEDIUM)
            return SyncOperationResult(success=False, error=e)
        except PermissionError as e:
            self._log_error(f"权限错误: {e}", ErrorSeverity.CRITICAL)
            return SyncOperationResult(success=False, error=e)
        except Exception as e:
            self._log_error(f"未知错误: {e}", ErrorSeverity.CRITICAL)
            return SyncOperationResult(success=False, error=e)
    def _log_error(self, message: str, severity: ErrorSeverity):
        """记录错误日志"""
        log_methods = {
            ErrorSeverity.LOW: self.logger.warning,
            ErrorSeverity.MEDIUM: self.logger.error,
            ErrorSeverity.HIGH: self.logger.error,
            ErrorSeverity.CRITICAL: self.logger.critical
        }
        log_method = log_methods.get(severity, self.logger.error)
        log_method(f"[{severity.name}] {message}")
    def _try_reconnect(self, max_attempts: int = 3) -> bool:
        """尝试重新连接"""
        for attempt in range(max_attempts):
            try:
                self.logger.info(f"重新连接尝试 {attempt + 1}/{max_attempts}")
                # 执行重连逻辑
                time.sleep(1)
                return True
            except Exception as e:
                self.logger.error(f"重连失败: {e}")
        return False

使用示例

# 示例1：基础使用
handler = StandardSyncHandler("数据同步服务")
def sync_user_data():
    # 模拟同步操作
    if random.random() < 0.5:
        raise ConnectionError("数据库连接失败")
    return {"users": ["Alice", "Bob"]}
result = handler.execute_with_standard_handling(
    sync_user_data,
    ErrorSeverity.HIGH
)
if result.is_success():
    print(f"同步成功: {result.data}")
else:
    print(f"同步失败: {result.error}")
# 示例2：装饰器使用
@sync_retry(max_retries=5, retry_delay=1.0)
def sync_to_external_service():
    # 同步到外部服务
    response = requests.post("https://api.example.com/sync", timeout=10)
    response.raise_for_status()
    return response.json()
# 示例3：上下文管理器
def batch_sync():
    with sync_operation_context("批量数据同步"):
        for item in items:
            process_item(item)

最佳实践建议

# 1. 定义自定义异常层级
class SyncError(Exception):
    """同步基础异常"""
    pass
class ConnectionSyncError(SyncError):
    """连接同步异常"""
    pass
class DataSyncError(SyncError):
    """数据同步异常"""
    pass
# 2. 使用类型提示
from typing import Union, Optional, Tuple
def sync_with_timeout(
    func: Callable[..., Any],
    timeout: float = 30.0,
    retries: int = 3
) -> Tuple[bool, Optional[Any], Optional[Exception]]:
    """带超时和重试的同步"""
    # 实现...
    pass
# 3. 统一错误码
class SyncErrorCode(enum.Enum):
    SUCCESS = 0
    CONNECTION_ERROR = 1001
    TIMEOUT_ERROR = 1002
    DATA_ERROR = 1003
    UNKNOWN_ERROR = 9999

标准化同步异常处理的要点：