本文目录导读：
我来详细介绍Python脚本数据鲁棒控制的设计方法：
异常处理机制
基础异常处理
import sys
import logging
from functools import wraps
from typing import Any, Callable
def robust_decorator(func: Callable) -> Callable:
"""鲁棒性装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logging.error(f"Error in {func.__name__}: {e}")
# 返回默认值或None
return None
return wrapper
分层异常处理
class DataProcessingError(Exception):
"""自定义数据处理异常"""
pass
class DataValidationError(DataProcessingError):
"""数据验证异常"""
pass
class DataTransformationError(DataProcessingError):
"""数据转换异常"""
pass
def safe_data_processing(data: Any) -> dict:
"""安全的数据处理函数"""
try:
# 第1层：数据验证
if not isinstance(data, (list, dict)):
raise DataValidationError(f"Invalid data type: {type(data)}")
# 第2层：数据处理
processed = transform_data(data)
# 第3层：结果验证
validate_result(processed)
return processed
except DataValidationError as e:
logging.warning(f"Validation failed: {e}")
return {}
except DataTransformationError as e:
logging.error(f"Transformation failed: {e}")
return None
except Exception as e:
logging.critical(f"Unexpected error: {e}")
return None
数据验证与清洗
使用Pydantic进行数据验证
from pydantic import BaseModel, validator, Field
from typing import Optional, List
from datetime import datetime
class DataPoint(BaseModel):
"""数据点模型"""
id: int
value: float
timestamp: datetime
status: Optional[str] = "active"
@validator('value')
def validate_value(cls, v):
if v < 0 or v > 100:
raise ValueError(f"Value {v} out of valid range [0, 100]")
return v
@validator('timestamp')
def validate_timestamp(cls, v):
if v > datetime.now():
raise ValueError("Timestamp cannot be in the future")
return v
def process_data_point(raw_data: dict) -> Optional[DataPoint]:
"""安全地处理数据点"""
try:
# 清理和规范化数据
cleaned_data = {
'id': int(raw_data.get('id', 0)),
'value': float(raw_data.get('value', 0)),
'timestamp': parse_timestamp(raw_data.get('timestamp')),
'status': raw_data.get('status', 'active')
}
# 验证并返回数据点
return DataPoint(**cleaned_data)
except (ValueError, TypeError, ValidationError) as e:
logging.error(f"Invalid data point: {e}")
return None
错误重试机制
实现重试逻辑
import time
from functools import wraps
def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0):
"""重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
wait_time = delay * (backoff ** attempt)
logging.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait_time}s...")
time.sleep(wait_time)
logging.error(f"All {max_attempts} attempts failed")
raise last_exception
return wrapper
return decorator
# 使用示例
@retry(max_attempts=5, delay=0.5, backoff=1.5)
def fetch_data_from_api(url: str) -> dict:
"""从API获取数据（带重试）"""
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.json()
数据完整性检查
实现数据完整性校验
import hashlib
import json
from typing import Any
class DataIntegrityChecker:
"""数据完整性检查器"""
@staticmethod
def calculate_checksum(data: Any) -> str:
"""计算数据校验和"""
data_str = json.dumps(data, sort_keys=True)
return hashlib.sha256(data_str.encode()).hexdigest()
@staticmethod
def verify_integrity(data: Any, expected_checksum: str) -> bool:
"""验证数据完整性"""
actual_checksum = DataIntegrityChecker.calculate_checksum(data)
return actual_checksum == expected_checksum
@classmethod
def process_with_integrity_check(cls, data: Any) -> tuple:
"""处理数据并返回校验和"""
checksum = cls.calculate_checksum(data)
return data, checksum
# 使用示例
def safe_data_transfer(data: dict) -> bool:
"""安全数据传输"""
try:
# 计算校验和
data_with_hash = DataIntegrityChecker.process_with_integrity_check(data)
# 模拟网络传输
received_data, received_hash = data_with_hash
# 验证完整性
if DataIntegrityChecker.verify_integrity(received_data, received_hash):
logging.info("Data integrity verified")
return True
else:
logging.error("Data integrity check failed")
return False
except Exception as e:
logging.error(f"Data transfer failed: {e}")
return False
资源管理
使用上下文管理器
from contextlib import contextmanager
import sqlite3
@contextmanager
def safe_database_connection(db_path: str):
"""安全的数据库连接管理"""
conn = None
try:
conn = sqlite3.connect(db_path)
yield conn
except sqlite3.Error as e:
logging.error(f"Database error: {e}")
raise
finally:
if conn:
conn.close()
# 使用示例
def query_data_safely(query: str, params: tuple = None):
"""安全地查询数据库"""
with safe_database_connection('data.db') as conn:
cursor = conn.cursor()
try:
if params:
cursor.execute(query, params)
else:
cursor.execute(query)
return cursor.fetchall()
except sqlite3.Error as e:
logging.error(f"Query failed: {e}")
return []
性能监控与预防
实现监控机制
import time
from threading import Lock
class DataProcessMonitor:
"""数据处理监控器"""
def __init__(self, max_processing_time: float = 10.0):
self.max_processing_time = max_processing_time
self._lock = Lock()
self.processing_times = []
def __enter__(self):
self.start_time = time.time()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
processing_time = time.time() - self.start_time
with self._lock:
self.processing_times.append(processing_time)
# 检查是否超时
if processing_time > self.max_processing_time:
logging.warning(f"Processing time {processing_time:.2f}s exceeded limit")
# 检查平均处理时间
avg_time = sum(self.processing_times) / len(self.processing_times)
if avg_time > self.max_processing_time * 0.8:
logging.warning(f"Average processing time {avg_time:.2f}s approaching limit")
# 使用示例
def process_large_dataset(data: list) -> list:
"""处理大数据集（带性能监控）"""
with DataProcessMonitor(max_processing_time=5.0):
result = []
for item in data:
processed = process_item(item)
result.append(processed)
return result
完整的鲁棒控制框架
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Union
class DataQuality(Enum):
"""数据质量等级"""
EXCELLENT = "excellent"
GOOD = "good"
FAIR = "fair"
POOR = "poor"
@dataclass
class RobustControlConfig:
"""鲁棒控制配置"""
enable_validation: bool = True
enable_checksum: bool = True
max_retry_attempts: int = 3
retry_delay: float = 1.0
timeout_seconds: float = 30.0
quality_threshold: DataQuality = DataQuality.GOOD
class RobustDataProcessor:
"""鲁棒数据处理框架"""
def __init__(self, config: RobustControlConfig = None):
self.config = config or RobustControlConfig()
self._monitor = DataProcessMonitor()
def process(self, data: Any) -> Optional[Any]:
"""主处理逻辑"""
try:
# 阶段1：数据验证
if self.config.enable_validation:
if not self._validate_data(data):
logging.warning("Data validation failed")
return None
# 阶段2：数据转换
processed = self._transform_data(data)
# 阶段3：完整性检查
if self.config.enable_checksum:
if not self._verify_integrity(processed):
logging.error("Data integrity check failed")
return None
# 阶段4：质量评估
quality = self._assess_quality(processed)
if quality.value < self.config.quality_threshold.value:
logging.warning(f"Data quality {quality} below threshold")
return processed
except TimeoutError:
logging.error("Data processing timeout")
return None
except Exception as e:
logging.error(f"Unexpected error: {e}")
return None
def _validate_data(self, data: Any) -> bool:
"""数据验证"""
# 实现具体验证逻辑
pass
def _transform_data(self, data: Any) -> Any:
"""数据转换"""
# 实现具体转换逻辑
pass
def _verify_integrity(self, data: Any) -> bool:
"""完整性验证"""
# 实现具体完整性检查逻辑
pass
def _assess_quality(self, data: Any) -> DataQuality:
"""数据质量评估"""
# 实现具体质量评估逻辑
pass
- 多层防御：异常处理→验证→重试→完整性检查
- 失败优雅：提供默认值、备用方案
- 监控告警：超时、性能、错误率监控
- 资源管理：正确的资源释放、连接池管理
- 数据完整性：校验和、事务管理
- 可观测性：完善的日志记录、指标收集
- 降级策略：功能关闭、部分处理、默认行为
这样的设计能确保Python脚本在各种异常情况下都能保持稳定运行。
