BaseCache.__delitem__() - Code Metrics - Inspection of "update doc, improve event source" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-0.5.2 ( b6a9d6...3bdb98 )

by Felipe A.

created 2016-12-01 16:37 UTC

BaseCache.delitem() A

↳ Parent: BaseCache

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	2
rs	10



import functools
import threading


NOT_FOUND = object()


class HashTuple(tuple):
    '''
    Tuple object with cached __hash__ and __repr__.
    '''

    def __hash__(self):
        '''
        Self-caching hash method.

        x.__hash__() <==> hash(x)
        '''
        res = tuple.__hash__(self)
        self.__hash__ = lambda x: res
        return res

    def __repr__(self):
        '''
        Self-caching representation method.

        x.__repr__() <==> repr(x)
        '''
        res = (
            '<{0.__class__.__module__}.{0.__class__.__name__} {1} {2}>'
            ).format(self, hash(self), tuple(self))
        self.__repr__ = lambda x: res
        return res


class MemoizeManager(object):
    '''
    Cache manager object exposed via :attr:`cache` of memoized functions (see
    :meth:`BaseCache.memoize`).
    '''
    hashtuple_class = HashTuple
    NOT_CACHED = object()

    def __init__(self, cache):
        self._hits = 0
        self._misses = 0
        self._cache = cache

    @property
    def hits(self):
        '''
        :returns: number of times result was cached
        :rtype: integer
        '''
        return self._hits

    @property
    def misses(self):
        '''
        :returns: number of times result was not cached
        :rtype: integer
        '''
        return self._missses

    @classmethod
    def make_key(cls, args, kwargs, tuple=tuple):
        '''
        Make a cache key from optionally typed positional and keyword arguments

        The key is constructed in a way that is flat as possible rather than
        as a nested structure that would take more memory.

        If there is only a single argument and its data type is known to cache
        its hash value, then that argument is returned without a wrapper.  This
        saves space and improves lookup speed.
        '''
        key_args = tuple([(arg, arg.__class__) for arg in args])
        key_kwargs = tuple([
            (key, value, value.__class__)
            for key, value in sorted(kwargs.items())
            ])
        return cls.hashtuple_class((key_args, key_kwargs))

    def run(self, fnc, *args, **kwargs):
        key = self.make_key(args, kwargs)
        result = self._cache.get(key, self.NOT_CACHED)
        if result is not self.NOT_CACHED:
            self._hits += 1
            return result
        self._misses += 1
        self._cache[key] = result = fnc(*args, **kwargs)
        return result

    def pop(self, *args, **kwargs):
        key = self.make_key(args, kwargs)
        return self._cache.pop(key, None)

    def clear(self):
        '''
        Clear both cache and statistics.
        '''
        self._hits = 0
        self._misses = 0
        self._cache.clear()


class BaseCache(object):
    '''
    Abstract cache implementation.

    This class does not really store anything, this is let to implementations.
    '''
    memoize_class = MemoizeManager

    def __setitem__(self, key, value):
        pass

    def __getitem__(self, key):
        raise KeyError(key)

    def __delitem__(self, key):
        raise KeyError(key)

    def __contains__(self, key):
        return False

    def pop(self, key, default=NOT_FOUND):
        if default is NOT_FOUND:
            raise KeyError(key)
        return default

    def get(self, key, default=None):
        return default

    def clear(self):
        pass

    def memoize(self, fnc):
        '''
        Decorate given function and memoize its results using currente cache.

        :param fnc: function to decorate
        :type fnc: collections.abc.Callable
        :param maxsize: optional maximum size (defaults to 1024)
        :type maxsize: int
        :returns: wrapped cached function
        :rtype: function
        '''
        manager = self.memoize_class(self)

        @functools.wraps(fnc)
        def wrapped(*args, **kwargs):
            return manager.run(fnc, *args, **kwargs)
        wrapped.cache = manager
        return wrapped


class LRUCache(BaseCache):
    '''
    Cache object evicting least recently used objects (LRU) when maximum cache
    size is reached.

    It's way slower than python's :func:`functools.lru_cache` but provides an
    object-oriented implementation and manual cache eviction.
    '''
    lock_class = threading.RLock

    @property
    def full(self):
        '''
        :returns: True if size reached maxsize, False otherwise
        :rtype: bool
        '''
        return self._full

    @property
    def size(self):
        '''
        :returns: current size of result cache
        :rtype: int
        '''
        return len(self._cache)

    @property
    def maxsize(self):
        '''
        :returns: maximum size of result cache
        :rtype: int
        '''
        return self._maxsize

    def __init__(self, maxsize=1024):
        '''
        :param fnc:
        :type fnc: collections.abc.Callable
        :param maxsize: optional maximum cache size (defaults to 1024)
        :type maxsize: int
        '''
        self._cache = {}
        self._full = False
        self._maxsize = maxsize
        self._lock = self.lock_class()
        self._root = root = []
        root[:] = [root, root, None, None]

    def _bumpitem(self, key):
        PREV, NEXT = 0, 1
        root = self._root
        last = root[PREV]
        last[NEXT] = root[PREV] = link = self._cache[key]
        link[PREV] = last
        link[NEXT] = root

    def _getitem(self, key, default=NOT_FOUND):
        RESULT = 3
        link = self._cache.get(key)
        if link is not None:
            self._bumpitem(key)
            return link[RESULT]
        if default is NOT_FOUND:
            raise KeyError(key)
        return default

    def _setitem(self, key, value):
        PREV, NEXT, KEY, RESULT = 0, 1, 2, 3
        cache = self._cache
        root = self._root
        if key in cache:
            self._bumpitem(key)
            root[PREV][RESULT] = value
        elif self._full:
            # reuse old root
            oldroot = root
            oldroot[KEY] = key
            oldroot[RESULT] = value
            root = oldroot[NEXT]
            oldkey = root[KEY]
            root[KEY] = root[RESULT] = None
            del cache[oldkey]
            cache[key] = oldroot
            self._root = root
        else:
            last = root[PREV]
            link = [last, root, key, value]
            last[NEXT] = root[PREV] = cache[key] = link
            self._full = (len(cache) >= self._maxsize)
        return value

    def _popitem(self, key, default=NOT_FOUND):
        PREV, NEXT = 0, 1
        if default is NOT_FOUND:
            link_prev, link_next, key, result = self._cache.pop(key)
        else:
            link_prev, link_next, key, result = self._cache.pop(key, default)
        link_prev[NEXT] = link_next
        link_next[PREV] = link_prev
        return result

    def __getitem__(self, key):
        with self._lock:
            return self._getitem(key)

    def __setitem__(self, key, value):
        with self._lock:
            return self._setitem(key, value)

    def __delitem__(self, key):
        with self._lock:
            self._popitem(key)

    def __contains__(self, key):
        return key in self._cache

    def pop(self, key, default=NOT_FOUND):
        with self._lock:
            return self._popitem(key, default)

    def get(self, key, default=None):
        with self._lock:
            return self._getitem(key, default)

    def clear(self):
        with self._lock:
            self._cache.clear()
            self._root = root = []
            root[:] = [root, root, None, None]
            self._hits = 0
            self._misses = 0
            self._full = False


1
2
3			import functools
4			import threading
5
6
7			NOT_FOUND = object()
8
9
10			class HashTuple(tuple):
11			'''
12			Tuple object with cached __hash__ and __repr__.
13			'''
14
15			def __hash__(self):
16			'''
17			Self-caching hash method.
18
19			x.__hash__() <==> hash(x)
20			'''
21			res = tuple.__hash__(self)
22			self.__hash__ = lambda x: res
23			return res
24
25			def __repr__(self):
26			'''
27			Self-caching representation method.
28
29			x.__repr__() <==> repr(x)
30			'''
31			res = (
32			'<{0.__class__.__module__}.{0.__class__.__name__} {1} {2}>'
33			).format(self, hash(self), tuple(self))
34			self.__repr__ = lambda x: res
35			return res
36
37
38			class MemoizeManager(object):
39			'''
40			Cache manager object exposed via :attr:`cache` of memoized functions (see
41			:meth:`BaseCache.memoize`).
42			'''
43			hashtuple_class = HashTuple
44			NOT_CACHED = object()
45
46			def __init__(self, cache):
47			self._hits = 0
48			self._misses = 0
49			self._cache = cache
50
51			@property
52			def hits(self):
53			'''
54			:returns: number of times result was cached
55			:rtype: integer
56			'''
57			return self._hits
58
59			@property
60			def misses(self):
61			'''
62			:returns: number of times result was not cached
63			:rtype: integer
64			'''
65			return self._missses
66
67			@classmethod
68			def make_key(cls, args, kwargs, tuple=tuple):
69			'''
70			Make a cache key from optionally typed positional and keyword arguments
71
72			The key is constructed in a way that is flat as possible rather than
73			as a nested structure that would take more memory.
74
75			If there is only a single argument and its data type is known to cache
76			its hash value, then that argument is returned without a wrapper. This
77			saves space and improves lookup speed.
78			'''
79			key_args = tuple([(arg, arg.__class__) for arg in args])
80			key_kwargs = tuple([
81			(key, value, value.__class__)
82			for key, value in sorted(kwargs.items())
83			])
84			return cls.hashtuple_class((key_args, key_kwargs))
85
86			def run(self, fnc, args, *kwargs):
87			key = self.make_key(args, kwargs)
88			result = self._cache.get(key, self.NOT_CACHED)
89			if result is not self.NOT_CACHED:
90			self._hits += 1
91			return result
92			self._misses += 1
93			self._cache[key] = result = fnc(args, *kwargs)
94			return result
95
96			def pop(self, args, *kwargs):
97			key = self.make_key(args, kwargs)
98			return self._cache.pop(key, None)
99
100			def clear(self):
101			'''
102			Clear both cache and statistics.
103			'''
104			self._hits = 0
105			self._misses = 0
106			self._cache.clear()
107
108
109			class BaseCache(object):
110			'''
111			Abstract cache implementation.
112
113			This class does not really store anything, this is let to implementations.
114			'''
115			memoize_class = MemoizeManager
116
117			def __setitem__(self, key, value):
118			pass
119
120			def __getitem__(self, key):
121			raise KeyError(key)
122
123			def __delitem__(self, key):
124			raise KeyError(key)
125
126			def __contains__(self, key):
127			return False
128
129			def pop(self, key, default=NOT_FOUND):
130			if default is NOT_FOUND:
131			raise KeyError(key)
132			return default
133
134			def get(self, key, default=None):
135			return default
136
137			def clear(self):
138			pass
139
140			def memoize(self, fnc):
141			'''
142			Decorate given function and memoize its results using currente cache.
143
144			:param fnc: function to decorate
145			:type fnc: collections.abc.Callable
146			:param maxsize: optional maximum size (defaults to 1024)
147			:type maxsize: int
148			:returns: wrapped cached function
149			:rtype: function
150			'''
151			manager = self.memoize_class(self)
152
153			@functools.wraps(fnc)
154			def wrapped(args, *kwargs):
155			return manager.run(fnc, args, *kwargs)
156			wrapped.cache = manager
157			return wrapped
158
159
160			class LRUCache(BaseCache):
161			'''
162			Cache object evicting least recently used objects (LRU) when maximum cache
163			size is reached.
164
165			It's way slower than python's :func:`functools.lru_cache` but provides an
166			object-oriented implementation and manual cache eviction.
167			'''
168			lock_class = threading.RLock
169
170			@property
171			def full(self):
172			'''
173			:returns: True if size reached maxsize, False otherwise
174			:rtype: bool
175			'''
176			return self._full
177
178			@property
179			def size(self):
180			'''
181			:returns: current size of result cache
182			:rtype: int
183			'''
184			return len(self._cache)
185
186			@property
187			def maxsize(self):
188			'''
189			:returns: maximum size of result cache
190			:rtype: int
191			'''
192			return self._maxsize
193
194			def __init__(self, maxsize=1024):
195			'''
196			:param fnc:
197			:type fnc: collections.abc.Callable
198			:param maxsize: optional maximum cache size (defaults to 1024)
199			:type maxsize: int
200			'''
201			self._cache = {}
202			self._full = False
203			self._maxsize = maxsize
204			self._lock = self.lock_class()
205			self._root = root = []
206			root[:] = [root, root, None, None]
207
208			def _bumpitem(self, key):
209			PREV, NEXT = 0, 1
210			root = self._root
211			last = root[PREV]
212			last[NEXT] = root[PREV] = link = self._cache[key]
213			link[PREV] = last
214			link[NEXT] = root
215
216			def _getitem(self, key, default=NOT_FOUND):
217			RESULT = 3
218			link = self._cache.get(key)
219			if link is not None:
220			self._bumpitem(key)
221			return link[RESULT]
222			if default is NOT_FOUND:
223			raise KeyError(key)
224			return default
225
226			def _setitem(self, key, value):
227			PREV, NEXT, KEY, RESULT = 0, 1, 2, 3
228			cache = self._cache
229			root = self._root
230			if key in cache:
231			self._bumpitem(key)
232			root[PREV][RESULT] = value
233			elif self._full:
234			# reuse old root
235			oldroot = root
236			oldroot[KEY] = key
237			oldroot[RESULT] = value
238			root = oldroot[NEXT]
239			oldkey = root[KEY]
240			root[KEY] = root[RESULT] = None
241			del cache[oldkey]
242			cache[key] = oldroot
243			self._root = root
244			else:
245			last = root[PREV]
246			link = [last, root, key, value]
247			last[NEXT] = root[PREV] = cache[key] = link
248			self._full = (len(cache) >= self._maxsize)
249			return value
250
251			def _popitem(self, key, default=NOT_FOUND):
252			PREV, NEXT = 0, 1
253			if default is NOT_FOUND:
254			link_prev, link_next, key, result = self._cache.pop(key)
255			else:
256			link_prev, link_next, key, result = self._cache.pop(key, default)
257			link_prev[NEXT] = link_next
258			link_next[PREV] = link_prev
259			return result
260
261			def __getitem__(self, key):
262			with self._lock:
263			return self._getitem(key)
264
265			def __setitem__(self, key, value):
266			with self._lock:
267			return self._setitem(key, value)
268
269			def __delitem__(self, key):
270			with self._lock:
271			self._popitem(key)
272
273			def __contains__(self, key):
274			return key in self._cache
275
276			def pop(self, key, default=NOT_FOUND):
277			with self._lock:
278			return self._popitem(key, default)
279
280			def get(self, key, default=None):
281			with self._lock:
282			return self._getitem(key, default)
283
284			def clear(self):
285			with self._lock:
286			self._cache.clear()
287			self._root = root = []
288			root[:] = [root, root, None, None]
289			self._hits = 0
290			self._misses = 0
291			self._full = False
292

ergoithz / browsepy

Push — dev-0.5.2 ( b6a9d6...3bdb98 )

BaseCache.__delitem__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

BaseCache.delitem() A