Async Python with asyncio: TaskGroup, Timeouts and Gather
asyncio lets one thread wait on many network operations at once. It shines for I/O-heavy work: calling APIs, scraping, chat servers, database queries.
The basics
import asyncio
import httpx
async def fetch_status(client: httpx.AsyncClient, url: str) -> int:
response = await client.get(url)
return response.status_code
async def main():
async with httpx.AsyncClient(timeout=10) as client:
status = await fetch_status(client, "https://www.python.org")
print(status)
asyncio.run(main())
Run things concurrently with TaskGroup
async def main(urls: list[str]):
async with httpx.AsyncClient() as client:
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(fetch_status(client, u)) for u in urls]
# all tasks are done here
return [t.result() for t in tasks]
If one task fails, the TaskGroup cancels the others and raises an ExceptionGroup. No forgotten background tasks.
When you want every result, even failures
results = await asyncio.gather(*coros, return_exceptions=True)
errors = [r for r in results if isinstance(r, Exception)]
Timeouts
try:
async with asyncio.timeout(5):
data = await slow_call()
except TimeoutError:
data = None
Limit concurrency
semaphore = asyncio.Semaphore(10)
async def polite_fetch(client, url):
async with semaphore: # at most 10 requests at a time
return await client.get(url)
Pitfalls
- Blocking calls (
time.sleep,requests.get, heavy CPU work) freeze the event loop. Use async libraries orawait asyncio.to_thread(fn). - A coroutine you never
awaitnever runs. - Async is about waiting efficiently, not raw CPU speed. For CPU-bound work, use processes (or free-threaded Python).
Was this article helpful?
Comments
No comments yet. Have a question or a tip? Be the first to share it.