{"id":5106,"library":"aiotools","title":"Aiotools: Idiomatic Asyncio Utilities","description":"aiotools is a collection of idiomatic utilities designed to reduce boilerplate code when working with `asyncio`. It provides robust solutions for safe cancellation, structured concurrency through `TaskScope`, asynchronous context managers, multi-process server daemons, and other high-level coroutine utilities. The library is actively maintained and currently at version 2.2.3, with a release cadence that includes regular bug fixes and feature enhancements, targeting Python 3.11 and newer.","status":"active","version":"2.2.3","language":"en","source_language":"en","source_url":"https://github.com/achimnol/aiotools","tags":["asyncio","concurrency","task management","server","utilities","structured concurrency","cancellation"],"install":[{"cmd":"pip install aiotools","lang":"bash","label":"Install stable version"}],"dependencies":[{"reason":"Requires Python 3.11 or newer for core features and API compatibility.","package":"python","optional":false}],"imports":[{"note":"`TaskGroup` was deprecated since v2.0 in favor of `TaskScope`, which provides more consistent behavior with `asyncio.TaskGroup` (Python 3.11+) and better lifecycle tracking.","wrong":"from aiotools import TaskGroup","symbol":"TaskScope","correct":"from aiotools import TaskScope"},{"note":"A utility for structured cancellation that re-raises `CancelledError` on external requests and absorbs it otherwise, improving cancellation consistency.","symbol":"cancel_and_wait","correct":"from aiotools.cancel import cancel_and_wait"},{"note":"Decorator for easily creating asynchronous context managers, similar to `contextlib.asynccontextmanager`.","symbol":"actxmgr","correct":"from aiotools.context import actxmgr"},{"note":"A high-level utility for launching asyncio-based server daemons with automatic signal handling and multi-process support.","symbol":"start_server","correct":"from aiotools.server import start_server"},{"note":"`Supervisor` is an alias of `TaskScope` and designed for resilient server-oriented task management where sibling tasks are not cancelled if one fails.","symbol":"Supervisor","correct":"from aiotools.supervisor import Supervisor"}],"quickstart":{"code":"import asyncio\nfrom aiotools import TaskScope\n\nasync def worker(name, delay):\n    try:\n        print(f\"Worker {name}: Starting...\")\n        await asyncio.sleep(delay)\n        print(f\"Worker {name}: Finished after {delay}s.\")\n        return f\"Result from {name}\"\n    except asyncio.CancelledError:\n        print(f\"Worker {name}: Was cancelled!\")\n        raise\n    except Exception as e:\n        print(f\"Worker {name}: Encountered error: {e}\")\n        raise\n\nasync def main():\n    print(\"Main: Starting TaskScope example\")\n    async with TaskScope() as scope:\n        task1 = scope.create_task(worker(\"Alpha\", 2))\n        task2 = scope.create_task(worker(\"Beta\", 1))\n        task3 = scope.create_task(worker(\"Gamma\", 3))\n\n        # You can await individual tasks within the scope\n        print(f\"Main: Awaiting Task Beta...\")\n        result2 = await task2\n        print(f\"Main: Task Beta finished with: {result2}\")\n\n    print(\"Main: All tasks in TaskScope are complete or cancelled.\")\n    # After exiting the TaskScope, all tasks are guaranteed to be done.\n    # Results and exceptions from other tasks can be retrieved if needed.\n    print(f\"Main: Task Alpha result: {task1.result()}\")\n    print(f\"Main: Task Gamma result: {task3.result()}\")\n\nif __name__ == \"__main__\":\n    asyncio.run(main())\n","lang":"python","description":"This quickstart demonstrates `TaskScope`, a core feature for structured concurrency. It launches multiple worker coroutines within a `TaskScope`. The `async with TaskScope()` block ensures that all child tasks created within it are either completed or cancelled before the block is exited, providing safe lifecycle management. This example also shows how to await individual tasks and retrieve their results."},"warnings":[{"fix":"Migrate usage from `aiotools.TaskGroup` to `aiotools.TaskScope` for new and existing code. `TaskScope` handles sibling task failures gracefully without cancelling others.","message":"The `TaskGroup` class has been deprecated since `aiotools` v2.0 in favor of `TaskScope`. While `TaskGroup` still exists, `TaskScope` is the recommended high-level API for structured concurrency and provides behavior consistent with `asyncio.TaskGroup` introduced in Python 3.11.","severity":"breaking","affected_versions":">=2.0.0"},{"fix":"Replace `aiotools.func.apartial` with `functools.partial`.","message":"The `aiotools.func.apartial` utility was deprecated in `aiotools` v2.0. Python's built-in `functools.partial()` now works seamlessly with asynchronous functions as of Python 3.8 and should be used instead.","severity":"deprecated","affected_versions":">=2.0.0"},{"fix":"Avoid using `VirtualClock` in cross-platform test suites or provide platform-specific test runners. Consider alternative time-mocking libraries for Windows compatibility if `VirtualClock`'s specific features are not strictly required.","message":"The `aiotools.timer.VirtualClock` feature, used for deterministic testing of `asyncio.sleep()` calls, relies on patching event loop internals and is only functional on UNIX-like operating systems. It will not work on Windows.","severity":"gotcha","affected_versions":"all"},{"fix":"Design your concurrency patterns carefully. Use `TaskScope` when you need independent tasks within a group that can fail without affecting siblings (e.g., background workers in a server). Use `asyncio.TaskGroup` (or `TaskScope` if its behavior matches) when tasks are interdependent and a single failure should halt the entire group.","message":"Understanding the difference in error handling between `TaskScope` and `asyncio.TaskGroup` (which `aiotools.TaskGroup` wrapped prior to Python 3.11, and `TaskScope` now extends). `TaskScope` is designed for server-oriented tasks where the failure of one child task does NOT automatically cancel all other sibling tasks, allowing for more resilient services. `asyncio.TaskGroup` (and older `aiotools.TaskGroup` behavior) will cancel all siblings if one task raises an unhandled exception.","severity":"gotcha","affected_versions":"all"},{"fix":"Always use `await aiotools.cancel.cancel_and_wait(task)` when you need to cancel an `asyncio` task, instead of `task.cancel(); await task` directly. This ensures predictable handling of `CancelledError`.","message":"When manually cancelling `asyncio` tasks, ensure to use `aiotools.cancel.cancel_and_wait()` for consistent behavior. Without it, managing `CancelledError` propagation (re-raising vs. absorbing) can be tricky and lead to inconsistencies across your codebase, especially prior to Python 3.11's structured concurrency improvements.","severity":"gotcha","affected_versions":"<3.11"}],"env_vars":null,"last_verified":"2026-04-13T00:00:00.000Z","next_check":"2026-07-12T00:00:00.000Z"}