Skip to content

Instantly share code, notes, and snippets.

@privatwolke
Created September 20, 2017 13:27
Show Gist options
  • 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())
)
}
@Yelinz
Copy link

Yelinz commented Aug 26, 2025

Variation for nested dicts/lists

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

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

async def gather_dict(tasks: Mapping[_K, Awaitable[_V] | Mapping | list]) -> dict[_K, _V]:
    async def resolve(value):
        if isinstance(value, Mapping):
            return await gather_dict(value)
        elif isinstance(value, list):
            return [await resolve(v) for v in value]
        elif inspect.isawaitable(value):
            return await value
        else:
            return value

    async def inner(key: _K, value) -> tuple[_K, Any]:
        return key, await resolve(value)

    pair_tasks = (inner(key, value) for key, value 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