aiotask-context Library

0.6.1 · active · verified Wed Apr 15

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

Warnings

Install

Imports

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

view raw JSON →