Coverage for hermes/backend/__init__.py: 100%

1from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Union 

2 

3 

4if TYPE_CHECKING: 

5 from .. import Mangler # @UnusedImport 

6 

7 

8__all__ = 'AbstractBackend', 

9 

10 

11class AbstractLock: 

12 ''' 

13 Base locking class. Implements context manger protocol. Mocks 

14 ``acquire`` and ``release`` i.e. it always acquires. 

15 ''' 

16 

17 key: str 

18 '''Implementation may be key-aware.''' 

19 

20 def __init__(self, key: str): 

21 self.key = key 

22 

23 def __enter__(self): 

24 '''Enter context manager by acquiring the distributed lock.''' 

25 

26 self.acquire() 

27 

28 def __exit__(self, type, value, traceback): 

29 '''Exit context manager by releasing the distributed lock.''' 

30 

31 self.release() 

32 

33 def acquire(self, wait = True) -> bool: 

34 ''' 

35 Acquire distributed lock. 

36 

37 :argument wait: Whether to wait for the lock. 

38 :return: Whether the lock was acquired. 

39 ''' 

40 

41 return True 

42 

43 def release(self): 

44 '''Release distributed lock.''' 

45 

46 

47class AbstractBackend: 

48 '''Base backend class. It's also the no-op implementation.''' 

49 

50 mangler: 'Mangler' 

51 '''Key manager responsible for creating keys, hashing and serialisation.''' 

52 

53 def __init__(self, mangler: 'Mangler'): 

54 self.mangler = mangler 

55 

56 def lock(self, key: str) -> AbstractLock: 

57 '''Create lock object for the key.''' 

58 

59 return AbstractLock(self.mangler.nameLock(key)) 

60 

61 def save(self, mapping: Dict[str, Any], *, ttl: Optional[int] = None): 

62 ''' 

63 Save cache entry. 

64 

65 :argument mapping: ``key``-``value`` mapping for a bulk save. 

66 :argument ttl: Cache entry time-to-live . 

67 ''' 

68 

69 def load(self, keys: Union[str, Iterable[str]]) -> Optional[Union[Any, Dict[str, Any]]]: 

70 ''' 

71 Load cache entry(ies). 

72 

73 Note, when handling a multiple key call, absent value keys 

74 should be excluded from resulting dictionary. 

75 ''' 

76 

77 return None if isinstance(keys, str) else {} 

78 

79 def remove(self, keys: Union[str, Iterable[str]]): 

80 '''Remove given keys.''' 

81 

82 def clean(self): 

83 '''Purge the backend storage.'''