Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/cattrs/dispatch.py: 66%

1from __future__ import annotations 

2 

3from functools import lru_cache, singledispatch 

4from typing import TYPE_CHECKING, Any, Callable, Generic, Literal, TypeVar 

5 

6from attrs import Factory, define 

7 

8from ._compat import TypeAlias 

9from .fns import Predicate 

10 

11if TYPE_CHECKING: 

12 from .converters import BaseConverter 

13 

14TargetType: TypeAlias = Any 

15UnstructuredValue: TypeAlias = Any 

16StructuredValue: TypeAlias = Any 

17 

18StructureHook: TypeAlias = Callable[[UnstructuredValue, TargetType], StructuredValue] 

19UnstructureHook: TypeAlias = Callable[[StructuredValue], UnstructuredValue] 

20 

21Hook = TypeVar("Hook", StructureHook, UnstructureHook) 

22HookFactory: TypeAlias = Callable[[TargetType], Hook] 

23 

24 

25@define 

26class _DispatchNotFound: 

27 """A dummy object to help signify a dispatch not found.""" 

28 

29 

30@define 

31class FunctionDispatch: 

32 """ 

33 FunctionDispatch is similar to functools.singledispatch, but 

34 instead dispatches based on functions that take the type of the 

35 first argument in the method, and return True or False. 

36 

37 objects that help determine dispatch should be instantiated objects. 

38 

39 :param converter: A converter to be used for factories that require converters. 

40 

41 .. versionchanged:: 24.1.0 

42 Support for factories that require converters, hence this requires a 

43 converter when creating. 

44 """ 

45 

46 _converter: BaseConverter 

47 _handler_pairs: list[tuple[Predicate, Callable[[Any, Any], Any], bool, bool]] = ( 

48 Factory(list) 

49 ) 

50 

51 def register( 

52 self, 

53 predicate: Predicate, 

54 func: Callable[..., Any], 

55 is_generator=False, 

56 takes_converter=False, 

57 ) -> None: 

58 self._handler_pairs.insert(0, (predicate, func, is_generator, takes_converter)) 

59 

60 def dispatch(self, typ: Any) -> Callable[..., Any] | None: 

61 """ 

62 Return the appropriate handler for the object passed. 

63 """ 

64 for can_handle, handler, is_generator, takes_converter in self._handler_pairs: 

65 # can handle could raise an exception here 

66 # such as issubclass being called on an instance. 

67 # it's easier to just ignore that case. 

68 try: 

69 ch = can_handle(typ) 

70 except Exception: # noqa: S112 

71 continue 

72 if ch: 

73 if is_generator: 

74 if takes_converter: 

75 return handler(typ, self._converter) 

76 return handler(typ) 

77 

78 return handler 

79 return None 

80 

81 def get_num_fns(self) -> int: 

82 return len(self._handler_pairs) 

83 

84 def copy_to(self, other: FunctionDispatch, skip: int = 0) -> None: 

85 other._handler_pairs = self._handler_pairs[:-skip] + other._handler_pairs 

86 

87 

88@define(init=False) 

89class MultiStrategyDispatch(Generic[Hook]): 

90 """ 

91 MultiStrategyDispatch uses a combination of exact-match dispatch, 

92 singledispatch, and FunctionDispatch. 

93 

94 :param fallback_factory: A hook factory to be called when a hook cannot be 

95 produced. 

96 :param converter: A converter to be used for factories that require converters. 

97 

98 .. versionchanged:: 23.2.0 

99 Fallbacks are now factories. 

100 .. versionchanged:: 24.1.0 

101 Support for factories that require converters, hence this requires a 

102 converter when creating. 

103 """ 

104 

105 _fallback_factory: HookFactory[Hook] 

106 _direct_dispatch: dict[TargetType, Hook] 

107 _function_dispatch: FunctionDispatch 

108 _single_dispatch: Any 

109 dispatch: Callable[[TargetType, BaseConverter], Hook] 

110 

111 def __init__( 

112 self, fallback_factory: HookFactory[Hook], converter: BaseConverter 

113 ) -> None: 

114 self._fallback_factory = fallback_factory 

115 self._direct_dispatch = {} 

116 self._function_dispatch = FunctionDispatch(converter) 

117 self._single_dispatch = singledispatch(_DispatchNotFound) 

118 self.dispatch = lru_cache(maxsize=None)(self.dispatch_without_caching) 

119 

120 def dispatch_without_caching(self, typ: TargetType) -> Hook: 

121 """Dispatch on the type but without caching the result.""" 

122 try: 

123 dispatch = self._single_dispatch.dispatch(typ) 

124 if dispatch is not _DispatchNotFound: 

125 return dispatch 

126 except Exception: # noqa: S110 

127 pass 

128 

129 direct_dispatch = self._direct_dispatch.get(typ) 

130 if direct_dispatch is not None: 

131 return direct_dispatch 

132 

133 res = self._function_dispatch.dispatch(typ) 

134 return res if res is not None else self._fallback_factory(typ) 

135 

136 def register_cls_list(self, cls_and_handler, direct: bool = False) -> None: 

137 """Register a class to direct or singledispatch.""" 

138 for cls, handler in cls_and_handler: 

139 if direct: 

140 self._direct_dispatch[cls] = handler 

141 else: 

142 self._single_dispatch.register(cls, handler) 

143 self.clear_direct() 

144 self.dispatch.cache_clear() 

145 

146 def register_func_list( 

147 self, 

148 pred_and_handler: list[ 

149 tuple[Predicate, Any] 

150 | tuple[Predicate, Any, bool] 

151 | tuple[Predicate, Callable[[Any, BaseConverter], Any], Literal["extended"]] 

152 ], 

153 ): 

154 """ 

155 Register a predicate function to determine if the handler 

156 should be used for the type. 

157 

158 :param pred_and_handler: The list of predicates and their associated 

159 handlers. If a handler is registered in `extended` mode, it's a 

160 factory that requires a converter. 

161 """ 

162 for tup in pred_and_handler: 

163 if len(tup) == 2: 

164 func, handler = tup 

165 self._function_dispatch.register(func, handler) 

166 else: 

167 func, handler, is_gen = tup 

168 if is_gen == "extended": 

169 self._function_dispatch.register( 

170 func, handler, is_generator=is_gen, takes_converter=True 

171 ) 

172 else: 

173 self._function_dispatch.register(func, handler, is_generator=is_gen) 

174 self.clear_direct() 

175 self.dispatch.cache_clear() 

176 

177 def clear_direct(self) -> None: 

178 """Clear the direct dispatch.""" 

179 self._direct_dispatch.clear() 

180 

181 def clear_cache(self) -> None: 

182 """Clear all caches.""" 

183 self._direct_dispatch.clear() 

184 self.dispatch.cache_clear() 

185 

186 def get_num_fns(self) -> int: 

187 return self._function_dispatch.get_num_fns() 

188 

189 def copy_to(self, other: MultiStrategyDispatch, skip: int = 0) -> None: 

190 self._function_dispatch.copy_to(other._function_dispatch, skip=skip) 

191 for cls, fn in self._single_dispatch.registry.items(): 

192 other._single_dispatch.register(cls, fn) 

193 other.clear_cache()