Python数据API批量操作怎么设计

wen python案例 2026-07-05 225

本文目录导读：

Python数据API批量操作怎么设计

核心设计模式
技术实现方案
API服务端设计
性能优化策略
监控与容错
最佳实践总结

设计Python数据API的批量操作时，核心目标是减少网络请求次数、提高吞吐量、保证数据一致性，下面从设计模式、技术实现和最佳实践三个维度来讲解。

核心设计模式

批量接口 vs 单条接口

# ❌ 不推荐：逐个调用单条接口
for item in items:
    requests.post("/api/resource", json=item)
# ✅ 推荐：一次性批量提交
requests.post("/api/resource/batch", json={"items": items})

请求合并（Request Batching）

将多个独立请求合并为一个批量请求：

class BatchAPI:
    def __init__(self, batch_size=100, max_wait=0.5):
        self.batch_size = batch_size
        self.max_wait = max_wait
        self.queue = []
        self.lock = threading.Lock()
    def add(self, item):
        with self.lock:
            self.queue.append(item)
            if len(self.queue) >= self.batch_size:
                self.flush()
    def flush(self):
        items = self.queue[:]
        self.queue = []
        # 发送批量请求
        self._send_batch(items)

技术实现方案

方案1：简单批量提交（同步）

import requests
from typing import List, Dict
class SimpleBatchClient:
    def __init__(self, base_url: str, batch_size: int = 100):
        self.base_url = base_url
        self.batch_size = batch_size
    def bulk_create(self, items: List[Dict]) -> List[Dict]:
        """批量创建资源"""
        results = []
        for i in range(0, len(items), self.batch_size):
            batch = items[i:i + self.batch_size]
            response = requests.post(
                f"{self.base_url}/api/resources/batch",
                json={"resources": batch}
            )
            results.extend(response.json()["data"])
        return results
    def bulk_update(self, items: List[Dict]) -> List[Dict]:
        """批量更新资源"""
        results = []
        for i in range(0, len(items), self.batch_size):
            batch = items[i:i + self.batch_size]
            response = requests.put(
                f"{self.base_url}/api/resources/batch",
                json={"resources": batch}
            )
            results.extend(response.json()["data"])
        return results

方案2：异步并发批量（推荐）

import asyncio
import aiohttp
from typing import List, Dict
class AsyncBatchClient:
    def __init__(self, base_url: str, batch_size: int = 100, concurrency: int = 5):
        self.base_url = base_url
        self.batch_size = batch_size
        self.semaphore = asyncio.Semaphore(concurrency)
    async def _send_batch(self, session: aiohttp.ClientSession, 
                          items: List[Dict]) -> List[Dict]:
        async with self.semaphore:
            async with session.post(
                f"{self.base_url}/api/resources/batch",
                json={"resources": items}
            ) as resp:
                return await resp.json()
    async def bulk_create(self, items: List[Dict]) -> List[Dict]:
        """异步批量创建"""
        async with aiohttp.ClientSession() as session:
            tasks = []
            for i in range(0, len(items), self.batch_size):
                batch = items[i:i + self.batch_size]
                tasks.append(self._send_batch(session, batch))
            results = await asyncio.gather(*tasks)
            return [item for batch in results for item in batch["data"]]

方案3：智能批处理（含重试和回退）

import time
import logging
from typing import List, Dict, Callable
from tenacity import retry, stop_after_attempt, wait_exponential
class SmartBatchProcessor:
    def __init__(self, batch_size: int = 100, max_retries: int = 3):
        self.batch_size = batch_size
        self.max_retries = max_retries
        self.logger = logging.getLogger(__name__)
    @retry(stop=stop_after_attempt(3), 
           wait=wait_exponential(multiplier=1, min=2, max=10))
    def _send_request(self, items: List[Dict]) -> Dict:
        """发送批量请求（带重试机制）"""
        # 实际请求逻辑
        response = requests.post(
            "https://api.example.com/batch",
            json={"items": items},
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    def _split_by_strategy(self, items: List[Dict], 
                           strategy: str = "size") -> List[List[Dict]]:
        """根据策略拆分批次"""
        if strategy == "size":
            return [items[i:i + self.batch_size] 
                    for i in range(0, len(items), self.batch_size)]
        elif strategy == "adaptive":
            # 自适应拆分逻辑
            return self._adaptive_split(items)
    def process(self, items: List[Dict], 
                callback: Callable = None) -> List[Dict]:
        """批量处理主逻辑"""
        results = []
        failed_items = []
        batches = self._split_by_strategy(items)
        for batch in batches:
            try:
                response = self._send_request(batch)
                if callback:
                    callback(batch, response)
                results.extend(response["data"])
                # 成功批量后动态调整批次大小
                self._adjust_batch_size(len(batch), success=True)
            except Exception as e:
                self.logger.error(f"Batch failed: {e}")
                # 对失败的批次进行降级处理
                if len(batch) > 1:
                    # 拆分为更小的批次重试
                    smaller_batches = self._split_by_strategy(
                        batch, "adaptive"
                    )
                    results.extend(self.process(smaller_batches))
                else:
                    failed_items.append(batch[0])
        return results
    def _adjust_batch_size(self, batch_size: int, success: bool):
        """动态调整批次大小"""
        if success:
            self.batch_size = min(self.batch_size * 1.1, 500)
        else:
            self.batch_size = max(self.batch_size * 0.8, 10)

API服务端设计

批量端点设计

# FastAPI 批量端点示例
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
class ItemBatchRequest(BaseModel):
    items: List[dict]
    options: Optional[dict] = None
class BatchResponse(BaseModel):
    success_count: int
    failed_count: int
    results: List[dict]
    errors: List[dict] = []
@app.post("/api/resources/batch")
async def batch_create(request: ItemBatchRequest):
    """批量创建资源"""
    results = []
    errors = []
    for item in request.items:
        try:
            # 业务逻辑处理
            result = await create_resource(item)
            results.append(result)
        except Exception as e:
            errors.append({
                "item": item,
                "error": str(e)
            })
    return BatchResponse(
        success_count=len(results),
        failed_count=len(errors),
        results=results,
        errors=errors
    )

事务性批处理

from contextlib import asynccontextmanager
class TransactionalBatchProcessor:
    def __init__(self, db_session):
        self.db = db_session
    @asynccontextmanager
    async def transaction(self):
        """事务上下文管理器"""
        async with self.db.begin() as transaction:
            try:
                yield transaction
                await transaction.commit()
            except Exception:
                await transaction.rollback()
                raise
    async def process_batch(self, items: List[Dict]):
        """事务性批量处理"""
        async with self.transaction() as tx:
            # 所有操作在同一个事务中
            results = []
            for item in items:
                result = await self._process_item(tx, item)
                results.append(result)
            # 验证所有操作成功
            if not self._validate_results(results):
                raise ValueError("Batch validation failed")
        return results

性能优化策略

连接池复用

import aiohttp
class ConnectionPoolClient:
    def __init__(self):
        # 创建连接池
        connector = aiohttp.TCPConnector(
            limit=100,  # 最大连接数
            limit_per_host=20,  # 每台主机的最大连接
            ttl_dns_cache=300,  # DNS缓存时间
        )
        self.session = aiohttp.ClientSession(connector=connector)

请求压缩

import gzip
import json
def compress_request(data: dict) -> bytes:
    """压缩请求数据"""
    json_str = json.dumps(data)
    return gzip.compress(json_str.encode())
# 请求时加入压缩
headers = {"Content-Encoding": "gzip"}
response = requests.post(url, data=compress_request(data), headers=headers)

响应流式处理

async def stream_batch_response(session, url, items):
    """流式处理批量响应"""
    async with session.post(url, json={"items": items}) as resp:
        async for chunk in resp.content.iter_chunks():
            process_chunk(chunk)

监控与容错

指标收集

from prometheus_client import Counter, Histogram
class BatchMetrics:
    def __init__(self):
        self.batch_counter = Counter(
            'batch_requests_total', 
            'Total batch requests',
            ['status']
        )
        self.batch_duration = Histogram(
            'batch_processing_seconds',
            'Batch processing duration',
            buckets=[0.1, 0.5, 1, 5, 10]
        )
    def record(self, success: bool, duration: float):
        status = "success" if success else "failed"
        self.batch_counter.labels(status=status).inc()
        self.batch_duration.observe(duration)

熔断机制

from pybreaker import CircuitBreaker
class CircuitBreakerClient:
    def __init__(self):
        self.breaker = CircuitBreaker(
            fail_max=5,  # 连续失败次数
            reset_timeout=60,  # 重置超时
        )
    def batch_request(self, items):
        @self.breaker
        def do_request():
            return requests.post("/batch", json={"items": items})
        try:
            return do_request()
        except CircuitBreakerError:
            # 熔断时降级处理
            return self._fallback_sequential(items)

最佳实践总结

场景	推荐方案	注意事项
实时数据处理	异步并发+小批次	控制并发数，避免过载
离线批量同步	同步分批+大批次	设置超时和重试
混合场景	自适应批处理	动态调整批次大小
事务要求高	事务性批处理	考虑锁和死锁问题