fork download
copy 
  1. import asyncio
  2. from contextvars import ContextVar
  3.  
  4. active_steps = ContextVar[list[int]]("active_steps", default=None)
  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.  
  34. if __name__ == "__main__":
  35.     asyncio.run(main())
  36.  
Success #stdin #stdout 0.49s 26088KB
comments (?)
stdin
copy 
Standard input is empty
stdout
copy 
Task 1 final result: [1, 2, 3]
Task 2 final result: [4, 5, 6]
https://ideone.com/9LVQrT
language:
Python 3 (python  3.12)
created:
336 days ago
visibility:
secret

Share or Embed source code

Discover > Sphere Engine API

The brand new service which powers Ideone!

Discover > IDE Widget

Widget for compiling and running the source code in a web browser!