Python: Gather a dictionary of asyncio Task instances while preserving keys

gather_dict.py

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())
        )
    }

❤️

it helped a lot! here's if someone needs request version

async def single_stat_fast(self, metricgroup):
      loop = asyncio.get_event_loop()

      async def get_metric(metric, query):
          return metric, await loop.run_in_executor(None, requests.get, self.prometheus_address + '/api/v1/query' + '?query=' + query)
      results = {
          metric: result.json()['data']['result'][0]['value'][1]
          for metric, result in await asyncio.gather(
              *(get_metric(metric, query) for metric, query in self.metric2query[metricgroup].items())
          )
      }

      return results

Here's version with type hints:

import asyncio
from typing import Awaitable, Dict, Tuple

async def gather_dict(tasks: Dict[str, Awaitable[str]]) -> Dict[str, str]:
    async def mark(key: str, coro: Awaitable[str]) -> Tuple[str, str]:
        return key, await coro

    return {
        key: result
        for key, result in await asyncio.gather(
            *(mark(key, coro) for key, coro in tasks.items())
        )
    }

And to test it:

async def get_html() -> str:
    await asyncio.sleep(1)
    html = 'html codes...'
    return html

async def get_data() -> str:
    await asyncio.sleep(1)
    data = [
        {
            'id': 1,
            'name': 'Sally',
        },
        {
            'id': 2,
            'name': 'John',
        }
    ]
    return data

async def main():
    result = await gather_dict({'html': get_html(), 'data': get_data()})
    print(result)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

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))