filters.RepeateMessageFilter.filter() - Code Metrics - Inspection of "Add in log filter for pacing logs" - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#497)

unknown

created 2024-08-30 21:02 UTC

filters.RepeateMessageFilter.filter() B

↳ Parent: filters

Complexity

Conditions

Size

Total Lines	17
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	17
nop	2
dl	0
loc	17
rs	8.6166
c	0
b	0
f	0


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

kytos-ng / kytos

Pull Request — master (#497)

filters.RepeateMessageFilter.filter() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like