aiotask-context Library
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.
Common errors
-
ModuleNotFoundError: No module named 'aiotask_context'
cause The 'aiotask_context' module is not installed in the Python environment.fixInstall the module using pip: 'pip install aiotask-context'. -
ImportError: cannot import name 'context' from 'aiotask_context'
cause Attempting to import a non-existent 'context' attribute from the 'aiotask_context' module.fixUse the correct import statement: 'from aiotask_context import get, set'. -
AttributeError: module 'aiotask_context' has no attribute 'get_context'
cause The 'get_context' function does not exist in the 'aiotask_context' module.fixUse the correct function: 'from aiotask_context import get'. -
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.fixYou 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)`. -
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.fixSet 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`).
- 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.
- 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.
- 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.
Install
-
pip install aiotask-context
Imports
- context
from aiotask_context import Context
from aiotask_context import context
Quickstart
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())