Python批量安全案例如何保障批量操作

wen python案例 2026-06-28 235

本文目录导读：

事务性操作与原子性保证
幂等性设计
断点续传与进度保存
速率限制与退避策略
日志记录与审计
数据验证与清理
并发控制与资源管理
完整示例：安全的文件批量处理

在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批量安全的核心要点：