Skip to main content

Async/Await in Practice

Asyncio Core

Async/Await Crash Course

Coroutines let you write sequential-looking code that cooperatively multitasks.

Structure

import asyncio

async def fetch_order(order_id):
await asyncio.sleep(0.1)
return {'id': order_id}

async def main():
order = await fetch_order('123')
print(order)

asyncio.run(main())
  • async declares coroutines; await pauses until awaited tasks complete.
  • Use asyncio.run to bootstrap top-level coroutines.

Gathering tasks

async def main():
orders = await asyncio.gather(
fetch_order('123'),
fetch_order('456'),
fetch_order('789'),
return_exceptions=True,
)
print(orders)

asyncio.run(main())
  • asyncio.gather waits for all tasks; pass return_exceptions=True to collect exceptions as results instead of raising them immediately.
  • asyncio.create_task schedules background work; keep references so you can await or cancel later.

Task groups (Python 3.11+)

async def main():
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(fetch_order('123'))
t2 = tg.create_task(fetch_order('456'))
print(t1.result(), t2.result())

asyncio.run(main())
  • asyncio.TaskGroup is the recommended structured-concurrency approach for new code: the async with block waits for all tasks, and if one fails the siblings are cancelled and the failures are raised as an ExceptionGroup.
  • Prefer it over bare gather when tasks should succeed or fail together.

Error handling

  • Wrap awaits in try/except just like synchronous code.
  • Use asyncio.wait_for to enforce deadlines; on Python 3.11+ prefer the async with asyncio.timeout(10): context manager. Since 3.11, asyncio.TimeoutError is an alias of the builtin TimeoutError.
  • Cancel tasks with task.cancel() to clean up when shutting down.

Next up in your learning path