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.

[ice-tong]

Same feature request here! Expect the updates on this topic.

[Moreh-LeeJunhyeok]

also requests this feature. log_N_times is useful in simple warning cases

[AlexH-BoxFish]

I wanted to do something similar but only showing 1 of any message per minute. I don't know if there's any potential issues, but I managed to get this working through a custom 'Debounce' log level:

import sys
from datetime import datetime, timedelta

from loguru import logger

debounce_log_level = "DEBOUNCE"


class Debounce:
    last_debounced_message = datetime.now()
    debounce_interval = timedelta(minutes=1)

    @staticmethod
    def only_debounce(record):
        if record["level"].name == debounce_log_level:
            if datetime.now() - Debounce.last_debounced_message > Debounce.debounce_interval:
                Debounce.last_debounced_message = datetime.now()
                return True

        return False


logger.level(debounce_log_level, no=4, color="<blue>", icon="@")
logger.add(sys.stderr, filter=Debounce.only_debounce, level=debounce_log_level)
logger.log(debounce_log_level, "debounced message")

[drdarina]

Same feature request!

[noamgot]

I came up with this simple POC:

import atexit as _atexit
import sys as _sys
from collections import defaultdict

from loguru import _defaults
from loguru._logger import Core as _Core
from loguru._logger import Logger as _Logger


class MyLogger(_Logger):
    _msg_history = defaultdict(set)

    def _log(self, level, from_decorator, options, message, args, kwargs):
        once = kwargs.pop("once", False)
        if once:
            if message in self._msg_history[level]:
                return
            self._msg_history[level].add(message)
        super()._log(level, from_decorator, options, message, args, kwargs)


logger = MyLogger(
    core=_Core(),
    exception=None,
    depth=0,
    record=False,
    lazy=False,
    colors=False,
    raw=False,
    capture=True,
    patchers=[],
    extra={},
)
if _defaults.LOGURU_AUTOINIT and _sys.stderr:
    logger.add(_sys.stderr)

_atexit.register(logger.remove)

logger.warning("This is a warning message", once=True)
logger.warning("This is a warning message", once=True)
logger.warning("This is a warning message", once=True)
logger.warning("This is a warning message 123", once=True)


logger.info("This is a warning message", once=True)
logger.info("This is a warning message", once=True)

seems to work just fine:

Asking the core devs here - do you think it can be plugged into the current implementation?
I know it handles a very specific case (single logger), but maybe we can think of a way to extend it (I'm not familiar with the source code...)

[Delgan]

@noamgot It looks like you can use the existing filter parameter to implement such solution, without requiring to touch Loguru internals:

import sys
from collections import defaultdict

from loguru import logger

_msg_history = defaultdict(set)


def filter(record):
    if "once" in record["extra"]:
        level = record["level"].no
        message = record["message"]
        if message in _msg_history[level]:
            return False
        _msg_history[level].add(message)
    return True


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

logger.warning("This is a warning message", once=True)
logger.warning("This is a warning message", once=True)
logger.warning("This is a warning message", once=True)
logger.warning("This is a warning message 123", once=True)


logger.info("This is a warning message", once=True)
logger.info("This is a warning message", once=True)

[noamgot]

Looks like a decent solution, works for me. thanks @Delgan