Aiotools: Idiomatic Asyncio Utilities

Warnings

• breaking 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.
  Fix: Migrate usage from `aiotools.TaskGroup` to `aiotools.TaskScope` for new and existing code. `TaskScope` handles sibling task failures gracefully without cancelling others.
• deprecated 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.
  Fix: Replace `aiotools.func.apartial` with `functools.partial`.
• gotcha 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.
  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.
• gotcha 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.
  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.
• gotcha 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.
  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`.

Install

• pip install aiotools Install stable version

Imports

• TaskScope

  from aiotools import TaskScope

  `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.
• cancel_and_wait

  from aiotools.cancel import cancel_and_wait

  A utility for structured cancellation that re-raises `CancelledError` on external requests and absorbs it otherwise, improving cancellation consistency.
• actxmgr

  from aiotools.context import actxmgr

  Decorator for easily creating asynchronous context managers, similar to `contextlib.asynccontextmanager`.
• start_server

  from aiotools.server import start_server

  A high-level utility for launching asyncio-based server daemons with automatic signal handling and multi-process support.
• Supervisor

  from aiotools.supervisor import Supervisor

  `Supervisor` is an alias of `TaskScope` and designed for resilient server-oriented task management where sibling tasks are not cancelled if one fails.

Quickstart

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.

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())