MultiCache.delete() - Code Metrics - Inspection of "implement browse endpoint cache" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — 5.2-unstable ( ee49a9...946858 )

by Felipe A.

created 2016-12-07 14:50 UTC

MultiCache.delete() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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


import time
import threading

from werkzeug.contrib.cache import BaseCache


NOT_FOUND = object()


class SimpleLRUCache(BaseCache):
    '''
    Simple memory cache for simgle process environments. Thread safe using
    :attr:`lock_class` (threading.Lock)

    Cache object evicting least recently used objects (LRU) when maximum cache
    size is reached, so keys could be discarded independently of their
    timeout.

    '''
    lock_class = threading.Lock
    time_func = time.time

    @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, default_timeout=300, maxsize=1024):
        self._default_timeout = default_timeout
        self._cache = {}
        self._full = False
        self._maxsize = maxsize
        self._lock = self.lock_class()
        self._key = None

    def _extract(self, link):
        PREV, NEXT, KEY = 0, 1, 2
        prev = link[PREV]
        if prev is link:
            self._key = None
            return link
        next = link[NEXT]
        prev[NEXT] = next
        next[PREV] = prev
        if link[KEY] == self._key:
            self._key = next[KEY]
        return link

    def _insert(self, link):
        PREV, NEXT, KEY = 0, 1, 2
        if self._cache:
            next = self._cache[self._key]
            prev = next[PREV]
            link[:2] = (prev, next)
            prev[NEXT] = link
            next[PREV] = link
        else:
            link[:2] = (link, link)
            self._key = link[KEY]
        return link

    def _shift(self, pop=False):
        NEXT, KEY = 1, 2
        if pop:
            link = self._cache.pop(self._key)
        else:
            link = self._cache[self._key]
        self._key = link[NEXT][KEY]
        return link

    def _bump(self, link):
        NEXT, KEY = 1, 2
        if link[KEY] == self._key:
            return self._shift()
        if link[NEXT][KEY] == self._key:
            return link
        return self._insert(self._extract(link))

    def _getitem(self, key, default=NOT_FOUND):
        VALUE, EXPIRE = 3, 4
        link = self._cache.get(key)
        if link is not None and link[EXPIRE] > self.time_func():
            self._bump(link)
            return link[VALUE]
        return default

    def _setitem(self, key, value, timeout=None):
        KEY, VALUE = 2, 3
        cache = self._cache
        expire = self.time_func() + (
            self._default_timeout
            if timeout is None else
            timeout
            )
        link = self._cache.get(key)
        if link:
            link = self._bump(link)
            link[VALUE:] = (value, expire)
        elif self._full:
            link = self._shift(pop=True)
            link[KEY:] = (key, value, expire)
            self._cache[key] = link
        else:
            self._cache[key] = self._insert([None, None, key, value, expire])
            self._full = (len(cache) >= self._maxsize)
        return value

    def _popitem(self, key, default=NOT_FOUND):
        VALUE = 3
        link = self._cache.pop(key, None)
        return self._extract(link)[VALUE] if link else default

    def add(self, key, value, timeout=None):
        with self._lock:
            if key in self._cache:
                return False
            self._setitem(key, value, timeout)
            return True

    def set(self, key, value, timeout=None):
        with self._lock:
            self._setitem(key, value, timeout)
            return True

    def set_many(self, mapping, timeout=None):
        with self._lock:
            for key, value in mapping.items():
                self._setitem(key, value, timeout)
            return True

    def inc(self, key, delta=1):
        VALUE = 3
        with self._lock:
            return self._setitem(key, self._cache.get(key, 0)[VALUE] + delta)

    def dec(self, key, delta=1):
        return self.inc(key, delta=-delta)

    def delete(self, key):
        with self._lock:
            return self._popitem(key) is not NOT_FOUND

    def delete_many(self, *keys):
        with self._lock:
            return NOT_FOUND not in [self._popitem(key) for key in keys]

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

    def get_dict(self, *keys):
        with self._lock:
            return {key: self._getitem(key, None) for key in keys}

    def get_many(self, *keys):
        with self._lock:
            return [self._getitem(key, None) for key in keys]

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

    def clear(self):
        '''
        Clear cache.
        '''
        with self._lock:
            self._cache.clear()
            self._full = False
            self._key = None
            return True


1
2			import time
3			import threading
4
5			from werkzeug.contrib.cache import BaseCache
6
7
8			NOT_FOUND = object()
9
10
11			class SimpleLRUCache(BaseCache):
12			'''
13			Simple memory cache for simgle process environments. Thread safe using
14			:attr:`lock_class` (threading.Lock)
15
16			Cache object evicting least recently used objects (LRU) when maximum cache
17			size is reached, so keys could be discarded independently of their
18			timeout.
19
20			'''
21			lock_class = threading.Lock
22			time_func = time.time
23
24			@property
25			def full(self):
26			'''
27			:returns: True if size reached maxsize, False otherwise
28			:rtype: bool
29			'''
30			return self._full
31
32			@property
33			def size(self):
34			'''
35			:returns: current size of result cache
36			:rtype: int
37			'''
38			return len(self._cache)
39
40			@property
41			def maxsize(self):
42			'''
43			:returns: maximum size of result cache
44			:rtype: int
45			'''
46			return self._maxsize
47
48			def __init__(self, default_timeout=300, maxsize=1024):
49			self._default_timeout = default_timeout
50			self._cache = {}
51			self._full = False
52			self._maxsize = maxsize
53			self._lock = self.lock_class()
54			self._key = None
55
56			def _extract(self, link):
57			PREV, NEXT, KEY = 0, 1, 2
58			prev = link[PREV]
59			if prev is link:
60			self._key = None
61			return link
62			next = link[NEXT]
63			prev[NEXT] = next
64			next[PREV] = prev
65			if link[KEY] == self._key:
66			self._key = next[KEY]
67			return link
68
69			def _insert(self, link):
70			PREV, NEXT, KEY = 0, 1, 2
71			if self._cache:
72			next = self._cache[self._key]
73			prev = next[PREV]
74			link[:2] = (prev, next)
75			prev[NEXT] = link
76			next[PREV] = link
77			else:
78			link[:2] = (link, link)
79			self._key = link[KEY]
80			return link
81
82			def _shift(self, pop=False):
83			NEXT, KEY = 1, 2
84			if pop:
85			link = self._cache.pop(self._key)
86			else:
87			link = self._cache[self._key]
88			self._key = link[NEXT][KEY]
89			return link
90
91			def _bump(self, link):
92			NEXT, KEY = 1, 2
93			if link[KEY] == self._key:
94			return self._shift()
95			if link[NEXT][KEY] == self._key:
96			return link
97			return self._insert(self._extract(link))
98
99			def _getitem(self, key, default=NOT_FOUND):
100			VALUE, EXPIRE = 3, 4
101			link = self._cache.get(key)
102			if link is not None and link[EXPIRE] > self.time_func():
103			self._bump(link)
104			return link[VALUE]
105			return default
106
107			def _setitem(self, key, value, timeout=None):
108			KEY, VALUE = 2, 3
109			cache = self._cache
110			expire = self.time_func() + (
111			self._default_timeout
112			if timeout is None else
113			timeout
114			)
115			link = self._cache.get(key)
116			if link:
117			link = self._bump(link)
118			link[VALUE:] = (value, expire)
119			elif self._full:
120			link = self._shift(pop=True)
121			link[KEY:] = (key, value, expire)
122			self._cache[key] = link
123			else:
124			self._cache[key] = self._insert([None, None, key, value, expire])
125			self._full = (len(cache) >= self._maxsize)
126			return value
127
128			def _popitem(self, key, default=NOT_FOUND):
129			VALUE = 3
130			link = self._cache.pop(key, None)
131			return self._extract(link)[VALUE] if link else default
132
133			def add(self, key, value, timeout=None):
134			with self._lock:
135			if key in self._cache:
136			return False
137			self._setitem(key, value, timeout)
138			return True
139
140			def set(self, key, value, timeout=None):
141			with self._lock:
142			self._setitem(key, value, timeout)
143			return True
144
145			def set_many(self, mapping, timeout=None):
146			with self._lock:
147			for key, value in mapping.items():
148			self._setitem(key, value, timeout)
149			return True
150
151			def inc(self, key, delta=1):
152			VALUE = 3
153			with self._lock:
154			return self._setitem(key, self._cache.get(key, 0)[VALUE] + delta)
155
156			def dec(self, key, delta=1):
157			return self.inc(key, delta=-delta)
158
159			def delete(self, key):
160			with self._lock:
161			return self._popitem(key) is not NOT_FOUND
162
163			def delete_many(self, *keys):
164			with self._lock:
165			return NOT_FOUND not in [self._popitem(key) for key in keys]
166
167			def get(self, key):
168			with self._lock:
169			return self._getitem(key, None)
170
171			def get_dict(self, *keys):
172			with self._lock:
173			return {key: self._getitem(key, None) for key in keys}
174
175			def get_many(self, *keys):
176			with self._lock:
177			return [self._getitem(key, None) for key in keys]
178
179			def has(self, key):
180			return key in self._cache
181
182			def clear(self):
183			'''
184			Clear cache.
185			'''
186			with self._lock:
187			self._cache.clear()
188			self._full = False
189			self._key = None
190			return True
191

ergoithz / browsepy

Push — 5.2-unstable ( ee49a9...946858 )

MultiCache.delete() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like