python-3.x - 如何在 Trio 中收集任务结果？

Question

我编写了一个脚本，它使用一个 Nursery 和 asks 模块来循环并根据循环变量调用一个 API。我收到回复，但不知道如何像使用 asyncio 一样返回数据。

我还有一个关于将 API 限制为每秒 5 个的问题。

from datetime import datetime
import asks
import time
import trio

asks.init("trio")
s = asks.Session(connections=4)

async def main():
    start_time = time.time()

    api_key = 'API-KEY'
    org_id = 'ORG-ID'
    networkIds = ['id1','id2','idn']

    url = 'https://api.meraki.com/api/v0/networks/{0}/airMarshal?timespan=3600'
    headers = {'X-Cisco-Meraki-API-Key': api_key, 'Content-Type': 'application/json'}

    async with trio.open_nursery() as nursery:
        for i in networkIds:
            nursery.start_soon(fetch, url.format(i), headers)

    print("Total time:", time.time() - start_time)



async def fetch(url, headers):
    print("Start: ", url)
    response = await s.get(url, headers=headers)
    print("Finished: ", url, len(response.content), response.status_code)




if __name__ == "__main__":
    trio.run(main)

当我运行nursery.start_soon(fetch...) 时，我在fetch 中打印数据，但是如何返回数据？我没有看到任何类似于 asyncio.gather(*tasks) 函数的东西。

此外，我可以将 session 数限制为 1-4，这有助于低于每秒 5 个 API 的限制，但我想知道是否有内置方法可以确保在任何给定的秒内调用不超过 5 个 API ?

Answer 1

基于 this answers ，您可以定义以下函数:

async def gather(*tasks):

    async def collect(index, task, results):
        task_func, *task_args = task
        results[index] = await task_func(*task_args)

    results = {}
    async with trio.open_nursery() as nursery:
        for index, task in enumerate(tasks):
            nursery.start_soon(collect, index, task, results)
    return [results[i] for i in range(len(tasks))]

然后，您可以通过简单地修补 trio(添加收集功能)以与 asyncio 完全相同的方式使用 trio:

import trio
trio.gather = gather

这是一个实际的例子:

async def child(x):
    print(f"Child sleeping {x}")
    await trio.sleep(x)
    return 2*x

async def parent():
    tasks = [(child, t) for t in range(3)]
    return await trio.gather(*tasks)

print("results:", trio.run(parent))