filters.RepeateMessageFilter.filter() - Code Metrics - Inspection of "Merge pull request #497 from kytos-ng/logging/filt..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( bd16ba...38fe60 )

unknown

created 2024-09-05 15:03 UTC

filters.RepeateMessageFilter.filter() B

↳ Parent: filters

Complexity

Conditions

Size

Total Lines	16
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	16
nop	2
dl	0
loc	16
rs	8.6666
c	0
b	0
f	0


from collections import OrderedDict
from logging import LogRecord
from threading import Lock


class RepeateMessageFilter:
    lockout_time: float
    cache_size: int
    _cache: OrderedDict[tuple, float]
    _lock: Lock

    def __init__(self, lockout_time: float, cache_size: int = 512):
        self.lockout_time = lockout_time
        self.cache_size = cache_size
        self._cache = OrderedDict()
        self._lock = Lock()

    def filter(self, record: LogRecord) -> bool:
        key = self._record_key(record)
        current_time = record.created
        with self._lock:
            if key not in self._cache:
                self._cache[key] = current_time
                if len(self._cache) > self.cache_size:
                    self._cache.popitem(last=False)
                return True
            elif current_time - self._cache[key] > self.lockout_time:
                self._cache[key] = current_time
                self._cache.move_to_end(key)
                if len(self._cache) > self.cache_size:
                    self._cache.popitem(last=False)
                return True
            return False

    @staticmethod
    def _record_key(record: LogRecord):
        return (
            record.pathname,
            record.module,
            record.lineno,
            record.levelno,
            record.msg,
            record.args
        )


1
2			from collections import OrderedDict
3			from logging import LogRecord
4			from threading import Lock
5
6
7			class RepeateMessageFilter:
8			lockout_time: float
9			cache_size: int
10			_cache: OrderedDict[tuple, float]
11			_lock: Lock
12
13			def __init__(self, lockout_time: float, cache_size: int = 512):
14			self.lockout_time = lockout_time
15			self.cache_size = cache_size
16			self._cache = OrderedDict()
17			self._lock = Lock()
18
19			def filter(self, record: LogRecord) -> bool:
20			key = self._record_key(record)
21			current_time = record.created
22			with self._lock:
23			if key not in self._cache:
24			self._cache[key] = current_time
25			if len(self._cache) > self.cache_size:
26			self._cache.popitem(last=False)
27			return True
28			elif current_time - self._cache[key] > self.lockout_time:
29			self._cache[key] = current_time
30			self._cache.move_to_end(key)
31			if len(self._cache) > self.cache_size:
32			self._cache.popitem(last=False)
33			return True
34			return False
35
36			@staticmethod
37			def _record_key(record: LogRecord):
38			return (
39			record.pathname,
40			record.module,
41			record.lineno,
42			record.levelno,
43			record.msg,
44			record.args
45			)
46

kytos-ng / kytos

Push — master ( bd16ba...38fe60 )

filters.RepeateMessageFilter.filter() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like