fork download
  1. import asyncio
  2. from contextvars import ContextVar
  3.  
  4. active_steps = ContextVar[list[int]]("active_steps", default=[])
  5.  
  6.  
  7. async def add_steps(task_name, steps):
  8. for step in steps:
  9. # get the active steps
  10. current_steps = active_steps.get()
  11.  
  12. # add a new step
  13. current_steps.append(step) # mutates the shared list
  14.  
  15. # set the variable with the new steps
  16. active_steps.set(current_steps)
  17.  
  18. await asyncio.sleep(0.1) # simulate async work
  19.  
  20. print(f"{task_name} final result: {active_steps.get()}")
  21. # reset the variable with an empty array
  22. active_steps.set([])
  23.  
  24.  
  25. async def main() -> None:
  26. requests = [add_steps("Task 1", [1, 2, 3]), add_steps("Task 2", [4, 5, 6])]
  27. await asyncio.gather(*requests)
  28.  
  29. # Expected output:
  30. # Task 1 final result: [1, 2, 3]
  31. # Task 2 final result: [4, 5, 6]
  32.  
  33. if __name__ == "__main__":
  34. asyncio.run(main())
  35.  
Success #stdin #stdout 0.38s 26024KB
stdin
Standard input is empty
stdout
Task 1 final result: [1, 4, 2, 5, 3, 6]
Task 2 final result: [1, 4, 2, 5, 3, 6]