Coverage for src/blob_dict/dict/proxy.py: 0%

1import sys 

2from collections.abc import Iterator, MutableMapping 

3from datetime import timedelta 

4from typing import Any, Literal, override 

5 

6from cachetools import TTLCache 

7 

8from ..blob import BytesBlob 

9from . import MutableBlobDictBase 

10 

11 

12class ProxyBlobDict(MutableBlobDictBase): 

13 __EXTERNAL_DICT_TTL_ERROR_MESSAGE: str = "Cannot specify `ttl` for external `data_dict`" 

14 

15 def __init__( 

16 self, 

17 data_dict: MutableMapping[str, BytesBlob] | None = None, 

18 *, 

19 ttl: timedelta | None = None, 

20 ) -> None: 

21 super().__init__() 

22 

23 if data_dict is not None and ttl is not None: 

24 raise ValueError(ProxyBlobDict.__EXTERNAL_DICT_TTL_ERROR_MESSAGE) 

25 

26 self.__dict: MutableMapping[str, BytesBlob] = ( 

27 ( 

28 {} if ttl is None 

29 else TTLCache[str, BytesBlob](sys.maxsize, ttl.total_seconds()) 

30 ) if data_dict is None 

31 else data_dict 

32 ) 

33 

34 @override 

35 def __len__(self) -> int: 

36 return len(self.__dict) 

37 

38 @override 

39 def __contains__(self, key: Any) -> bool: 

40 return str(key) in self.__dict 

41 

42 @override 

43 def get[T: Any]( 

44 self, 

45 key: str, 

46 /, 

47 default: BytesBlob | T = None, 

48 ) -> BytesBlob | T: 

49 return self.__dict.get(key, default) 

50 

51 @override 

52 def __getitem__(self, key: str, /) -> BytesBlob: 

53 return self.__dict[key] 

54 

55 @override 

56 def __iter__(self) -> Iterator[str]: 

57 yield from ( 

58 key for key in self.__dict 

59 ) 

60 

61 @override 

62 def clear(self) -> None: 

63 self.__dict.clear() 

64 

65 @override 

66 def pop[T: Any]( 

67 self, 

68 key: str, 

69 /, 

70 default: BytesBlob | T | Literal["__DEFAULT"] = "__DEFAULT", 

71 ) -> BytesBlob | T: 

72 if default == "__DEFAULT": 

73 return self.__dict.pop(key) 

74 

75 return self.__dict.pop(key, default) 

76 

77 @override 

78 def __delitem__(self, key: str, /) -> None: 

79 del self.__dict[key] 

80 

81 @override 

82 def __setitem__(self, key: str, blob: BytesBlob, /) -> None: 

83 self.__dict[key] = blob