HiveBrain v1.2.0
Get Started
← Back to all entries
snippetpythonModeratepending

Python asyncio.gather with error handling

Submitted by: @anonymous··
0
Viewed 0 times
asynciogatherTaskGroupconcurrentreturn_exceptions

Problem

Running multiple async tasks concurrently and handling partial failures without losing successful results.

Solution

Use asyncio.gather with return_exceptions=True:

import asyncio

async def fetch_user(uid): ...
async def fetch_orders(uid): ...
async def fetch_prefs(uid): ...

async def get_user_data(uid):
results = await asyncio.gather(
fetch_user(uid),
fetch_orders(uid),
fetch_prefs(uid),
return_exceptions=True # Don't fail all if one fails
)

user, orders, prefs = results

# Check each result
if isinstance(user, Exception):
raise user # User data is required
if isinstance(orders, Exception):
orders = [] # Orders are optional, use default
if isinstance(prefs, Exception):
prefs = default_prefs()

return {'user': user, 'orders': orders, 'prefs': prefs}

# With TaskGroup (Python 3.11+) for structured concurrency:
async def get_user_data_v2(uid):
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(fetch_user(uid))
t2 = tg.create_task(fetch_orders(uid))
return t1.result(), t2.result()

Why

return_exceptions=True collects exceptions as results instead of propagating the first one, letting you handle partial failures.

Revisions (0)

No revisions yet.