Coverage for src/extratools_core/crudl.py: 100%

1from collections.abc import Callable, Iterable, Iterator, Mapping, MutableMapping 

2from typing import Any, cast 

3 

4from .typing import SearchableMapping 

5 

6 

7class RLWrapper[KT: Any, VT: Any]: 

8 def __init__( 

9 self, 

10 mapping: Mapping[KT, VT], 

11 *, 

12 values_in_list: bool = False, 

13 ) -> None: 

14 self.mapping = mapping 

15 

16 self.__values_in_list = values_in_list 

17 

18 def read(self, key: KT) -> VT: 

19 return self.mapping[key] 

20 

21 def list( 

22 self, 

23 filter_func: Callable[[KT], bool] | None = None, 

24 ) -> Iterable[tuple[KT, VT | None]]: 

25 if filter_func is None: 

26 if self.__values_in_list: 

27 yield from self.mapping.items() 

28 else: 

29 for key in self.mapping: 

30 yield key, None 

31 else: 

32 for key in filter(filter_func, self.mapping): 

33 yield key, self.mapping[key] if self.__values_in_list else None 

34 

35 

36class CRUDLWrapper[KT: Any, VT: Any](RLWrapper[KT, VT]): 

37 def __init__( 

38 self, 

39 mapping: MutableMapping[KT, VT], 

40 *, 

41 values_in_list: bool = False, 

42 ) -> None: 

43 super().__init__( 

44 mapping, 

45 values_in_list=values_in_list, 

46 ) 

47 

48 self.mapping = mapping 

49 

50 def create(self, key: KT, value: VT) -> VT: 

51 if key in self.mapping: 

52 raise KeyError 

53 

54 self.mapping[key] = value 

55 return value 

56 

57 def update(self, key: KT, value: VT) -> VT: 

58 if key not in self.mapping: 

59 raise KeyError 

60 

61 self.mapping[key] = value 

62 return value 

63 

64 def delete(self, key: KT) -> VT: 

65 default = object() 

66 value = self.mapping.pop(key, default) 

67 if value == default: 

68 raise KeyError 

69 

70 return cast("VT", value) 

71 

72 

73class RLDict[KT: Any, VT: Any](SearchableMapping[KT, VT]): 

74 def __init__( 

75 self, 

76 *, 

77 read_func: Callable[[KT], VT], 

78 list_func: Callable[[Any | None], Iterable[tuple[KT, VT | None]]], 

79 ) -> None: 

80 self.__read_func = read_func 

81 self.__list_func = list_func 

82 

83 def __getitem__(self, key: KT) -> VT: 

84 return self.__read_func(key) 

85 

86 def __iter__(self) -> Iterator[KT]: 

87 return self.search() 

88 

89 def search(self, filter_body: Any = None) -> Iterator[KT]: 

90 for key, _ in self.__list_func(filter_body): 

91 yield key 

92 

93 def __len__(self) -> int: 

94 # Cannot use `count` in `toolz` as itself depends on this function 

95 count = 0 

96 for _ in self: 

97 count += 1 

98 return count 

99 

100 

101class CRUDLDict[KT: Any, VT: Any](MutableMapping[KT, VT], RLDict[KT, VT]): 

102 def __init__( 

103 self, 

104 *, 

105 create_func: Callable[[KT | None, Any], VT | None], 

106 read_func: Callable[[KT], VT], 

107 update_func: Callable[[KT, Any], VT | None], 

108 delete_func: Callable[[KT], VT | None], 

109 list_func: Callable[[Any | None], Iterable[tuple[KT, VT | None]]], 

110 ) -> None: 

111 RLDict.__init__( 

112 self, 

113 read_func=read_func, 

114 list_func=list_func, 

115 ) 

116 

117 self.__create_func = create_func 

118 self.__read_func = read_func 

119 self.__update_func = update_func 

120 self.__delete_func = delete_func 

121 self.__list_func = list_func 

122 

123 def __delitem__(self, key: KT) -> None: 

124 self.__delete_func(key) 

125 

126 def __setitem__(self, key: KT | None, value: VT) -> None: 

127 if key is None or key not in self: 

128 self.__create_func(key, value) 

129 else: 

130 self.__update_func(key, value)