Python context asyncio

Question

In streaming mode, we have something called the “Stream Context” in which we can save some data (state) for access in a special stream. In asyncio, I need to save some state in the current execution path so that all subsequent coroutines can access it. What is the solution? Note. I know that every coroutine function is created for the execution path in asyncio, but for some reason I cannot save the state in the properties of the function. (Although this method is not very good at all)

Answer 1

You can look at tasklocals library . It introduces tasklocals.local as threading.local for multithreading.

Answer 2

There is also https://github.com/azazel75/metapensiero.asyncio.tasklocal , but you should be aware that tasks are often created inside the library as well as asyncio using ensure_future(a_coroutine) and there is no real way to track these new tasks and initialize them local residents (possibly with the tasks from which they are created). (a “hack” that should set the loop.set_task_factory() function with something that does the job, hoping all the code uses loop.create_task() to create tasks, which is not always true ...)

Another problem is that if any of your code runs inside the future callback function Task.current_task() , which is used by both libraries to select the correct copy of the locales to serve, None will always be returned ...

Answer 3

Starting with Python 3.7 you can use contextvars.ContextVar .

In the example below, I declared request_id and set the value to some_outer_coroutine , and then turned to it in some_inner_coroutine .

 import asyncio import contextvars # declare context var request_id = contextvars.ContextVar('Id of request.') async def some_inner_coroutine(): # get value print('Processed inner coroutine of request: {}'.format(request_id.get())) async def some_outer_coroutine(req_id): # set value request_id.set(req_id) await some_inner_coroutine() # get value print('Processed outer coroutine request: {}'.format(request_id.get())) async def main(): tasks = [] for req_id in range(1, 5): tasks.append(asyncio.create_task(some_outer_coroutine(req_id))) await asyncio.gather(*tasks) if __name__ == '__main__': asyncio.run(main()) 

Exit:

 Processed inner coroutine of request: 1 Processed outer coroutine request: 1 Processed inner coroutine of request: 2 Processed outer coroutine request: 2 Processed inner coroutine of request: 3 Processed outer coroutine request: 3 Processed inner coroutine of request: 4 Processed outer coroutine request: 4