aiotask-context

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