Using loguru + `.conteztualize()` to provide context for async task execution (hashtag614)

[vikramsubramanian]

Hello!

I'm trying to use .contextualize() to generate a logger with a different context per-task that gets executed in async.gather(), but I can't seem to get it to work. Would appreciate some help if possible.

The basics of the code I have looks like this:

import sys
from asyncio import AbstractEventLoop, gather, get_event_loop

def do_thing() -> None:
    loop: AbstractEventLoop = get_event_loop()
    loop.run_until_complete(do_async_thing())

async do_async_thing() -> None:
    tasks = []

    for thing in list_declared_somewhere_else:
        tasks.append(do_the_real_async_thing())

    hashtag This part is where I'm struggling
    logger.add(sys.stdout, format="{message} | {extra}")
    with logger.contextualize(task_id="blah"):
        await gather(*tasks)

async do_the_real_async_thing() -> None:
    hashtag ... some more async stuff ...

I'm trying to figure out how to make a new "context" for each task that gets executed, such that I could theoretically filter by that context to just see logs related to a single task. I assume that this is possible, but I can't really get to the bottom of it. Any pointers?

[vikramsubramanian]

Hi.

The contextualize() method needs to be called from within the task you want to be uniquely identified. Your current implementation is calling contextualize() from within the parent task so the worker tasks won't inherit the value.

Here is an example:

import sys
from asyncio import AbstractEventLoop, gather, get_event_loop

from loguru import logger

list_declared_somewhere_else = [1, 2, 3]

def do_thing() -> None:
    loop: AbstractEventLoop = get_event_loop()
    loop.run_until_complete(do_async_thing())

async def do_async_thing() -> None:
    tasks = []

    for thing in list_declared_somewhere_else:
        tasks.append(do_the_real_async_thing(thing))

    await gather(*tasks)

async def do_the_real_async_thing(task_id) -> None:
    with logger.contextualize(task_id=task_id):
        await subtask()

async def subtask():
    logger.info("Exec task")

if __name__ == "__main__":
    logger.remove()
    logger.add(sys.stdout, format="{message} | {extra}")
    do_thing()

The contexutalize() method is mainly useful when you have nested calls and you don't wish to carry the value everywhere. You can also take a look at the bind() method.

[vikramsubramanian]

oh my goodness, now that you say that the solution seems obvious 😅 thank you so much! Works like a charm.