pocketutils.misc.mem_cache.MemCache.remove() - Code Metrics - Inspection of "feat: deprecated code; fixed from_toml; added logu..." - dmyersturnbull/pocketutils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( b312d0...29adcf )

by Douglas

created 2021-10-05 00:04 UTC

pocketutils.misc.mem_cache.MemCache.remove() A

↳ Parent: pocketutils.misc.mem_cache

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	4
nop	2
dl	0
loc	4
rs	10
c	0
b	0
f	0

import operator

import sys
from abc import ABC
from datetime import datetime
from typing import Any, Callable, Dict, Generic, Iterator, Mapping, Optional, Sequence, TypeVar

from psutil import virtual_memory


# noinspection PyProtectedMember
from pocketutils.core._internal import nicesize

K = TypeVar("K", covariant=True)

V = TypeVar("V", contravariant=True)



class MemCachePolicy(Generic[K, V]):

    def should_archive(self) -> bool:

        raise NotImplementedError()

    def can_archive(self) -> bool:

        raise NotImplementedError()

    def reindex(self, items: Mapping[K, V]) -> None:

        raise NotImplementedError()

    def items(self) -> Iterator[K]:

        # TODO this is not overloaded!!

        raise NotImplementedError()

    def accessed(self, key: K) -> None:

        pass

    def added(self, key: K, value: V) -> None:

        pass

    def removed(self, key: K) -> None:

        pass


class MemoryLimitingPolicy(MemCachePolicy, ABC):

    def __init__(
        self,

        max_memory_bytes: Optional[int] = None,

        max_fraction_available_bytes: Optional[float] = None,

    ):
        self._max_memory_bytes = max_memory_bytes
        self._max_fraction_available_bytes = max_fraction_available_bytes
        self._total_memory_bytes = 0
        self._usage_bytes = {}  # type: Dict[K, int]
        self._last_accessed = {}  # type: Dict[K, datetime]
        self._created = {}  # type: Dict[K, datetime]

    def can_archive(self) -> bool:
        return len(self._last_accessed) > 0

    def should_archive(self) -> bool:
        return (
            self._max_memory_bytes is not None and self._total_memory_bytes > self._max_memory_bytes
        ) or (
            self._max_fraction_available_bytes is not None
            and self._total_memory_bytes
            > virtual_memory().available * self._max_fraction_available_bytes
        )

    def reindex(self, items: Mapping[K, V]) -> None:
        for key in set(self._last_accessed.keys()) - set(items.keys()):
            if key not in items.keys():
                self.removed(key)
        for key, value in items.items():
            self._usage_bytes[key] = sys.getsizeof(value)
        self._total_memory_bytes = sum(self._usage_bytes.values())

    def accessed(self, key: K) -> None:
        self._last_accessed[key] = datetime.now()

    def added(self, key: K, value: V) -> None:
        now = datetime.now()
        self._created[key] = now
        self._last_accessed[key] = now
        self._usage_bytes[key] = sys.getsizeof(value)
        self._total_memory_bytes += self._usage_bytes[key]

    def removed(self, key: K) -> None:
        self._total_memory_bytes -= self._usage_bytes[key]
        del self._last_accessed[key]
        del self._created[key]
        del self._usage_bytes[key]

    def __str__(self):
        available = virtual_memory().available
        name = self.__class__.__name__
        n_items = len(self._usage_bytes)
        real = nicesize(self._total_memory_bytes)
        cap_raw = "-" if self._max_memory_bytes is None else nicesize(self._max_memory_bytes)
        cap_percent = (
            "-"
            if self._max_fraction_available_bytes is None
            else nicesize(available * self._max_fraction_available_bytes)
        )
        percent = (
            "-"
            if self._max_fraction_available_bytes is None
            else round(
                100 * self._total_memory_bytes / (available * self._max_fraction_available_bytes),
                3,
            )
        )
        return f"{name}(n={n_items}, {real}/{cap_raw}, {real}/{cap_percent}={percent}%"

    def __repr__(self):
        ordered = list(self.items())
        ss = []

        current_day = None
        for k in ordered:
            dt = self._last_accessed[k]

            if current_day is None or current_day.date() != dt.date():
                current_day = dt
                ss.append("#" + current_day.strftime("%Y-%m-%d") + "...")
            nice = nicesize(self._usage_bytes[k])
            access = self._last_accessed[k].strftime("%H:%M:%S")
            ss.append(f"{k}:{nice}@{access}")
        return f"{str(self)}@{hex(id(self))}: [{', '.join(ss)}]"


class MemoryLruPolicy(MemoryLimitingPolicy):

    def items(self) -> Iterator[K]:
        return iter([k for k, v in sorted(self._last_accessed.items(), key=operator.itemgetter(1))])


class MemoryMruPolicy(MemoryLimitingPolicy):

    def items(self) -> Iterator[K]:
        rev = reversed(sorted(self._last_accessed.items(), key=operator.itemgetter(1)))
        return iter([k for k, v in rev])


class MemCache(Generic[K, V]):

    def __init__(
        self, loader: Callable[[K], V], policy: MemCachePolicy, *, log: Callable[[str], Any]

    ):
        self._loader = loader
        self._items = {}  # type: Dict[K, V]
        self._policy = policy
        self._log = lambda _: None if log is None else log

    def __getitem__(self, key: K) -> V:
        self._policy.accessed(key)
        if key in self._items:

            return self._items[key]
        else:
            value = self._loader(key)
            self._log(f"Loaded {key}")
            self._items[key] = value
            self._policy.added(key, value)
            self.archive()
            return value

    def __call__(self, key: K) -> V:
        return self[key]

    def archive(self, at_least: Optional[int] = None) -> Sequence[K]:

        it = self._policy.items()

        archived = []
        while self._policy.can_archive() and (
            at_least is not None and len(archived) < at_least or self._policy.should_archive()

        ):
            key = next(it)
            self._policy.removed(key)
            del self._items[key]
            archived.append(key)
            self._log(f"Archived {len(archived)} items: {archived}")
        return archived

    def clear(self) -> None:

        it = self._policy.items()

        cleared = []
        while self._policy.can_archive():
            key = next(it)
            self._policy.removed(key)
            del self._items[key]
            cleared.append(key)
        self._log(f"Cleared {len(cleared)} items: {cleared}")

    def remove(self, key: K) -> None:

        if key in self:
            self._policy.removed(key)
            del self._items[key]

    def __contains__(self, key: K):
        return key in self._items

    def __delitem__(self, key):
        self.remove(key)

    def __repr__(self):
        return str(self) + "@" + hex(id(self))

    def __str__(self):
        return f"{self.__class__.__name__}({repr(self._policy)})"


__all__ = [
    "MemCachePolicy",
    "MemCache",
    "MemoryLimitingPolicy",
    "MemoryLruPolicy",
    "MemoryMruPolicy",
]


1			import operator
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			import sys
3			from abc import ABC
4			from datetime import datetime
5			from typing import Any, Callable, Dict, Generic, Iterator, Mapping, Optional, Sequence, TypeVar
6
7			from psutil import virtual_memory
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Unable to import 'psutil' Loading history...
8
9			# noinspection PyProtectedMember
10			from pocketutils.core._internal import nicesize
11
12			K = TypeVar("K", covariant=True)
			0 ignored issues – show Coding Style Naming introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Class name "K" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
13			V = TypeVar("V", contravariant=True)
			0 ignored issues – show Coding Style Naming introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Class name "V" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
14
15
16			class MemCachePolicy(Generic[K, V]):
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
17			def should_archive(self) -> bool:
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
18			raise NotImplementedError()
19
20			def can_archive(self) -> bool:
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
21			raise NotImplementedError()
22
23			def reindex(self, items: Mapping[K, V]) -> None:
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
24			raise NotImplementedError()
25
26			def items(self) -> Iterator[K]:
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
27			# TODO this is not overloaded!!
			0 ignored issues – show Coding Style introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
28			raise NotImplementedError()
29
30			def accessed(self, key: K) -> None:
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
31			pass
32
33			def added(self, key: K, value: V) -> None:
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
34			pass
35
36			def removed(self, key: K) -> None:
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
37			pass
38
39
40			class MemoryLimitingPolicy(MemCachePolicy, ABC):
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
41			def __init__(
42			self,
			0 ignored issues – show Coding Style introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
43			max_memory_bytes: Optional[int] = None,
			0 ignored issues – show Coding Style introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
44			max_fraction_available_bytes: Optional[float] = None,
			0 ignored issues – show Coding Style introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
45			):
46			self._max_memory_bytes = max_memory_bytes
47			self._max_fraction_available_bytes = max_fraction_available_bytes
48			self._total_memory_bytes = 0
49			self._usage_bytes = {} # type: Dict[K, int]
50			self._last_accessed = {} # type: Dict[K, datetime]
51			self._created = {} # type: Dict[K, datetime]
52
53			def can_archive(self) -> bool:
54			return len(self._last_accessed) > 0
55
56			def should_archive(self) -> bool:
57			return (
58			self._max_memory_bytes is not None and self._total_memory_bytes > self._max_memory_bytes
59			) or (
60			self._max_fraction_available_bytes is not None
61			and self._total_memory_bytes
62			> virtual_memory().available * self._max_fraction_available_bytes
63			)
64
65			def reindex(self, items: Mapping[K, V]) -> None:
66			for key in set(self._last_accessed.keys()) - set(items.keys()):
67			if key not in items.keys():
68			self.removed(key)
69			for key, value in items.items():
70			self._usage_bytes[key] = sys.getsizeof(value)
71			self._total_memory_bytes = sum(self._usage_bytes.values())
72
73			def accessed(self, key: K) -> None:
74			self._last_accessed[key] = datetime.now()
75
76			def added(self, key: K, value: V) -> None:
77			now = datetime.now()
78			self._created[key] = now
79			self._last_accessed[key] = now
80			self._usage_bytes[key] = sys.getsizeof(value)
81			self._total_memory_bytes += self._usage_bytes[key]
82
83			def removed(self, key: K) -> None:
84			self._total_memory_bytes -= self._usage_bytes[key]
85			del self._last_accessed[key]
86			del self._created[key]
87			del self._usage_bytes[key]
88
89			def __str__(self):
90			available = virtual_memory().available
91			name = self.__class__.__name__
92			n_items = len(self._usage_bytes)
93			real = nicesize(self._total_memory_bytes)
94			cap_raw = "-" if self._max_memory_bytes is None else nicesize(self._max_memory_bytes)
95			cap_percent = (
96			"-"
97			if self._max_fraction_available_bytes is None
98			else nicesize(available * self._max_fraction_available_bytes)
99			)
100			percent = (
101			"-"
102			if self._max_fraction_available_bytes is None
103			else round(
104			100 * self._total_memory_bytes / (available * self._max_fraction_available_bytes),
105			3,
106			)
107			)
108			return f"{name}(n={n_items}, {real}/{cap_raw}, {real}/{cap_percent}={percent}%"
109
110			def __repr__(self):
111			ordered = list(self.items())
112			ss = []
			0 ignored issues – show Coding Style Naming introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Variable name "ss" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
113			current_day = None
114			for k in ordered:
115			dt = self._last_accessed[k]
			0 ignored issues – show Coding Style Naming introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Variable name "dt" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
116			if current_day is None or current_day.date() != dt.date():
117			current_day = dt
118			ss.append("#" + current_day.strftime("%Y-%m-%d") + "...")
119			nice = nicesize(self._usage_bytes[k])
120			access = self._last_accessed[k].strftime("%H:%M:%S")
121			ss.append(f"{k}:{nice}@{access}")
122			return f"{str(self)}@{hex(id(self))}: [{', '.join(ss)}]"
123
124
125			class MemoryLruPolicy(MemoryLimitingPolicy):
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
126			def items(self) -> Iterator[K]:
127			return iter([k for k, v in sorted(self._last_accessed.items(), key=operator.itemgetter(1))])
128
129
130			class MemoryMruPolicy(MemoryLimitingPolicy):
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
131			def items(self) -> Iterator[K]:
132			rev = reversed(sorted(self._last_accessed.items(), key=operator.itemgetter(1)))
133			return iter([k for k, v in rev])
134
135
136			class MemCache(Generic[K, V]):
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
137			def __init__(
138			self, loader: Callable[[K], V], policy: MemCachePolicy, *, log: Callable[[str], Any]
			0 ignored issues – show Coding Style introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
139			):
140			self._loader = loader
141			self._items = {} # type: Dict[K, V]
142			self._policy = policy
143			self._log = lambda _: None if log is None else log
144
145			def __getitem__(self, key: K) -> V:
146			self._policy.accessed(key)
147			if key in self._items:
			0 ignored issues – show unused-code introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
148			return self._items[key]
149			else:
150			value = self._loader(key)
151			self._log(f"Loaded {key}")
152			self._items[key] = value
153			self._policy.added(key, value)
154			self.archive()
155			return value
156
157			def __call__(self, key: K) -> V:
158			return self[key]
159
160			def archive(self, at_least: Optional[int] = None) -> Sequence[K]:
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
161			it = self._policy.items()
			0 ignored issues – show Coding Style Naming introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Variable name "it" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
162			archived = []
163			while self._policy.can_archive() and (
164			at_least is not None and len(archived) < at_least or self._policy.should_archive()
			0 ignored issues – show Coding Style introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
165			):
166			key = next(it)
167			self._policy.removed(key)
168			del self._items[key]
169			archived.append(key)
170			self._log(f"Archived {len(archived)} items: {archived}")
171			return archived
172
173			def clear(self) -> None:
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
174			it = self._policy.items()
			0 ignored issues – show Coding Style Naming introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Variable name "it" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
175			cleared = []
176			while self._policy.can_archive():
177			key = next(it)
178			self._policy.removed(key)
179			del self._items[key]
180			cleared.append(key)
181			self._log(f"Cleared {len(cleared)} items: {cleared}")
182
183			def remove(self, key: K) -> None:
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
184			if key in self:
185			self._policy.removed(key)
186			del self._items[key]
187
188			def __contains__(self, key: K):
189			return key in self._items
190
191			def __delitem__(self, key):
192			self.remove(key)
193
194			def __repr__(self):
195			return str(self) + "@" + hex(id(self))
196
197			def __str__(self):
198			return f"{self.__class__.__name__}({repr(self._policy)})"
199
200
201			__all__ = [
202			"MemCachePolicy",
203			"MemCache",
204			"MemoryLimitingPolicy",
205			"MemoryLruPolicy",
206			"MemoryMruPolicy",
207			]
208

dmyersturnbull / pocketutils

Push — main ( b312d0...29adcf )

pocketutils.misc.mem_cache.MemCache.remove() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like