本文目录导读：
在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的分页机制、数据量大小和业务需求，对于大规模数据，建议使用异步并发方式提高效率，同时做好错误处理和重试机制。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
