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())
)
}
@cosenmarco
Copy link

❤️

@yunusemrecatalcam
Copy link

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

@jchan172
Copy link

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

@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