Log something only once (or only N times)

[MatthewScholefield]

I have a situation where I want to log something only once. It would be nice to have some logger.opt(once=True). For example:

def fetch_document(...):  # Repeatedly called
    document = database.retrieve(...)
    if document.endswith(r'\r\n'):
        logger.opt(once=True).warning(r'Database is storing \r\n  endings. Please reconfigure.')

Without such an option, either a non-fatal warning bloats the logs repeating the same message, or I'd need to unpleasantly use a global variable:

did_log_database_warning = False
def fetch_document(...):
    global did_log_database_warning
    ...
    if document.endswith(r'\r\n') and not did_log_database_warning:
        did_log_database_warning = True
        logger.warning(...)

Naturally, this would lead to me writing a help function, log_once. So, I figured it would make sense to include something like this with loguru. What do you think?

To support the log_once = logger.opt(once=True); log_once('here'); log_once('there'), it could store a set of printed log statements based on the line number of the logger call. Alternatively, using a new method, it could simply store it once per instance like logger.limit(1).warning('hello') which would implicitly allow another use case of constraining the number of lines of some related set of warnings ie.:

io_logger = logger.limit(100, overflow_msg='Suppressing future io errors')
def decrypt_chars(data):
    for offset, c in enumerate(data):
        if cannot_decrypt(c):
            io_logger.warning('Could not decrypt byte {} at offset {}', hex(c), offset)
            continue
        d = decrypt(c)
        if is_invalid_char(d):
            io_logger.warning('Invalid character {!r} at offset {}', d, offset)
        yield d

[Delgan]

Hi @MatthewScholefield. Sorry for replying so late.

This seems to be a very interesting idea. I remember at some point I was also looking for a way to prevent excessive logging of the same message. Finally I did not implement anything because I was not sure such rate-limiting for logging was good practice or not.

I wish I could just extend opt() but it seems too restrictive. We probably need to suppress logs for a period of time only (eg. no more than 100 calls every minute), which means more attributes are needed. The proposed .limit() method looks good.

Alternatively, one could use another library like ratelimiter. It should work perfectly well. The downside is that it require wrapping logging call with a context manager, which is surely less convenient that using logger.limit() in-line.

So, yeah, thanks for your suggestion! It could be worth integrating it into Loguru, I will keep thinking about it.

[liz-zither]

I'm quite interested in this feature as well. Can you expand on the ratelimiter idea with some pseudo code? I'm not sure how to implement.

[Delgan]

@liz-zither Ok, my bad. The ratelimiter library is definitely not designed for this use case. Sorry, I didn't look into the details.

Here is another example using a custom class:

from loguru import logger
from unittest.mock import MagicMock


class LimitedLogger:
    def __init__(self, limit=None):
        self._limit = limit
        self._count = 0

    def get(self):
        if self._limit is None or self._count < self._limit:
            self._count += 1
            return logger
        else:
            return MagicMock()


limited_logger = LimitedLogger(limit=10)

for i in range(100):
    limited_logger.get().info("Iteration #{}", i)

[DeflateAwning]

It seems that the documentation on this feature got missed. Can someone write the final usage of this feature here? Perhaps we can also add it to the Help Guide.

[Delgan]

@DeflateAwning It's not yet documented because it's not yet implemented. :)

Actually I don't know if this should be provided as built-in in Loguru or as a third-library.

[cdgiv]

I'd love this feature as well! I've worked with internal logging systems before that offered a log_every_N() type method which, not quite the same as what's requested here, but I think is flexible enough to satisfy several needs.

Playing around a bit, I can get a log_every_N functionality with a filter:

def log_every_n_filter(record, everyN={}):
    msg = record["message"]
    if everyN.get(msg, 0) % 10 == 0:
        everyN[msg] += 1
        return True
    return False

logger.add(sys.stderr, filter=log_every_n_filter)

This would apply to every message, so I'd need to think of a way to easily set per-message N values.