Senior LevelPython
How does asyncio.TaskGroup (Python 3.11+) implement Structured Concurrency to eliminate leaked background tasks?
❓ Interview Prompt: Explain Structured Concurrency, `asyncio.TaskGroup`, exception propagation, `ExceptionGroup`, and eliminating unhandled dangling background tasks.
Prior to Python 3.11, launching background tasks ('asyncio.create_task()') ran the risk of leaking dangling tasks if an exception occurred in a sibling coroutine!
'asyncio.TaskGroup' (Python 3.11+):
Implements Structured Concurrency:
•
Automatically waits for all child tasks to complete upon exiting the 'async with TaskGroup()' block.
•
If any child task raises an exception, 'TaskGroup' automatically cancels all remaining child tasks and raises an 'ExceptionGroup'!
Code Implementation Example
python
import asyncio
async def fetch_user(user_id):
await asyncio.sleep(0.1)
return f'User_{user_id}'
async def main():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(fetch_user(1))
task2 = tg.create_task(fetch_user(2))
# Both tasks guaranteed complete here!
print(task1.result(), task2.result())
asyncio.run(main())💡Executing concurrent async tasks safely using asyncio.TaskGroup.
🎯 Interview Key Takeaway
Use asyncio.TaskGroup in Python 3.11+ to enforce structured concurrency and eliminate leaked background tasks.
#TaskGroup#Structured Concurrency#asyncio#Python 3.11#Senior
Back to all Python Questions→