aiotools

import asyncio from aiotools import TaskScope async def worker(name, delay): try: print(f"Worker {name}: Starting...") await asyncio.sleep(delay) print(f"Worker {name}: Finished after {delay}s.") return f"Result from {name}" except asyncio.CancelledError: print(f"Worker {name}: Was cancelled!") raise except Exception as e: print(f"Worker {name}: Encountered error: {e}") raise async def main(): print("Main: Starting TaskScope example") async with TaskScope() as scope: task1 = scope.create_task(worker("Alpha", 2)) task2 = scope.create_task(worker("Beta", 1)) task3 = scope.create_task(worker("Gamma", 3)) # You can await individual tasks within the scope print(f"Main: Awaiting Task Beta...") result2 = await task2 print(f"Main: Task Beta finished with: {result2}") print("Main: All tasks in TaskScope are complete or cancelled.") # After exiting the TaskScope, all tasks are guaranteed to be done. # Results and exceptions from other tasks can be retrieved if needed. print(f"Main: Task Alpha result: {task1.result()}") print(f"Main: Task Gamma result: {task3.result()}") if __name__ == "__main__": asyncio.run(main())