1 import typing as _t 2 3 4 class BaseCache: 5 """Baseclass for the cache systems. All the cache systems implement this 6 API or a superset of it. 7 8 :param default_timeout: the default timeout (in seconds) that is used if 9 no timeout is specified on :meth:`set`. A timeout 10 of 0 indicates that the cache never expires. 11 """ 12 13 def __init__(self, default_timeout: int = 300): 14 self.default_timeout = default_timeout 15 16 def _normalize_timeout(self, timeout: _t.Optional[int]) -> int: 17 if timeout is None: 18 timeout = self.default_timeout 19 return timeout 20 21 def get(self, key: str) -> _t.Any: 22 """Look up key in the cache and return the value for it. 23 24 :param key: the key to be looked up. 25 :returns: The value if it exists and is readable, else ``None``. 26 """ 27 return None 28 29 def delete(self, key: str) -> bool: 30 """Delete `key` from the cache. 31 32 :param key: the key to delete. 33 :returns: Whether the key existed and has been deleted. 34 :rtype: boolean 35 """ 36 return True 37 38 def get_many(self, *keys: str) -> _t.List[_t.Any]: 39 """Returns a list of values for the given keys. 40 For each key an item in the list is created:: 41 42 foo, bar = cache.get_many("foo", "bar") 43 44 Has the same error handling as :meth:`get`. 45 46 :param keys: The function accepts multiple keys as positional 47 arguments. 48 """ 49 return [self.get(k) for k in keys] 50 51 def get_dict(self, *keys: str) -> _t.Dict[str, _t.Any]: 52 """Like :meth:`get_many` but return a dict:: 53 54 d = cache.get_dict("foo", "bar") 55 foo = d["foo"] 56 bar = d["bar"] 57 58 :param keys: The function accepts multiple keys as positional 59 arguments. 60 """ 61 return dict(zip(keys, self.get_many(*keys))) # noqa: B905 62 63 def set( 64 self, key: str, value: _t.Any, timeout: _t.Optional[int] = None 65 ) -> _t.Optional[bool]: 66 """Add a new key/value to the cache (overwrites value, if key already 67 exists in the cache). 68 69 :param key: the key to set 70 :param value: the value for the key 71 :param timeout: the cache timeout for the key in seconds (if not 72 specified, it uses the default timeout). A timeout of 73 0 indicates that the cache never expires. 74 :returns: ``True`` if key has been updated, ``False`` for backend 75 errors. Pickling errors, however, will raise a subclass of 76 ``pickle.PickleError``. 77 :rtype: boolean 78 """ 79 return True 80 81 def add(self, key: str, value: _t.Any, timeout: _t.Optional[int] = None) -> bool: 82 """Works like :meth:`set` but does not overwrite the values of already 83 existing keys. 84 85 :param key: the key to set 86 :param value: the value for the key 87 :param timeout: the cache timeout for the key in seconds (if not 88 specified, it uses the default timeout). A timeout of 89 0 indicates that the cache never expires. 90 :returns: Same as :meth:`set`, but also ``False`` for already 91 existing keys. 92 :rtype: boolean 93 """ 94 return True 95 96 def set_many( 97 self, mapping: _t.Dict[str, _t.Any], timeout: _t.Optional[int] = None 98 ) -> _t.List[_t.Any]: 99 """Sets multiple keys and values from a mapping. 100 101 :param mapping: a mapping with the keys/values to set. 102 :param timeout: the cache timeout for the key in seconds (if not 103 specified, it uses the default timeout). A timeout of 104 0 indicates that the cache never expires. 105 :returns: A list containing all keys successfully set 106 :rtype: boolean 107 """ 108 set_keys = [] 109 for key, value in mapping.items(): 110 if self.set(key, value, timeout): 111 set_keys.append(key) 112 return set_keys 113 114 def delete_many(self, *keys: str) -> _t.List[_t.Any]: 115 """Deletes multiple keys at once. 116 117 :param keys: The function accepts multiple keys as positional 118 arguments. 119 :returns: A list containing all successfully deleted keys 120 :rtype: boolean 121 """ 122 deleted_keys = [] 123 for key in keys: 124 if self.delete(key): 125 deleted_keys.append(key) 126 return deleted_keys 127 128 def has(self, key: str) -> bool: 129 """Checks if a key exists in the cache without returning it. This is a 130 cheap operation that bypasses loading the actual data on the backend. 131 132 :param key: the key to check 133 """ 134 raise NotImplementedError( 135 "%s doesn't have an efficient implementation of `has`. That " 136 "means it is impossible to check whether a key exists without " 137 "fully loading the key's data. Consider using `self.get` " 138 "explicitly if you don't care about performance." 139 ) 140 141 def clear(self) -> bool: 142 """Clears the cache. Keep in mind that not all caches support 143 completely clearing the cache. 144 145 :returns: Whether the cache has been cleared. 146 :rtype: boolean 147 """ 148 return True 149 150 def inc(self, key: str, delta: int = 1) -> _t.Optional[int]: 151 """Increments the value of a key by `delta`. If the key does 152 not yet exist it is initialized with `delta`. 153 154 For supporting caches this is an atomic operation. 155 156 :param key: the key to increment. 157 :param delta: the delta to add. 158 :returns: The new value or ``None`` for backend errors. 159 """ 160 value = (self.get(key) or 0) + delta 161 return value if self.set(key, value) else None 162 163 def dec(self, key: str, delta: int = 1) -> _t.Optional[int]: 164 """Decrements the value of a key by `delta`. If the key does 165 not yet exist it is initialized with `-delta`. 166 167 For supporting caches this is an atomic operation. 168 169 :param key: the key to increment. 170 :param delta: the delta to subtract. 171 :returns: The new value or `None` for backend errors. 172 """ 173 value = (self.get(key) or 0) - delta 174 return value if self.set(key, value) else None 175 176 177 class NullCache(BaseCache): 178 """A cache that doesn't cache. This can be useful for unit testing. 179 180 :param default_timeout: a dummy parameter that is ignored but exists 181 for API compatibility with other caches. 182 """ 183 184 def has(self, key: str) -> bool: 185 return False 186