如何使用 asyncio 和 aiohttp 异步通过 api 响应进行分页

Posted 2023-03-27

【中文标题】如何使用 asyncio 和 aiohttp 异步通过 api 响应进行分页

【英文标题】：How paginate through api response asynchronously with asyncio and aiohttp

【发布时间】：2019-05-22 02:49:58

【问题描述】：

我正在尝试使用 python 异步进行 api 调用。我在一个列表中有多个端点，每个端点都会返回分页结果。我可以设置异步通过多个端点，但是无法返回每个端点的分页结果。

通过调试，我发现fetch_more() 函数运行while 循环，但实际上并没有通过session.get() 的异步。所以基本上。函数 fetch_more() 旨在从每个端点的 api 调用中获取剩余结果，但是我发现无论是否使用 fetch_more() 函数都会返回相同数量的结果。我尝试寻找使用 asyncio 进行分页的示例，但运气不佳。

据我了解，我不应该在 while 循环内发出请求，但是，我不确定是否有办法绕过它以获得分页结果。

if __name__ == 'main':

    starter_fun(url, header, endpoints):

starter_func(url, header, endpoints):

    loop = asyncio.get_event_loop() #event loop
    future = asyncio.ensure_future(fetch_all(url, header, endpoints))
    loop.run_until_complete(future) #loop until done

async def fetch_all(url, header, endpoints):

    async with ClientSession() as session:
        for endpoint in endpoints:
           task = asyncio.ensure_future(fetch(url, header, endpoint))
           tasks.append(task)
        res = await asyncio.gather(*tasks) # gather task responses
        return res

async def fetch(url, header, endpoint): 

    total_tasks = []
    async with session.get(url, headers=header, params=params, ssl=False) as response:
        response_json = await response.json()
        data = response_json['key']
       tasks = asyncio.ensure_future(fetch_more(response_json, data, params, header, url, endpoint, session)) //this is where I am getting stuck
        total_tasks.append(tasks)
    return data


//function to get paginated results of api endpoint
async def fetch_more(response_json, data, params, header, url, endpoint, session): //this is where I am getting stuck

    while len(response_json['key']) >= params['limit']:
        params['offset'] = response_json['offset'] + len(response_json['key'])
        async with session.get(url, headers=header, params=params, ssl=False) as response_continued:
            resp_continued_json = await response_continued.json()
            data.extend(resp_continued_json[kebab_to_camel(endpoint)])
   return data

目前，无论是否使用 fetch_more 函数，我都会得到 1000 个结果，但是使用 fetch_more 应该会更多。关于如何处理异步分页的任何想法？

【参考方案1】：

from aiohttp import web

 async def fetch(self size: int = 10):

    data = "some code to fetch data here"

    def paginate(_data, _size):
        import itertools

        while True:
            i1, i2 = itertools.tee(_data)
            _data, page = (itertools.islice(i1, _size, None),
                              list(itertools.islice(i2, _size)))
            if len(page) == 0:
                break
            yield page

    return web.json_response(list(paginate(_data=data, _size=size)))