decorated_function() - Code Metrics - Inspection of "ongoing extension refactor, favicons" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — 5.2-unstable ( 1c6b27 )

by Felipe A.

created 2016-12-09 13:40 UTC

decorated_function() A

↳ Parent: cachedview()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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


import time
import threading

from functools import wraps
from flask import request, current_app
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


def cachedview(timeout=5 * 60, key='view/%s'):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            cache_key = key % request.path
            cache = current_app.extensions['plugin_manager'].cache
            rv = cache.get(cache_key)
            if rv is not None:
                return rv
            rv = f(*args, **kwargs)
            cache.set(cache_key, rv, timeout=timeout)
            return rv
        return decorated_function
    return decorator


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

ergoithz / browsepy

Push — 5.2-unstable ( 1c6b27 )

decorated_function() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like