本文目录导读：
在Python中进行批量安全操作（例如批量文件操作、批量网络请求、批量数据库操作等）时，保障安全性和可靠性是非常关键的，以下是一些核心策略和实践，可以帮助你实现“批量安全”：
事务性操作与原子性保证
核心思路：要么全部成功，要么全部回滚到初始状态。
import logging
from contextlib import contextmanager
@contextmanager
def batch_operation(operations):
"""批量操作的原子性上下文管理器"""
completed = []
try:
for op in operations:
result = op()
completed.append(result)
yield completed
except Exception as e:
# 回滚已完成的操作
for op in reversed(completed):
try:
op.rollback() # 假设每个操作有rollback方法
except Exception as rollback_error:
logging.error(f"回滚失败: {rollback_error}")
raise e
# 使用示例
def safe_batch_process():
ops = [operation1, operation2, operation3]
with batch_operation(ops) as results:
# 所有操作成功完成
pass
幂等性设计
核心思路：相同的操作执行多次，结果应该相同。
import hashlib
import json
class IdempotentOperation:
def __init__(self, operation_id, data):
self.operation_id = operation_id
self.data = data
self.executed = False
self.result = None
def execute(self):
if not self.executed:
self.result = self._do_execute()
self.executed = True
return self.result
def _do_execute(self):
# 实际的业务逻辑
return f"Processed {self.data}"
# 追踪已执行的操作
def is_duplicate(operation_id, executed_set):
return operation_id in executed_set
断点续传与进度保存
核心思路：批量操作失败时，能从失败点继续。
import pickle
import os
class BatchProcessor:
def __init__(self, checkpoint_file="batch_checkpoint.pkl"):
self.checkpoint_file = checkpoint_file
self.progress = self.load_checkpoint()
def load_checkpoint(self):
if os.path.exists(self.checkpoint_file):
with open(self.checkpoint_file, 'rb') as f:
return pickle.load(f)
return {"completed": [], "failed": []}
def save_checkpoint(self):
with open(self.checkpoint_file, 'wb') as f:
pickle.dump(self.progress, f)
def process_batch(self, items):
for i, item in enumerate(items):
if i in self.progress["completed"]:
continue # 跳过已完成的
try:
result = self.process_item(item)
self.progress["completed"].append(i)
self.save_checkpoint() # 每完成一个就保存
except Exception as e:
self.progress["failed"].append((i, str(e)))
self.save_checkpoint()
raise e # 或选择继续处理下一个
速率限制与退避策略
核心思路：防止系统过载，优雅处理错误。
import time
import random
from functools import wraps
def rate_limiter(max_per_second=10):
"""速率限制装饰器"""
last_called = [0.0]
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
wait_time = 1.0 / max_per_second - elapsed
if wait_time > 0:
time.sleep(wait_time)
result = func(*args, **kwargs)
last_called[0] = time.time()
return result
return wrapper
return decorator
def retry_with_backoff(max_retries=3, base_delay=1, max_delay=10):
"""带退避的重试机制"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise e
delay = min(base_delay * (2 ** attempt), max_delay)
delay += random.uniform(0, 0.1 * delay) # 抖动
time.sleep(delay)
return None
return wrapper
return decorator
# 使用示例
@rate_limiter(max_per_second=5)
@retry_with_backoff(max_retries=3)
def batch_api_call(data):
# 实际API调用
pass
日志记录与审计
核心思路：记录每个操作的详细日志，便于事后排查。
import logging
import json
from datetime import datetime
class AuditLogger:
def __init__(self, log_file="batch_audit.log"):
logging.basicConfig(
filename=log_file,
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def log_operation(self, operation_id, status, details=None):
log_entry = {
"timestamp": datetime.now().isoformat(),
"operation_id": operation_id,
"status": status, # STARTED, COMPLETED, FAILED
"details": details
}
self.logger.info(json.dumps(log_entry))
def log_batch_summary(self, total, success, failed):
summary = {
"timestamp": datetime.now().isoformat(),
"total": total,
"success": success,
"failed": failed,
"duration": None # 可以计算实际耗时
}
self.logger.info(f"BATCH_SUMMARY: {json.dumps(summary)}")
数据验证与清理
核心思路：在操作前验证数据，清理风险内容。
import bleach
from pydantic import BaseModel, validator
class BatchItem(BaseModel):
id: int
name: str
email: str
@validator('name')
def sanitize_name(cls, v):
# 清理潜在的XSS攻击
return bleach.clean(v, tags=[], strip=True)
@validator('email')
def validate_email(cls, v):
# 简单的邮箱验证
if '@' not in v:
raise ValueError('Invalid email format')
return v
def validate_batch_data(items):
valid_items = []
for item in items:
try:
validated = BatchItem(**item)
valid_items.append(validated)
except Exception as e:
logging.error(f"验证失败: {item} - {e}")
return valid_items
并发控制与资源管理
核心思路：限制并发数，防止资源耗尽。
import asyncio
from asyncio import Semaphore
class SafeBatchProcessor:
def __init__(self, max_concurrent=10):
self.semaphore = Semaphore(max_concurrent)
async def process_safe(self, item):
async with self.semaphore:
# 实际的异步处理
await asyncio.sleep(1)
return f"Processed {item}"
async def process_batch(self, items):
tasks = [self.process_safe(item) for item in items]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# 线程安全版本
from threading import BoundedSemaphore
class ThreadSafeProcessor:
def __init__(self, max_concurrent=5):
self.semaphore = BoundedSemaphore(max_concurrent)
def process(self, item):
with self.semaphore:
# 实际的同步处理
pass
完整示例：安全的文件批量处理
import os
import shutil
import hashlib
from pathlib import Path
class SafeFileBatchProcessor:
def __init__(self, source_dir, dest_dir, checkpoint_file="file_batch.checkpoint"):
self.source_dir = Path(source_dir)
self.dest_dir = Path(dest_dir)
self.checkpoint_file = checkpoint_file
self.processed = set()
self.failed = {}
def file_checksum(self, filepath):
"""计算文件校验和"""
hasher = hashlib.sha256()
with open(filepath, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
hasher.update(chunk)
return hasher.hexdigest()
def safe_copy(self, src, dst):
"""安全的文件复制，包含验证"""
# 创建目标目录
dst.parent.mkdir(parents=True, exist_ok=True)
# 如果目标已存在，验证完整性
if dst.exists():
if self.file_checksum(src) == self.file_checksum(dst):
return True # 已完成
# 复制文件
shutil.copy2(src, dst)
# 验证复制结果
if self.file_checksum(src) != self.file_checksum(dst):
raise ValueError(f"文件复制后校验和不匹配: {src}")
return True
def process_batch(self):
files = list(self.source_dir.rglob('*.*')) # 递归获取所有文件
for file in files:
if str(file) in self.processed:
continue
try:
dest = self.dest_dir / file.relative_to(self.source_dir)
self.safe_copy(file, dest)
self.processed.add(str(file))
self.save_progress()
except Exception as e:
self.failed[str(file)] = str(e)
self.save_progress()
# 可以选择继续处理或停止
return {
"total": len(files),
"processed": len(self.processed),
"failed": self.failed
}
def save_progress(self):
with open(self.checkpoint_file, 'w') as f:
f.write(f"{json.dumps(list(self.processed))}\n")
f.write(json.dumps(self.failed))
def load_progress(self):
if os.path.exists(self.checkpoint_file):
with open(self.checkpoint_file, 'r') as f:
lines = f.readlines()
if lines:
self.processed = set(json.loads(lines[0]))
if len(lines) > 1:
self.failed = json.loads(lines[1])
保障Python批量安全的核心要点：
- 事务性：确保操作的原子性
- 幂等性：重复执行结果一致
- 断点续传：支持从失败点恢复
- 速率控制：避免系统过载
- 详细日志：便于审计和排错
- 数据验证：防止恶意数据注入
- 资源管理：控制并发和内存使用
根据具体场景选择合适的策略组合,并在生产环境中充分测试边界情况。
