Python — Advanced Async Patterns

Introduction

Once you understand basic coroutines and await, the real power of asyncio comes from combining tasks, handling errors, and building production-grade concurrent systems. This guide covers the patterns you need for real-world async code.

These patterns apply to web servers, API clients, data pipelines, database operations, and any application that performs I/O-bound work concurrently.

asyncio.gather

gather runs multiple coroutines concurrently and returns their results in the same order you passed them. It is the most common way to parallelize I/O-bound work.

gather.py

Python

1	import asyncio
2	import httpx
3
4	async def fetch_url(client: httpx.AsyncClient, url: str) -> dict:
5	response = await client.get(url)
6	return {"url": url, "status": response.status_code, "length": len(response.text)}
7
8	async def main():
9	urls = [
10	"https://httpbin.org/get",
11	"https://httpbin.org/ip",
12	"https://httpbin.org/user-agent",
13	"https://httpbin.org/headers",
14	]
15
16	async with httpx.AsyncClient() as client:
17	# All requests run concurrently — total time ≈ slowest single request
18	results = await asyncio.gather(
19	*[fetch_url(client, url) for url in urls]
20	)
21
22	for r in results:
23	print(f"{r['url']}: {r['status']} ({r['length']} bytes)")
24
25	asyncio.run(main())
26
27	# gather with error handling
28	async def might_fail(n: int) -> int:
29	if n == 3:
30	raise ValueError(f"Task {n} failed!")
31	return n * 10
32
33	async def main_safe():
34	# return_exceptions=True prevents one failure from cancelling others
35	results = await asyncio.gather(
36	*[might_fail(i) for i in range(5)],
37	return_exceptions=True,
38	)
39	for i, result in enumerate(results):
40	if isinstance(result, Exception):
41	print(f"Task {i}: FAILED — {result}")
42	else:
43	print(f"Task {i}: {result}")
44
45	asyncio.run(main_safe())

✓

best practice

Always use return_exceptions=True when you want all tasks to complete regardless of individual failures. Without it, the first exception cancels all remaining tasks.

asyncio.create_task

create_task.py

Python

1	import asyncio
2
3	async def background_worker(name: str, interval: float):
4	while True:
5	print(f"[{name}] tick")
6	await asyncio.sleep(interval)
7
8	async def main():
9	# create_task schedules coroutine immediately — runs in background
10	task1 = asyncio.create_task(background_worker("Worker-A", 1.0))
11	task2 = asyncio.create_task(background_worker("Worker-B", 1.5))
12
13	# Let them run for 5 seconds
14	await asyncio.sleep(5)
15
16	# Cancel both
17	task1.cancel()
18	task2.cancel()
19
20	# Wait for cancellation to complete
21	await asyncio.gather(task1, task2, return_exceptions=True)
22	print("All workers stopped")
23
24	asyncio.run(main())
25
26	# Task groups — structured concurrency (Python 3.11+)
27	async def process_item(item: int) -> str:
28	await asyncio.sleep(0.1)
29	return f"processed-{item}"
30
31	async def main_groups():
32	results = []
33	async with asyncio.TaskGroup() as tg:
34	tasks = [tg.create_task(process_item(i)) for i in range(10)]
35	# All tasks guaranteed complete here
36	results = [t.result() for t in tasks]
37	print(results)
38
39	asyncio.run(main_groups())

Task Groups

Task groups (Python 3.11+) provide structured concurrency: if any task in the group fails, all other tasks are cancelled and an ExceptionGroup is raised. This prevents orphaned tasks and makes error handling predictable.

task_groups.py

Python

1	import asyncio
2
3	async def fetch_user(user_id: int) -> dict:
4	await asyncio.sleep(0.1)
5	if user_id == 3:
6	raise ConnectionError("User 3 DB timeout")
7	return {"id": user_id, "name": f"User_{user_id}"}
8
9	async def main():
10	try:
11	async with asyncio.TaskGroup() as tg:
12	task1 = tg.create_task(fetch_user(1))
13	task2 = tg.create_task(fetch_user(2))
14	task3 = tg.create_task(fetch_user(3)) # this will fail
15	task4 = tg.create_task(fetch_user(4))
16	except* ConnectionError as eg:
17	# ExceptionGroup handling — Python 3.11+
18	for exc in eg.exceptions:
19	print(f"Connection error: {exc}")
20	except* Exception as eg:
21	for exc in eg.exceptions:
22	print(f"Other error: {exc}")
23
24	# When using gather with return_exceptions:
25	# You check results after. With TaskGroup, exceptions propagate
26	# immediately and cancel sibling tasks.
27
28	asyncio.run(main())

ℹ

info

Prefer TaskGroup over gather for new code. Task groups enforce structured concurrency — every task is guaranteed to complete before the block exits, and failures propagate cleanly.

Async Generators

async_generators.py

Python

1	import asyncio
2
3	# Async generator — yields values over time
4	async def stream_data(n: int):
5	for i in range(n):
6	await asyncio.sleep(0.1) # simulate I/O
7	yield {"index": i, "value": i ** 2}
8
9	async def main():
10	async for item in stream_data(5):
11	print(item)
12
13	# Async generator with send() — bidirectional
14	async def accumulator():
15	total = 0
16	while True:
17	value = yield total
18	total += value
19
20	async def main_accum():
21	gen = accumulator()
22	await gen.asend(None) # prime the generator
23	print(await gen.asend(10)) # → 10
24	print(await gen.asend(20)) # → 30
25	print(await gen.asend(5)) # → 35
26
27	# Async list comprehension
28	async def main_comp():
29	results = [item async for item in stream_data(10)]
30	print(results)
31
32	# Async generator expression
33	async def main_gen():
34	squares = [x async for x in stream_data(5) if x["value"] > 10]
35	print(squares)
36
37	# Pagination pattern
38	async def paginate_all(url: str):
39	page = 1
40	while True:
41	data = await fetch_page(url, page) # your async fetch
42	if not data["results"]:
43	break
44	for item in data["results"]:
45	yield item
46	page += 1
47
48	async def main_pagination():
49	async for item in paginate_all("https://api.example.com/items"):
50	process(item) # handle each item as it arrives

Async Context Managers

async_context.py

Python

1	import asyncio
2	from contextlib import asynccontextmanager
3
4	# Class-based async context manager
5	class AsyncDatabasePool:
6	def __init__(self, max_connections: int = 5):
7	self.max_connections = max_connections
8	self.connections = []
9	self.semaphore = None
10
11	async def __aenter__(self):
12	self.semaphore = asyncio.Semaphore(self.max_connections)
13	print(f"Pool created with {self.max_connections} connections")
14	return self
15
16	async def __aexit__(self, exc_type, exc_val, exc_tb):
17	print("Pool shutting down...")
18	self.connections.clear()
19	return False # don't suppress exceptions
20
21	async def acquire(self):
22	async with self.semaphore:
23	conn = f"conn-{len(self.connections)}"
24	self.connections.append(conn)
25	return conn
26
27	# Decorator-based async context manager
28	@asynccontextmanager
29	async def managed_resource(name: str):
30	print(f"Acquiring {name}")
31	resource = {"name": name, "active": True}
32	try:
33	yield resource
34	except Exception as e:
35	print(f"Error in {name}: {e}")
36	raise
37	finally:
38	resource["active"] = False
39	print(f"Released {name}")
40
41	# Using nested async context managers
42	async def main():
43	async with AsyncDatabasePool(max_connections=10) as pool:
44	async with managed_resource("cache") as cache:
45	conn = await pool.acquire()
46	print(f"Using {conn} with cache={cache['name']}")
47
48	asyncio.run(main())

Async Queues

async_queues.py

Python

1	import asyncio
2
3	# Producer-Consumer with backpressure
4	async def producer(queue: asyncio.Queue, n: int):
5	for i in range(n):
6	item = {"id": i, "data": f"payload-{i}"}
7	await queue.put(item) # blocks if queue is full
8	print(f"Produced: {item['id']}")
9	await asyncio.sleep(0.05)
10	# Signal completion to each consumer
11	for _ in range(3):
12	await queue.put(None)
13
14	async def consumer(queue: asyncio.Queue, name: str):
15	processed = 0
16	while True:
17	item = await queue.get()
18	if item is None:
19	queue.task_done()
20	break
21	await asyncio.sleep(0.1) # simulate processing
22	processed += 1
23	print(f" {name}: processed {item['id']}")
24	queue.task_done()
25	print(f" {name}: done — {processed} items")
26
27	async def main():
28	queue = asyncio.Queue(maxsize=10) # backpressure at 10
29	await asyncio.gather(
30	producer(queue, 30),
31	consumer(queue, "Worker-1"),
32	consumer(queue, "Worker-2"),
33	consumer(queue, "Worker-3"),
34	)
35
36	asyncio.run(main())
37
38	# Priority queue
39	async def priority_example():
40	pq = asyncio.PriorityQueue()
41	await pq.put((1, "low priority"))
42	await pq.put((0, "high priority"))
43	await pq.put((2, "lowest priority"))
44
45	while not pq.empty():
46	priority, item = await pq.get()
47	print(f"[{priority}] {item}") # high, low, lowest

Cancellation & Error Handling

cancellation.py

Python

1	import asyncio
2
3	# Timeout with asyncio.timeout (Python 3.11+)
4	async def slow_operation():
5	await asyncio.sleep(10)
6	return "done"
7
8	async def main_timeout():
9	try:
10	async with asyncio.timeout(2.0):
11	result = await slow_operation()
12	except TimeoutError:
13	print("Operation timed out!")
14
15	# Cancellation with cleanup
16	async def cancellable_fetch(url: str):
17	try:
18	print(f"Starting fetch: {url}")
19	await asyncio.sleep(5) # simulate network
20	return f"data from {url}"
21	except asyncio.CancelledError:
22	print(f"Cancellation received for {url}")
23	# Cleanup: close connections, free resources
24	raise # MUST re-raise or task state is corrupted
25
26	async def main_cancel():
27	task = asyncio.create_task(cancellable_fetch("https://example.com"))
28	await asyncio.sleep(1)
29	task.cancel()
30
31	try:
32	await task
33	except asyncio.CancelledError:
34	print("Task was cancelled successfully")
35
36	# Shield — protect critical operations from cancellation
37	async def critical_write(data):
38	await asyncio.sleep(2)
39	print(f"Written: {data}")
40
41	async def main_shield():
42	task = asyncio.create_task(critical_write("important"))
43	await asyncio.sleep(0.5)
44	# Even if outer code cancels, shield protects the write
45	try:
46	result = await asyncio.shield(task)
47	except asyncio.CancelledError:
48	# Shield was cancelled but inner task still runs
49	pass
50	await task # wait for it to finish
51
52	# Structured error handling with TaskGroup
53	async def main_errors():
54	async def safe_operation(n: int):
55	await asyncio.sleep(0.1)
56	if n == 2:
57	raise ValueError(f"Bad input: {n}")
58	return n
59
60	try:
61	async with asyncio.TaskGroup() as tg:
62	tasks = [tg.create_task(safe_operation(i)) for i in range(5)]
63	except* ValueError as eg:
64	print(f"Got {len(eg.exceptions)} errors")
65	for exc in eg.exceptions:
66	print(f" {exc}")
67	# tasks list is available — successful results in .result()
68	# failed tasks have no result (exception was raised)
69
70	asyncio.run(main_errors())

✓

best practice

Always re-raise CancelledError inside your cleanup code. Swallowing it corrupts the task state and makes debugging nearly impossible.

$Blueprint — Engineering Documentation·Section ID: PYTHON-ASYNC-PAT·Revision: 1.0

X GitHub YouTube Instagram Facebook Discord Telegram Buy me a coffee Sponsor