Module livekit.plugins.krisp_internal.log

Classes

class ThrottlingFilter (interval: float = 5)

Expand source code

class ThrottlingFilter(logging.Filter):
    """A logging filter that throttles repeated log messages."""

    def __init__(self, interval: float = 5):
        super().__init__()
        self.interval = interval
        self.last_logged: dict[tuple[int, object], float] = {}

    def filter(self, record: logging.LogRecord) -> bool:
        now = time.time()
        key = (record.levelno, record.msg)
        if now - self.last_logged.get(key, 0) > self.interval:
            self.last_logged[key] = now
            return True
        return False

A logging filter that throttles repeated log messages.

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

Ancestors

• logging.Filter

Methods

def filter(self, record: logging.LogRecord) ‑> bool

Expand source code

def filter(self, record: logging.LogRecord) -> bool:
    now = time.time()
    key = (record.levelno, record.msg)
    if now - self.last_logged.get(key, 0) > self.interval:
        self.last_logged[key] = now
        return True
    return False

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.