Loguru: Capturing and replaying logs from nested functions

[lgvaz]

Hi! I just started using loguru this week and there is a specific behaviour I want to achieve but I'm not sure how to search for it. Let me provide a very simple example of what I'm talking about:

I have a function that may output logs

def bark():
    logger.info('whoof')

In another part of the code, I want to capture the log and add something before it:

# intercept message coming from bark
bark()
# add something or log something before it
logger.info('A dog will bark')
# replay message from bark

The final output I'm looking for is:

A dog will bark
whoof

The example heavily simplifies my use case, here calling logger.info('A dog will bark') before bark() would solve the issue, but for my use case that is not possible

---

I took a look at InterceptHandler but I don't think that is related. I'm guessing there is a way to use sinks to achieve this, but I'm not sure how.

[Delgan]

Hi @lgvaz. This looks like a quite complicated request. :smile:

I see no other solution than to use a custom sink. For example:

class DeferingSink:

    def __init__(self):
        self.defered = []

    def write(self, message):
        record = message.record

        if record["function"] == "bark":
            self.defered.append(message)
            return

        while self.defered:
            print(self.defered.pop(0), end="")

        print(message, end="")

sink = DeferingSink()
logger.add(sink)

If you need to identify specific log messages, you can make use of bind():

def bark():
    logger.bind(defered=True).info("Whoof")
    logger.info("Not defered")

logger.bind(replay=True).info("A dog will bark")


class DeferingSink:

    def __init__(self):
        self.defered = []

    def write(self, message):
        record = message.record

        if record["extra"].get("defered", False):
            self.defered.append(message)
            return

        if record["extra"].get("replay", False):
            while self.defered:
                print(self.defered.pop(0), end="")

        print(message, end="")

[lgvaz]

Thanks a lot for the quick response @Delgan! Your answer does indeed solve my problem

I adapted your answer a bit to use a context manager, I removed the if statements to make the code shorter but those can still be used.

class DeferringSinkContext:
    def __init__(self):
        self.deferred = []

    def __enter__(self):
        logger.remove()
        logger.add(self)
        return self

    def write(self, message):
        self.deferred.append(message)

    def __exit__(self, type, value, traceback):
        logger.remove()
        logger.add(sys.stderr)
        if self.deferred:
            logger.info('The dog will bark')
            for msg in self.deferred:
                logger.info(msg)

def bark():
    logger.info('whoof')

with DeferringSinkContext() as sink:
    bark()

This produces the (almost) wanted result:

2020-10-11 11:37:32.856 | INFO     | __main__:__exit__:14 - The dog will bark
2020-10-11 11:37:32.856 | INFO     | __main__:__exit__:16 - 2020-10-11 11:37:32.850 | INFO     | __main__:bark:2 - whoof

I just have a final question: Each of the items in self.deferred are of the type loguru._handler.Message which already contains record with all the information for this log, how can I log those directly? i.e. in my example, I'm calling logger.info which duplicates the log.

[Delgan]

You can patch() the logger before re-sending the log message in order to update the record dict:

for msg in self.deferred:
    record = msg.record
    patched = logger.patch(lambda r: r.update(record))
    patched.log(record["level"].no, record["message"])

[lgvaz]

Works perfectly! Thanks a lot for the amazing help ❤️

[lgvaz]

Just an additional quick question, the logs produced by patched don't have color, any idea why?

[Delgan]

Yeah, that's because we are calling logger.log() using an "anonymous" level: record["level"].no. We are just passing a level no which is not associated to any color.

We need to pass the record["level"].name instead, but we need first to make sure that such level exist. For example, if you use logger.log(42, "Foobar") somewhere in your code then record["level"].name is equals to "Level 42" but such level is not registered. This becomes a little bit more verbose unfortunately.

for msg in self.deferred:
    record = msg.record
    patched = logger.patch(lambda r: r.update(record))
    level = record["level"].name

    try:
        logger.level(level)
    except ValueError:
        # If the level 'name' is not registered, use its 'no' instead
        level = record["level"].no

    patched.log(level, record["message"])