Mocking Trio's open_nursery and spawn with asyncio.TaskGroup
Show how to mock Trio's nursery pattern using Python's asyncio.TaskGroup to simulate task spawning and completion.
Python code
20 linesimport asyncio
class MockSpawner:
async def spawn(self, nursery):
print("Spawning mock task...")
await asyncio.sleep(1)
print("Mock task completed")
async def open_nursery():
async with asyncio.TaskGroup() as nursery:
mock = MockSpawner()
nursery.create_task(mock.spawn(nursery))
print("Nursery closed")
async def main():
print("Opening nursery")
await open_nursery()
if __name__ == "__main__":
asyncio.run(main())Output
Opening nursery
Spawning mock task...
Mock task completed
Nursery closedHow it works
Used asyncio.TaskGroup (Python 3.11+) as a stand-in for Trio's trio.open_nursery context manager. create_task schedules a coroutine concurrently inside the task group. The async context manager waits for all tasks to finish before exiting, mirroring Trio's cancellation semantics. The mock spawner logs its progress to simulate real async work.
Common mistakes
- Forgetting to await the coroutine inside `create_task` (passing coroutine directly is fine, but not awaiting later).
- Not using `async with` for TaskGroup, causing tasks to be cancelled or not awaited.
- Assuming Trio's nursery functions work directly with asyncio; you must translate to `asyncio.TaskGroup`.
Variations
- Use `asyncio.gather` inside a plain async function instead of TaskGroup.
- Keep Trio imports and use `trio.open_nursery` if the actual runtime is Trio, not asyncio.
Real-world use cases
- Testing async code that expects nursery semantics by replacing Trio with asyncio for unit tests.
- Porting a Trio-based concurrency pattern to a standard asyncio application.
- Simulating task orchestration in a test suite without installing third-party async libraries.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.