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-09-03 21:16 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


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
                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			if len(self._cache) > self.cache_size:
27			self._cache.popitem(last=False)
28			return True
29			elif current_time - self._cache[key] > self.lockout_time:
30			self._cache[key] = current_time
31			self._cache.move_to_end(key)
32			if len(self._cache) > self.cache_size:
33			self._cache.popitem(last=False)
34			return True
35			return False
36
37			@staticmethod
38			def _record_key(record: LogRecord):
39			return (
40			record.pathname,
41			record.module,
42			record.lineno,
43			record.levelno,
44			record.msg,
45			record.args
46			)
47

kytos-ng / kytos

Pull Request — master (#497)

filters.RepeateMessageFilter.filter() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like