本文目录导读：
在Python中实现数据API的断点续传,主要涉及到如何记录已获取的数据位置和如何从断点处继续请求，以下提供几种常见场景的实现方式：
基于Offset/Page的断点续传
适用于分页API（如list接口），通过记录页码或偏移量实现。
import requests
import json
from pathlib import Path
class OffsetBasedResumableFetcher:
def __init__(self, checkpoint_file="checkpoint.json"):
self.checkpoint_file = Path(checkpoint_file)
self.checkpoint = self.load_checkpoint()
def load_checkpoint(self):
"""加载检查点"""
if self.checkpoint_file.exists():
with open(self.checkpoint_file, 'r') as f:
return json.load(f)
return {"next_offset": 0, "completed": False}
def save_checkpoint(self, offset, completed=False):
"""保存检查点"""
with open(self.checkpoint_file, 'w') as f:
json.dump({
"next_offset": offset,
"completed": completed
}, f)
def fetch_page(self, url, offset, limit=100):
"""获取单页数据"""
params = {"offset": offset, "limit": limit}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
return data.get("results", []), data.get("total", 0)
def fetch_all_data(self, url, batch_size=100):
"""主获取逻辑"""
offset = self.checkpoint["next_offset"]
while not self.checkpoint["completed"]:
try:
results, total = self.fetch_page(url, offset, batch_size)
if not results:
break
# 处理数据（示例：保存到文件）
self.process_data(results)
# 更新检查点
offset += len(results)
self.save_checkpoint(offset)
# 判断是否完成
if offset >= total:
self.save_checkpoint(offset, completed=True)
print("所有数据获取完成")
break
except (requests.RequestException, json.JSONDecodeError) as e:
print(f"获取数据失败，将在offset={offset}处重试: {e}")
self.save_checkpoint(offset)
raise
def process_data(self, data):
"""处理数据（示例：打印数据条数）"""
print(f"处理了 {len(data)} 条数据")
基于Cursor/Token的断点续传
适用于游标分页API（如Twitter、某些GraphQL API）。
import requests
import json
from pathlib import Path
class CursorBasedResumableFetcher:
def __init__(self, checkpoint_file="cursor_checkpoint.json"):
self.checkpoint_file = Path(checkpoint_file)
self.checkpoint = self.load_checkpoint()
def load_checkpoint(self):
if self.checkpoint_file.exists():
with open(self.checkpoint_file, 'r') as f:
return json.load(f)
return {"next_cursor": None, "completed": False}
def save_checkpoint(self, cursor, completed=False):
with open(self.checkpoint_file, 'w') as f:
json.dump({
"next_cursor": cursor,
"completed": completed
}, f)
def fetch_page(self, url, cursor=None, limit=100):
"""获取带游标的分页数据"""
params = {"limit": limit}
if cursor:
params["cursor"] = cursor
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
# 假设API返回格式为：
# { "data": [...], "next_cursor": "xxx", "has_more": true }
return data.get("data", []), data.get("next_cursor"), data.get("has_more", False)
def fetch_all_data(self, url, batch_size=100):
cursor = self.checkpoint["next_cursor"]
while not self.checkpoint["completed"]:
try:
results, next_cursor, has_more = self.fetch_page(url, cursor, batch_size)
if not results:
break
# 处理数据
self.process_data(results)
# 更新游标
cursor = next_cursor
self.save_checkpoint(cursor)
if not has_more:
self.save_checkpoint(cursor, completed=True)
print("所有数据获取完成")
break
except Exception as e:
print(f"获取数据失败，将在cursor={cursor}处重试: {e}")
self.save_checkpoint(cursor)
raise
def process_data(self, data):
print(f"处理了 {len(data)} 条数据")
基于时间范围的断点续传
适用于时间序列数据（如日志、事件流）。
import requests
import json
from datetime import datetime, timedelta
from pathlib import Path
class TimeBasedResumableFetcher:
def __init__(self, checkpoint_file="time_checkpoint.json"):
self.checkpoint_file = Path(checkpoint_file)
self.checkpoint = self.load_checkpoint()
def load_checkpoint(self):
if self.checkpoint_file.exists():
with open(self.checkpoint_file, 'r') as f:
return json.load(f)
return {
"last_fetch_time": (datetime.now() - timedelta(days=1)).isoformat(),
"completed": False
}
def save_checkpoint(self, last_time, completed=False):
with open(self.checkpoint_file, 'w') as f:
json.dump({
"last_fetch_time": last_time,
"completed": completed
}, f)
def fetch_time_range(self, url, start_time, end_time, limit=100):
"""按时间范围获取数据"""
params = {
"start_time": start_time,
"end_time": end_time,
"limit": limit
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
return data.get("results", []), data.get("has_more", False)
def fetch_all_data(self, url, batch_size=100, time_window_minutes=60):
last_time = datetime.fromisoformat(self.checkpoint["last_fetch_time"])
now = datetime.now()
while last_time < now and not self.checkpoint["completed"]:
end_time = min(last_time + timedelta(minutes=time_window_minutes), now)
try:
results, has_more = self.fetch_time_range(
url,
last_time.isoformat(),
end_time.isoformat(),
batch_size
)
if results:
self.process_data(results)
# 更新时间戳
last_time = end_time
self.save_checkpoint(last_time.isoformat())
if last_time >= now:
self.save_checkpoint(last_time.isoformat(), completed=True)
print("所有数据获取完成")
except Exception as e:
print(f"获取时间范围 {last_time} - {end_time} 失败: {e}")
self.save_checkpoint(last_time.isoformat())
raise
def process_data(self, data):
print(f"处理了 {len(data)} 条数据")
通用断点续传框架
适用于各种场景的通用解决方案：
import requests
import json
import hashlib
from pathlib import Path
from typing import Optional, Dict, Any
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ResumableFetcher:
def __init__(self,
api_base_url: str,
checkpoint_dir: str = "checkpoints",
session: Optional[requests.Session] = None):
self.api_base_url = api_base_url
self.checkpoint_dir = Path(checkpoint_dir)
self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
self.session = session or requests.Session()
# 生成唯一的检查点文件名（基于API URL的哈希）
self.checkpoint_name = hashlib.md5(api_base_url.encode()).hexdigest() + ".json"
self.checkpoint_file = self.checkpoint_dir / self.checkpoint_name
self.checkpoint = self.load_checkpoint()
def load_checkpoint(self) -> Dict[str, Any]:
"""加载或初始化检查点"""
if self.checkpoint_file.exists():
try:
with open(self.checkpoint_file, 'r') as f:
return json.load(f)
except (json.JSONDecodeError, IOError) as e:
logger.error(f"检查点文件损坏，将重新开始: {e}")
return {}
return {}
def save_checkpoint(self, **kwargs):
"""保存检查点"""
try:
self.checkpoint.update(kwargs)
with open(self.checkpoint_file, 'w') as f:
json.dump(self.checkpoint, f, indent=2, default=str)
logger.debug(f"检查点已保存: {self.checkpoint}")
except Exception as e:
logger.error(f"保存检查点失败: {e}")
def fetch(self, endpoint: str, params: Optional[Dict] = None,
pagination_type: str = "offset", **pagination_params):
"""
通用数据获取方法
Args:
endpoint: API端点
params: 额外的查询参数
pagination_type: 分页类型 ("offset", "cursor", "time")
**pagination_params: 分页相关参数
"""
url = f"{self.api_base_url}/{endpoint.lstrip('/')}"
paginators = {
"offset": self._fetch_with_offset,
"cursor": self._fetch_with_cursor,
"time": self._fetch_with_time
}
paginator = paginators.get(pagination_type)
if not paginator:
raise ValueError(f"不支持的分页类型: {pagination_type}")
return paginator(url, params or {}, **pagination_params)
def _fetch_with_offset(self, url: str, params: Dict, **kwargs):
"""基于偏移量的分页获取"""
limit = kwargs.get("limit", 100)
max_retries = kwargs.get("max_retries", 3)
offset = self.checkpoint.get("next_offset", 0)
while True:
page_params = {**params, "offset": offset, "limit": limit}
for attempt in range(max_retries):
try:
response = self.session.get(url, params=page_params, timeout=30)
response.raise_for_status()
data = response.json()
break
except Exception as e:
if attempt == max_retries - 1:
logger.error(f"获取数据失败，offset={offset}: {e}")
raise
logger.warning(f"重试 {attempt + 1}/{max_retries}: {e}")
results = data.get("results", data.get("data", []))
if not results:
break
yield results
offset += len(results)
self.save_checkpoint(next_offset=offset)
total = data.get("total", data.get("count", float('inf')))
if offset >= total:
self.save_checkpoint(next_offset=offset, completed=True)
break
def _fetch_with_cursor(self, url: str, params: Dict, **kwargs):
"""基于游标的分页获取"""
limit = kwargs.get("limit", 100)
cursor_field = kwargs.get("cursor_field", "next_cursor")
cursor = self.checkpoint.get("next_cursor")
while True:
page_params = {**params, "limit": limit}
if cursor:
page_params[cursor_field] = cursor
response = self.session.get(url, params=page_params, timeout=30)
response.raise_for_status()
data = response.json()
results = data.get("results", data.get("data", []))
if not results:
break
yield results
cursor = data.get("next_cursor", data.get("cursor"))
self.save_checkpoint(next_cursor=cursor)
if not data.get("has_more", True):
self.save_checkpoint(next_cursor=cursor, completed=True)
break
def _fetch_with_time(self, url: str, params: Dict, **kwargs):
"""基于时间范围的分页获取"""
start_time = self.checkpoint.get("last_fetch_time",
(datetime.now() - timedelta(days=1)).isoformat())
time_field = kwargs.get("time_field", "start_time")
page_params = {**params, time_field: start_time}
response = self.session.get(url, params=page_params, timeout=30)
response.raise_for_status()
data = response.json()
results = data.get("results", data.get("data", []))
if results:
yield results
last_time = results[-1].get("timestamp")
if last_time:
self.save_checkpoint(last_fetch_time=last_time)
self.save_checkpoint(completed=True)
# 使用示例
if __name__ == "__main__":
# 示例1：基于Offset的分页
fetcher1 = ResumableFetcher("https://api.example.com")
for batch in fetcher1.fetch("/users", pagination_type="offset", limit=50):
print(f"获取到 {len(batch)} 条用户数据")
# 示例2：基于Cursor的分页
fetcher2 = ResumableFetcher("https://api.example.com")
for batch in fetcher2.fetch("/tweets", pagination_type="cursor", limit=100):
print(f"获取到 {len(batch)} 条推文数据")
# 示例3：基于时间的分页
fetcher3 = ResumableFetcher("https://api.example.com")
for batch in fetcher3.fetch("/events", pagination_type="time", limit=200):
print(f"获取到 {len(batch)} 条事件数据")
高级特性：并发断点续传
使用aiohttp实现异步并发获取：
import aiohttp
import asyncio
import json
from pathlib import Path
class AsyncResumableFetcher:
def __init__(self, checkpoint_file="async_checkpoint.json", max_workers=5):
self.checkpoint_file = Path(checkpoint_file)
self.checkpoint = self.load_checkpoint()
self.max_workers = max_workers
self.semaphore = asyncio.Semaphore(max_workers)
def load_checkpoint(self):
if self.checkpoint_file.exists():
with open(self.checkpoint_file, 'r') as f:
return json.load(f)
return {"completed_pages": [], "in_progress": []}
def save_checkpoint(self):
with open(self.checkpoint_file, 'w') as f:
json.dump(self.checkpoint, f, indent=2)
async def fetch_page(self, session, url, page_num):
"""异步获取单页数据"""
async with self.semaphore:
params = {"page": page_num, "per_page": 100}
async with session.get(url, params=params) as response:
data = await response.json()
return page_num, data.get("results", [])
async def fetch_all_pages(self, url, total_pages):
"""并发获取所有页面"""
async with aiohttp.ClientSession() as session:
# 过滤掉已完成的页面
remaining_pages = [
p for p in range(1, total_pages + 1)
if p not in self.checkpoint["completed_pages"]
]
tasks = []
for page_num in remaining_pages:
task = self.fetch_page(session, url, page_num)
tasks.append(task)
# 使用gather并发执行，但会捕获异常
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
logger.error(f"页面获取失败: {result}")
continue
page_num, data = result
if data:
self.process_data(data)
self.checkpoint["completed_pages"].append(page_num)
self.save_checkpoint()
def process_data(self, data):
print(f"处理了 {len(data)} 条数据")
- 检查点文件：使用JSON文件保存当前进度，支持程序重启后恢复
- 原子性操作：处理完数据后再更新检查点，防止数据重复
- 错误处理：捕获网络异常，保存当前进度以便重试
- 数据一致性：确保处理完成的数据已持久化后，再更新检查点
- 幂等性：API调用和数据处理逻辑应支持重复执行
选择哪种方式取决于API的分页机制、数据量大小和业务需求，对于大规模数据，建议使用异步并发方式提高效率，同时做好错误处理和重试机制。
