Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/auth/auth_cache.py: 18%

1import time 1 ctx1a

2from typing import Any 1 ctx1a

3 

4 

5class AuthCache: 1 ctx1a

6 def __init__(self, threshold: int = 500, default_timeout: float = 300): 1 ctx1a

7 self.default_timeout = default_timeout 

8 self._cache: dict[str, tuple[float, Any]] = {} 

9 self.clear = self._cache.clear 

10 self._threshold = threshold 

11 

12 def _prune(self): 1 ctx1a

13 if len(self._cache) > self._threshold: 

14 now = time.time() 

15 toremove = [] 

16 for idx, (key, (expires, _)) in enumerate(self._cache.items()): 

17 if (expires != 0 and expires <= now) or idx % 3 == 0: 

18 toremove.append(key) 

19 for key in toremove: 

20 self._cache.pop(key, None) 

21 

22 def _normalize_timeout(self, timeout: float | None) -> float: 1 ctx1a

23 if timeout is None: 

24 timeout = self.default_timeout 

25 if timeout > 0: 

26 timeout = time.time() + timeout 

27 return timeout 

28 

29 async def get(self, key: str) -> Any: 1 ctx1a

30 try: 

31 expires, value = self._cache[key] 

32 if expires == 0 or expires > time.time(): 

33 return value 

34 except KeyError: 

35 return None 

36 

37 async def set(self, key: str, value: Any, timeout: float | None = None) -> bool: 1 ctx1a

38 expires = self._normalize_timeout(timeout) 

39 self._prune() 

40 self._cache[key] = (expires, value) 

41 return True 

42 

43 async def delete(self, key: str) -> bool: 1 ctx1a

44 return self._cache.pop(key, None) is not None 

45 

46 async def has(self, key: str) -> bool: 1 ctx1a

47 try: 

48 expires, value = self._cache[key] 

49 return expires == 0 or expires > time.time() 

50 except KeyError: 

51 return False