16.7. AsyncIO Task Group
16.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())
fetch before: https://python3.info/index.html
fetch before: https://python3.info/LICENSE.html
fetch before: https://python3.info/about/versions.html
fetch before: https://python3.info/about/references.html
fetch before: https://python3.info/about/history.html
fetch before: https://python3.info/about/links.html
fetch after: https://python3.info/LICENSE.html
fetch after: https://python3.info/about/history.html
fetch after: https://python3.info/about/versions.html
fetch after: https://python3.info/about/links.html
fetch after: https://python3.info/about/references.html
fetch after: https://python3.info/index.html
16.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]