IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

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

wen python案例 33

本文目录导读：

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

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

在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

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

  1. 统一日志格式：使用标准化的日志记录
  2. 区分异常类型：根据不同异常采取不同处理策略
  3. 实现重试机制：对可恢复的错误进行重试
  4. 封装返回结果：使用统一的结果封装对象
  5. 使用装饰器：减少重复代码
  6. 上下文管理：确保资源正确释放

这样可以在整个项目中实现一致的异常处理模式,提高代码的可维护性和可读性。

上一篇Python脚本如何自动化修复同步常规异常

下一篇Python脚本如何持续提升同步任务稳定性

相关文章

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