Asynchronous Python Beyond async/await: From Event Loop Basics to Structured Concurrency
by Gary Worthington, More Than Monkeys
Most Python developers can write an async def function and use await, but far fewer understand how the event loop works, how to manage concurrency safely, or how to prevent common pitfalls like orphaned tasks and swallowed exceptions.
This post takes you from the basics of asynchronous Python all the way to advanced structured concurrency patterns, with real code examples and explanations at each step.
We will focus on asyncio (built into Python), but also look at how trio and anyio approach the same problems.
1. Async 101 — What async and await Really Do
The simplest async function looks like this:
import asyncio
async def greet():
print("Hello")
await asyncio.sleep(1)
print("World")
asyncio.run(greet())
What is happening here:
- async def defines a coroutine; a special kind of function that can be paused and resumed.
- await pauses the coroutine until the awaited object finishes.
- asyncio.sleep() is non-blocking. It tells the event loop “pause here and let other tasks run until this timer finishes”.
- asyncio.run() starts the event loop, runs the coroutine, and stops the loop when done.
If you replaced await asyncio.sleep(1) with time.sleep(1), the whole program would block for a second, preventing any other async work from running.
2. Synchronous vs Asynchronous: The Core Difference
Before going deeper, it’s important to understand why async code exists and how it differs from traditional synchronous code.
Synchronous Python
In synchronous code, each operation waits for the previous one to complete before moving on.
import time
def main():
print("Start")
time.sleep(2) # blocks everything for 2 seconds
print("End")
main()
During time.sleep(2), nothing else can happen. If you had multiple I/O operations, they would run strictly one after another.
Asynchronous Python
In asynchronous code, tasks can overlap while waiting for I/O, so time isn’t wasted doing nothing.
import asyncio
async def task(name, delay):
print(f"{name} started")
await asyncio.sleep(delay)
print(f"{name} finished")
async def main():
await asyncio.gather(
task("A", 2),
task("B", 2)
)
asyncio.run(main())
Both tasks start together, and while one is waiting, the other can run. Total runtime here is ~2 seconds, not 4.
The takeaway
- Synchronous: predictable order, but wasteful for I/O-heavy work.
- Asynchronous: more efficient for I/O-bound tasks, but requires careful design to avoid pitfalls.
3. The Event Loop Under the Hood
The event loop is a scheduler. It keeps track of tasks and runs them in small chunks, switching between them when they yield control via await.
You can run multiple coroutines concurrently:
import asyncio
async def say(msg, delay):
print(f"Starting {msg}")
await asyncio.sleep(delay)
print(msg)
async def main():
task1 = asyncio.create_task(say("First", 2))
task2 = asyncio.create_task(say("Second", 1))
await task1
await task2
asyncio.run(main())
How this works:
- asyncio.create_task() schedules the coroutine to run in the background.
- Both tasks start immediately. While say("First") is sleeping, say("Second") can also run.
- The output order depends on when each task finishes.
This is cooperative multitasking. Tasks must explicitly yield (await) for others to run.
4. Common Async Pitfalls
Forgetting to await
If you call an async function without await, it returns a coroutine object that does nothing until awaited.
async def greet():
print("Hi")
greet() # Does nothing
Blocking the Event Loop
Running CPU-heavy or blocking code directly in async functions freezes everything.
import time
async def slow_block():
time.sleep(5) # blocks the loop
Fire-and-Forget Tasks Without Error Handling
If you create a task and never await it, exceptions are swallowed until Python logs them later.
import asyncio
async def risky():
raise ValueError("Something went wrong")
async def main():
asyncio.create_task(risky()) # not awaited
await asyncio.sleep(1)
asyncio.run(main())
You will see a warning in Python 3.11+, but older versions silently lose the exception.
5. Structured Concurrency: Why You Need It
Structured concurrency ensures all tasks are created in a well-defined scope, and that if one fails, related tasks are cancelled. This prevents “zombie” tasks running after the main function exits.
From Python 3.11 onwards, asyncio.TaskGroup provides this.
import asyncio
async def worker(n):
await asyncio.sleep(n)
print(f"Worker {n} done")
if n == 2:
raise RuntimeError("Boom")
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(worker(1))
tg.create_task(worker(2))
tg.create_task(worker(3))
except* RuntimeError as e:
print(f"Caught errors: {e}")
asyncio.run(main())
What’s happening:
- TaskGroup starts all tasks together.
- If one task fails, the others are cancelled.
- except* syntax handles multiple exceptions from different tasks.
6. Trio’s Take — Nurseries
trio pioneered structured concurrency in Python with nurseries.
import trio
async def worker(n):
await trio.sleep(n)
print(f"Worker {n} done")
if n == 2:
raise RuntimeError("Boom")
async def main():
try:
async with trio.open_nursery() as nursery:
nursery.start_soon(worker, 1)
nursery.start_soon(worker, 2)
nursery.start_soon(worker, 3)
except Exception as e:
print(f"Caught error: {e}")
trio.run(main)
Trio’s approach ensures:
- All tasks finish before leaving the nursery.
- Exceptions cancel the nursery automatically.
- You never forget to await a background task.
7. Bridging Sync and Async Worlds
If you have CPU-bound or blocking code, run it in a thread or process pool so it doesn’t block the event loop.
import asyncio
import time
def cpu_bound():
time.sleep(2)
return "Done"
async def main():
result = await asyncio.to_thread(cpu_bound)
print(result)
asyncio.run(main())
asyncio.to_thread() runs the function in a separate thread, letting the loop handle other work meanwhile.
8. Error Handling and Cancellation Propagation
When cancelling tasks, always catch asyncio.CancelledError to perform cleanup. The following example, the worker runs for 2 seconds before it is cancelled.
import asyncio
async def worker():
try:
while True:
print("Working...")
await asyncio.sleep(0.5)
except asyncio.CancelledError:
print("Worker shutting down gracefully")
raise
async def main():
task = asyncio.create_task(worker())
await asyncio.sleep(2)
task.cancel()
try:
await task
except asyncio.CancelledError:
print("Task cancelled")
asyncio.run(main())
Without catching CancelledError, you might leave resources (files, sockets, DB connections) open.
9. Performance Considerations
Async Python is best for I/O-bound workloads - network calls, file I/O, database queries. For CPU-bound work, use multiprocessing or native extensions.
Sequential vs Async Example
import asyncio
from datetime import datetime
async def download(n):
await asyncio.sleep(n)
return f"Finished {n}"
async def main():
# sequential
now = datetime.now()
for i in [1, 2, 3]:
await download(i)
print(f"time for seq: {datetime.now() - now}")
# concurrent
now = datetime.now()
results = await asyncio.gather(*(download(i) for i in [1, 2, 3]))
print(f"time for conc: {datetime.now() - now}")
asyncio.run(main())
Output:
time for seq: 0:00:06.003802
time for conc: 0:00:03.001615
10. Patterns for Production
- Always scope tasks in TaskGroup or nurseries.
- Cancel tasks gracefully with CancelledError.
- Use asyncio.timeout() to prevent runaway tasks:
import asyncio
async def slow():
await asyncio.sleep(5)
async def main():
try:
async with asyncio.timeout(2):
await slow()
except TimeoutError:
print("Timed out")
asyncio.run(main())
- Avoid mixing sync blocking calls in async code.
- For framework-agnostic async code, use anyio.
11. Final Thoughts
Async in Python is more than await. It’s about designing concurrency so that tasks start, fail, and clean up predictably.
If you take one thing away, it’s this: never start a background task without deciding how it will end.
Gary Worthington is a software engineer, delivery consultant, and agile coach who helps teams move fast, learn faster, and scale when it matters. He writes about modern engineering, product thinking, and helping teams ship things that matter.
Through his consultancy, More Than Monkeys, Gary helps startups and scaleups improve how they build software — from tech strategy and agile delivery to product validation and team development.
Visit morethanmonkeys.co.uk to learn how we can help you build better, faster.
Follow Gary on LinkedIn for practical insights into engineering leadership, agile delivery, and team performance