15.7. AsyncIO Task Group

15.7.1. Example

  • python3 -m pip install httpx

>>> import asyncio
>>> import httpx
>>>
>>> urls = [
...     'https://python3.info/index.html',
...     'https://python3.info/LICENSE.html',
...     'https://python3.info/about/versions.html',
...     'https://python3.info/about/references.html',
...     'https://python3.info/about/history.html',
...     'https://python3.info/about/links.html',
... ]
>>>
>>>
>>> async def fetch(url):
...     print(f'fetch before: {url}')
...     async with httpx.AsyncClient() as client:
...         response = await client.get(url)
...     print(f'fetch after: {url}')
...     return response.text
>>>
>>>
>>> async def main():
...     async with asyncio.TaskGroup() as group:
...         for url in urls:
...             group.create_task(fetch(url))
>>>
>>>
>>> asyncio.run(main())

15.7.2. Use Case - 1

>>> 
... async def main() -> list[Result]:
...     todo = []
...     async with asyncio.TaskGroup() as tg:
...         for file in get_files():
...             cmd = run(f'python -m doctest {file}')
...             task = tg.create_task(cmd, name=str(file))
...             todo.append(task)
...     return [Result(file=t.get_name(), output=t.result()) for t in todo]