aiotask-context Library

0.6.1 verified Wed Apr 15 auth: no python

The `aiotask-context` library provides a mechanism to store and retrieve context information within `asyncio.Task` objects. It is analogous to `threading.local` but designed for asynchronous Python applications, allowing for request-scoped or task-scoped data management without explicit parameter passing. The current version is 0.6.1, with a stable but infrequent release cadence.

pip install aiotask-context

Common errors

error ModuleNotFoundError: No module named 'aiotask_context' ↓

cause The 'aiotask_context' module is not installed in the Python environment.

fix

Install the module using pip: 'pip install aiotask-context'.

error ImportError: cannot import name 'context' from 'aiotask_context' ↓

cause Attempting to import a non-existent 'context' attribute from the 'aiotask_context' module.

fix

Use the correct import statement: 'from aiotask_context import get, set'.

error AttributeError: module 'aiotask_context' has no attribute 'get_context' ↓

cause The 'get_context' function does not exist in the 'aiotask_context' module.

fix

Use the correct function: 'from aiotask_context import get'.

error AttributeError: '_asyncio.Task' object has no attribute 'context' ↓

cause This error occurs when the `aiotask-context` task factory has not been properly set on the `asyncio` event loop, meaning tasks are created without the necessary 'context' attribute.

fix

You need to set the aiotask-context task factory for your event loop. For example: loop = asyncio.get_event_loop() followed by loop.set_task_factory(context.task_factory) or loop.set_task_factory(context.copying_task_factory).

error Context not propagating in child asyncio tasks ↓

cause By default, context information stored with `aiotask-context` does not automatically propagate to new `asyncio.Task` objects spawned from a parent task unless a specific task factory is set.

fix

Set a context-aware task factory on your event loop before creating tasks. Use loop.set_task_factory(context.task_factory) for a shared context, or loop.set_task_factory(context.copying_task_factory) to provide each new task with a fresh copy of the parent's context.

Warnings

breaking The package underwent a significant change in version 0.4.0. Prior to 0.4.0, `Context` was a class that needed to be instantiated (e.g., `c = Context()`). Since 0.4.0, `context` is a pre-instantiated singleton object directly imported (`from aiotask_context import context`). ↓

fix Update your import statement from `from aiotask_context import Context` and remove any instantiation, using `context` directly instead.

breaking The project's GitHub repository and primary maintainership transitioned from `vimeo/aiotask-context` to `python-aiotask-context/aiotask-context`. While the package name on PyPI remains the same, older documentation or references might point to the outdated repository. ↓

fix Always refer to the official repository at `https://github.com/python-aiotask-context/aiotask-context` for the most up-to-date information, issues, and contributions.

gotcha Context in `aiotask-context` is strictly task-local. When `asyncio.create_task` is used, the child task inherits a *copy* of the parent's context. Any modifications made by the child task to its context will not reflect in the parent task's context or sibling tasks' contexts. ↓

fix Understand that `aiotask-context` provides task-local storage, not global or shared state across tasks. If you need to communicate state *back* to a parent task, use standard asyncio mechanisms like `asyncio.Queue` or return values.

gotcha `aiotask-context` is designed for task-scoped data and should not be confused with `contextvars` (PEP 567) introduced in Python 3.7. While both provide context management, `aiotask-context` predates `contextvars` and relies on direct manipulation of `asyncio.Task` internals. `contextvars` offers a more robust and officially supported way for context propagation. ↓

fix For new projects on Python 3.7+, consider evaluating `contextvars` from the standard library. If using `aiotask-context`, be aware of its specific implementation and ensure it meets your requirements.

Imports

context
wrong
```
from aiotask_context import Context
```
correct
```
from aiotask_context import context
```
Since version 0.4.0, `context` is a pre-instantiated singleton object, not a class.

Quickstart

This quickstart demonstrates how `aiotask-context` allows setting context variables that are isolated to the current `asyncio.Task`. Sub-tasks inherit a copy of their parent's context, and modifications within a sub-task do not affect the parent or sibling tasks, ensuring task-local storage.

import asyncio
from aiotask_context import context

async def nested_task(task_id: str):
    """A sub-task that accesses and modifies context."""
    request_id = context.get("request_id")
    print(f"Task {task_id}: Retrieved request_id: {request_id}")

    # Set a task-specific value; this does not affect other tasks' contexts
    # nor the parent task's context once the child task's context is copied.
    context.set("task_data", f"Data from {task_id}")
    await asyncio.sleep(0.01) # Simulate async work
    print(f"Task {task_id}: Set task_data: {context.get('task_data')}")

async def main():
    print("--- aiotask-context Quickstart ---")

    # Set context in the main task
    context.set("request_id", "req-xyz-123")
    print(f"Main task: Initial request_id: {context.get('request_id')}")

    # Create and run sub-tasks. Each sub-task inherits a *copy* of the parent's context.
    task1 = asyncio.create_task(nested_task("A"))
    task2 = asyncio.create_task(nested_task("B"))

    await task1
    await task2

    # The main task's context remains unchanged by sub-tasks' modifications
    print(f"Main task: Final request_id: {context.get('request_id')}") # Should be 'req-xyz-123'
    # Task-specific data set in sub-tasks is not accessible here in the main task
    print(f"Main task: task_data: {context.get('task_data')}") # Should be None
    print("This demonstrates task-local context isolation in asyncio.")

if __name__ == "__main__":
    asyncio.run(main())