如何以更简洁的惯用方式从字典中的Futures收集结果？

I'm trying to write idiomatic code to collect results from Futures stored in a dictionary. Here's my current setup:

import asyncio
async def sleep(seconds):
    print(f'sleeping for {seconds} seconds')
    await asyncio.sleep(seconds)
    print(f'finished sleeping {seconds} seconds')
async def run():
    tasks = {
        '4': sleep(4),
        '3': sleep(3),
        '2': sleep(2),
        '1': sleep(1),
    }
    print(await gather_from_dict(tasks))
if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(run())

The expected output is:

sleeping for 2 seconds
sleeping for 1 seconds
sleeping for 4 seconds
sleeping for 3 seconds
finished sleeping 1 seconds
finished sleeping 2 seconds
finished sleeping 3 seconds
finished sleeping 4 seconds
{'4': None, '3': None, '2': None, '1': None}

I've come up with this concise solution:

async def gather_from_dict(tasks:Dict[Hashable, Awaitable], loop=None, return_exceptions=False) -> Dict:
    results = await asyncio.gather(
        *tasks.values(), loop=loop, return_exceptions=return_exceptions
    )
    return dict(zip(tasks.keys(), results))

Is there a simpler implementation for this? Thanks in advance!

Answer #

Great question! Your current implementation is already super concise and idiomatic for this use case—you’ve hit on the core pattern needed here. Let’s break this down and see if we can refine it slightly, but overall this is about as simple as it gets.

First, let’s confirm: the two essential steps here can’t be skipped:

1. Pass all the awaitable coroutines to asyncio.gather to run them concurrently and collect results
2. Map those results back to their original keys from the dictionary

Your code does both steps cleanly. The only small improvements we can make are around type safety (to make the function more robust for static type checkers) and minor readability tweaks:

from typing import Dict, Awaitable, Hashable, TypeVar

T = TypeVar('T')  # Generic type for the awaitable's return value

async def gather_from_dict(
    tasks: Dict[Hashable, Awaitable[T]],
    loop=None,
    return_exceptions: bool = False
) -> Dict[Hashable, T]:
    results = await asyncio.gather(
        *tasks.values(),
        loop=loop,
        return_exceptions=return_exceptions
    )
    return dict(zip(tasks.keys(), results))

This version adds a generic type variable T so the function can correctly infer the return type of the coroutines, making it safer to use with type checkers like mypy.

Is there a way to make this even shorter? Not really—any alternative approach would end up being more verbose or less robust. For example, you might consider using a dictionary comprehension with asyncio.create_task, but that requires manually tracking tasks and fetching results, which adds unnecessary complexity:

# Less concise alternative—avoid this
async def gather_from_dict(tasks: Dict[Hashable, Awaitable], **kwargs) -> Dict:
    task_map = {key: asyncio.create_task(coro) for key, coro in tasks.items()}
    await asyncio.gather(*task_map.values(), **kwargs)
    return {key: task.result() for key, task in task_map.items()}

This approach doesn’t handle return_exceptions cleanly (you’d have to add extra logic to catch exceptions from task.result()), so it’s worse than your original implementation.

To sum up: your solution is already the best, most straightforward way to gather results from a dictionary of asyncio awaitables. The only optional tweak is adding better type annotations for robustness.

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