如何用 python 快速拉取 shopify 订单数据

场景： 拉取一个店铺的所有订单数据 字段较多 后台的导出csv的方式字段不足 因此需要调接口拉取数据

如果订单数据较多 拉去下来还是非常耗费时间的，这里只是提供一种快速的拉取订单的办法

GraphQL 分页查询

接口就是使用 GraphQL orders 的查询，接口本就可以分页查，可以使用 first 和 after进行分页。然后接口是按顺序的返回数据的

query getAllOrders($first: Int, $after: String, $query: String){
    orders (first: $first, after: $after, query: $query, reverse:true){
        edges{
            node{
                id
                name
        }
        pageInfo {
            hasNextPage
            endCursor
        }
    }
}

在数据较多的情况下，本来就是网络IO最耗时，而且 shopify 会限制这种api的调用速度的，所以一般我都会设置等待几秒，这样一页一页查下来也是很慢的。

而且 一般情况 都是会想到用多线程加速的 比如同时调用多个接口 但是问题来了 只用当传具体的页码的时候才方便用多线程 但是这里接口的设计不是用页码的 是用 after 参数的。

after 是一个 cursor , 就是游标，

下一页的内容是要从

pageInfo { hasNextPage endCursor ———→ 这是下一页 after 的入参 }

这样就不好用多线程来同时请求多页了，但是没有关系 还有办法

query 参数结合多线程

注意上面的 查询语句 除了 first 和 after 之外 ，还有参数 query

query 就是用来过滤数据的， 这个参数允许传一个字符串来表示过滤，具体看文档

如果能使用 过滤 将每个线程的查询分开，每个线程查询一部分数据，且能保证数据都被查到，就可以做到使用多个线程来同时查询数据了！

对于订单数据，就可以这么干！我们只要通过 query 字段将订单按时间段划分，分为年或者月，这样就好啦！

下面这段代码 通过 month_ranges 方法生成12个月的 query 字段的前后时间点用于生成 query.

然后使用线程池进行处理，这样一次运行拉取一年的数据。

实测下来 五年大概几万条订单数据 我一年一年的处理 也才用了20分钟不到（当然也可以一次性处理）

from datetime import date, timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed
from orders import *

source_store = "teamwang"


def month_ranges(year):
    ranges = []
    for month in range(1, 13):
        start = date(year, month, 1)
        if month == 12:
            end = date(year, 12, 31)
        else:
            end = date(year, month + 1, 1) - timedelta(days=1)
        ranges.append((start, end))
    return ranges

year = 2025

with ThreadPoolExecutor(max_workers=5) as executor:
    results = []
    futures = []

    # 提交任务
    for start, end in month_ranges(year):
        print(f"Fetching {start} ~ {end} ...")

        # 文件命名改为 "orders_YYYY-MM.csv"
        csv_file = f"orders/orders_{start.strftime('%Y-%m')}.csv"

        # Shopify 查询语句
        query = f'created_at:>={start} created_at:<={end}'

        init_csv_file(csv_file)

        loader = ShopifyDataLoader(
            env=EnvLoader(source_store),
            query_gql=get_graphql_query("getAllOrders"),
            page_size=50,
            filters=query
        )

        futures.append(executor.submit(download_orders, loader=loader, csv_file=csv_file))

    for future in as_completed(futures):
        try:
            results.append(future.result())
        except Exception as e:
            print(f"线程运行出错: {e}")

需要注意的是：如果要保存至表格中，肯定会生成每个月的表格 最后可能需要合并起来。

（数据拉取下来后的保存也是非常耗时的，要注意线程安全）