如何用Python实现e-conomic API的千行分页跳过逻辑

核心思路 #

针对e-conomic API单次最多返回1000条数据的限制，需按以下逻辑实现全量数据获取：

1. 首次调用API获取总数据量，计算所需分页次数（总条数÷1000，向上取整）
2. 循环遍历每一页，通过skippages参数跳过已获取的页数，依次拉取所有分页数据
3. 合并所有分页数据后，写入目标存储（Excel/SQL）

修改后的Account API全量获取代码（写入Excel） #

基于你提供的代码，添加完整分页逻辑：

# 调用e-conomic的Account API并获取全部分页数据，写入Excel
import requests
import pandas as pd

# API配置
HEADERS = {
    "X-AgreementGrantToken": "demo",
    "X-AppSecretToken": "demo",
}
BASE_URL = "https://restapi.e-conomic.com"
PAGE_SIZE = 1000

def get_api_page(endpoint, skip_pages):
    """获取指定页码的API数据"""
    query = {
        "pagesize": str(PAGE_SIZE),
        "skippages": str(skip_pages)
    }
    response = requests.get(f"{BASE_URL}/{endpoint}", headers=HEADERS, params=query)
    response.raise_for_status()  # 捕获API请求错误
    return response.json()

def get_all_api_data(endpoint):
    """获取指定API端点的所有分页数据"""
    # 先获取第一页，同时拿到总数据量
    first_page = get_api_page(endpoint, skip_pages=0)
    total_items = first_page.get("pagination", {}).get("totalItems", len(first_page["collection"]))
    all_data = first_page["collection"].copy()
    
    # 计算总页数（向上取整）
    total_pages = (total_items + PAGE_SIZE - 1) // PAGE_SIZE
    
    # 循环拉取剩余页面
    for page in range(1, total_pages):
        current_page = get_api_page(endpoint, skip_pages=page)
        all_data.extend(current_page["collection"])
        
    return all_data

# 执行全量获取并写入Excel
all_accounts = get_all_api_data("accounts")
dataset = pd.DataFrame(all_accounts)
filtered_data = dataset[['accountNumber','accountType','name']]
filtered_data.to_excel('AccountDataAuto.xlsx', index=False)

修改后的Entries API全量获取代码（写入SQL） #

同样为Entries API补充分页逻辑：

# 调用e-conomic的Entries API并获取全部分页数据，写入SQL
import requests
import pandas as pd
import sqlalchemy.engine as sqle

HEADERS = {         
    "X-AgreementGrantToken": "demo",
    "X-AppSecretToken": "demo",
}
BASE_URL = "https://restapi.e-conomic.com"
PAGE_SIZE = 1000

def get_api_page(url, skip_pages):
    """获取指定页码的API数据"""
    query = {
        "pagesize": str(PAGE_SIZE),
        "skippages": str(skip_pages)
    }
    response = requests.get(url, headers=HEADERS, params=query)
    response.raise_for_status()
    return response.json()

def get_all_entries_data(entries_url):
    """获取所有分页的条目数据"""
    first_page = get_api_page(entries_url, skip_pages=0)
    total_items = first_page.get("pagination", {}).get("totalItems", len(first_page["collection"]))
    all_entries = first_page["collection"].copy()
    
    total_pages = (total_items + PAGE_SIZE - 1) // PAGE_SIZE
    
    for page in range(1, total_pages):
        current_page = get_api_page(entries_url, skip_pages=page)
        all_entries.extend(current_page["collection"])
        
    return all_entries

def get_db_engine():
    conn_str = "DRIVER={SQL SERVER};SERVER=JAKOB-MSI;DATABASE=MightyMonday;TRUSTED_CONNECTION=yes"
    conn_url = sqle.URL.create("mssql+pyodbc", query={'odbc_connect': conn_str})
    return sqle.create_engine(conn_url)

# 获取会计年度对应的条目URL并拉取全量数据
source = requests.get(f"{BASE_URL}/accounting-years", headers=HEADERS).json()
entries_url = source["collection"][0]["entries"]
all_entries = get_all_entries_data(entries_url)

# 处理字段并写入SQL
dataset = pd.DataFrame(all_entries)
dataset['accountNumber'] = [d.get('accountNumber') for d in dataset.account]
filtered_dataset = dataset[['accountNumber', 'amountInBaseCurrency', 'date', 'text']]
filtered_dataset.to_sql('Entries_demo_values', con=get_db_engine(), if_exists='replace', index=False)

关键说明 #

• 分页计算：通过API返回的pagination.totalItems字段获取总数据量，确保分页次数准确；若API未返回该字段，则用第一页数据量估算
• 错误处理：添加response.raise_for_status()捕获请求失败场景，便于调试
• 代码复用：将分页逻辑封装为独立函数，可直接复用在不同API端点

内容的提问来源于stack exchange，提问作者Jakob