pygotham_context_propagator.py

pygotham_context_propagator.py

import asyncio

def set_current_context(ctx, loop=None):
    # retrieves the current Task
    loop = loop or asyncio.get_event_loop()
    task = asyncio.Task.current_task(loop=loop)
    # attach the Context to the task!
    setattr(task, '__current_ctx', ctx)

def get_current_context(loop=None):
    # retrieves the current Task
    loop = loop or asyncio.get_event_loop()
    task = asyncio.Task.current_task(loop=loop)
    # get the Context
    ctx = getattr(task, '__current_ctx', None)
    return ctx