Skip to content

Instantly share code, notes, and snippets.

@privatwolke
Created September 20, 2017 13:27
Show Gist options
  • Star 14 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save privatwolke/11711cc26a843784afd1aeeb16308a30 to your computer and use it in GitHub Desktop.
Save privatwolke/11711cc26a843784afd1aeeb16308a30 to your computer and use it in GitHub Desktop.
Python: Gather a dictionary of asyncio Task instances while preserving keys
async def gather_dict(tasks: dict):
async def mark(key, coro):
return key, await coro
return {
key: result
for key, result in await gather(
*(mark(key, coro) for key, coro in tasks.items())
)
}
@Nnonexistent
Copy link

Nnonexistent commented Mar 22, 2024

Type hints with any types:

import asyncio
from collections.abc import Awaitable, Mapping
from typing import TypeVar

_K = TypeVar("_K")
_V = TypeVar("_V")


async def gather_dict(tasks: Mapping[_K, Awaitable[_V]]) -> dict[_K, _V]:
    async def inner(key: _K, coro: Awaitable[_V]) -> tuple[_K, _V]:
        return key, await coro

    pair_tasks = (inner(key, coro) for key, coro in tasks.items())
    return dict(await asyncio.gather(*pair_tasks))

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment