NodeCacheManager - Code Metrics - Inspection of "add event subsystem with fs events" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-0.5.2 ( 16c7f4...cb9d21 )

by Felipe A.

created 2016-11-30 16:43 UTC

NodeCacheManager A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	20
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
dl	0
loc	20
rs	10
c	1
b	0
f	0
wmc	4



import functools
import threading


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 LRUCache(object):
    '''
    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.
    '''
    NOT_FOUND = object()

    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 self.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 self.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


class MemoizeManager(object):
    cache_class = LRUCache
    hashtuple_class = HashTuple
    NOT_FOUND = object()

    def __init__(self, fnc, maxsize=1024):
        self._fnc = fnc
        self._hits = 0
        self._misses = 0
        self._cache = self.cache_class(maxsize)

    @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 __call__(self, *args, **kwargs):
        NOT_FOUND = self.NOT_FOUND
        key = self.make_key(args, kwargs)
        result = self._cache.get(key, NOT_FOUND)
        if result is not NOT_FOUND:
            self._hits += 1
            return result
        self._misses += 1
        self._cache[key] = result = self._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()

    @classmethod
    def wrap(cls, fnc, maxsize=1024):
        '''
        Decorate given function with a CancelableLRUCache instance.

        :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
        '''

        self = cls(fnc, maxsize)

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


def cached(func_or_size):
    def inner(func):
        size = 1024 if func is func_or_size else func_or_size
        return MemoizeManager.wrap(func, size)
    return inner(func_or_size) if callable(func_or_size) else inner


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

ergoithz / browsepy

Push — dev-0.5.2 ( 16c7f4...cb9d21 )

NodeCacheManager A

Complexity

Size/Duplication

Importance

Duplication Side-by-Side

Filter issues like