Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/core/cache/backends/locmem.py: 25%

1"Thread-safe in-memory cache backend." 

2import pickle 

3import time 

4from collections import OrderedDict 

5from threading import Lock 

6 

7from django.core.cache.backends.base import DEFAULT_TIMEOUT, BaseCache 

8 

9# Global in-memory store of cache data. Keyed by name, to provide 

10# multiple named local memory caches. 

11_caches = {} 

12_expire_info = {} 

13_locks = {} 

14 

15 

16class LocMemCache(BaseCache): 

17 pickle_protocol = pickle.HIGHEST_PROTOCOL 

18 

19 def __init__(self, name, params): 

20 super().__init__(params) 

21 self._cache = _caches.setdefault(name, OrderedDict()) 

22 self._expire_info = _expire_info.setdefault(name, {}) 

23 self._lock = _locks.setdefault(name, Lock()) 

24 

25 def add(self, key, value, timeout=DEFAULT_TIMEOUT, version=None): 

26 key = self.make_and_validate_key(key, version=version) 

27 pickled = pickle.dumps(value, self.pickle_protocol) 

28 with self._lock: 

29 if self._has_expired(key): 

30 self._set(key, pickled, timeout) 

31 return True 

32 return False 

33 

34 def get(self, key, default=None, version=None): 

35 key = self.make_and_validate_key(key, version=version) 

36 with self._lock: 

37 if self._has_expired(key): 

38 self._delete(key) 

39 return default 

40 pickled = self._cache[key] 

41 self._cache.move_to_end(key, last=False) 

42 return pickle.loads(pickled) 

43 

44 def _set(self, key, value, timeout=DEFAULT_TIMEOUT): 

45 if len(self._cache) >= self._max_entries: 

46 self._cull() 

47 self._cache[key] = value 

48 self._cache.move_to_end(key, last=False) 

49 self._expire_info[key] = self.get_backend_timeout(timeout) 

50 

51 def set(self, key, value, timeout=DEFAULT_TIMEOUT, version=None): 

52 key = self.make_and_validate_key(key, version=version) 

53 pickled = pickle.dumps(value, self.pickle_protocol) 

54 with self._lock: 

55 self._set(key, pickled, timeout) 

56 

57 def touch(self, key, timeout=DEFAULT_TIMEOUT, version=None): 

58 key = self.make_and_validate_key(key, version=version) 

59 with self._lock: 

60 if self._has_expired(key): 

61 return False 

62 self._expire_info[key] = self.get_backend_timeout(timeout) 

63 return True 

64 

65 def incr(self, key, delta=1, version=None): 

66 key = self.make_and_validate_key(key, version=version) 

67 with self._lock: 

68 if self._has_expired(key): 

69 self._delete(key) 

70 raise ValueError("Key '%s' not found" % key) 

71 pickled = self._cache[key] 

72 value = pickle.loads(pickled) 

73 new_value = value + delta 

74 pickled = pickle.dumps(new_value, self.pickle_protocol) 

75 self._cache[key] = pickled 

76 self._cache.move_to_end(key, last=False) 

77 return new_value 

78 

79 def has_key(self, key, version=None): 

80 key = self.make_and_validate_key(key, version=version) 

81 with self._lock: 

82 if self._has_expired(key): 

83 self._delete(key) 

84 return False 

85 return True 

86 

87 def _has_expired(self, key): 

88 exp = self._expire_info.get(key, -1) 

89 return exp is not None and exp <= time.time() 

90 

91 def _cull(self): 

92 if self._cull_frequency == 0: 

93 self._cache.clear() 

94 self._expire_info.clear() 

95 else: 

96 count = len(self._cache) // self._cull_frequency 

97 for i in range(count): 

98 key, _ = self._cache.popitem() 

99 del self._expire_info[key] 

100 

101 def _delete(self, key): 

102 try: 

103 del self._cache[key] 

104 del self._expire_info[key] 

105 except KeyError: 

106 return False 

107 return True 

108 

109 def delete(self, key, version=None): 

110 key = self.make_and_validate_key(key, version=version) 

111 with self._lock: 

112 return self._delete(key) 

113 

114 def clear(self): 

115 with self._lock: 

116 self._cache.clear() 

117 self._expire_info.clear()