Coverage for haystack/core/type

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>

3# SPDX-License-Identifier: Apache-2.0

5import collections.abc

6import inspect

7from collections.abc import Callable

8from enum import Enum

9from types import NoneType, UnionType

10from typing import Any, Union, get_args, get_origin, get_type_hints

12from haystack.dataclasses import ChatMessage

15class ConversionStrategy(Enum):

16 """

17 Strategies for converting values between compatible types in pipeline connections.

18 """

20 CHAT_MESSAGE_TO_STR = "chat_message_to_str"

21 STR_TO_CHAT_MESSAGE = "str_to_chat_message"

22 WRAP = "wrap"

23 WRAP_CHAT_MESSAGE_TO_STR = "wrap_chat_message_to_str"

24 WRAP_STR_TO_CHAT_MESSAGE = "wrap_str_to_chat_message"

25 UNWRAP = "unwrap"

26 UNWRAP_STR_TO_CHAT_MESSAGE = "unwrap_str_to_chat_message"

27 UNWRAP_CHAT_MESSAGE_TO_STR = "unwrap_chat_message_to_str"

30ConversionStrategyType = ConversionStrategy | None

32# Priority used to pick a strategy when a Union receiver admits more than one conversion

33_STRATEGY_PRIORITY = (

34 ConversionStrategy.WRAP,

35 ConversionStrategy.UNWRAP,

36 ConversionStrategy.CHAT_MESSAGE_TO_STR,

37 ConversionStrategy.STR_TO_CHAT_MESSAGE,

38 ConversionStrategy.WRAP_CHAT_MESSAGE_TO_STR,

39 ConversionStrategy.WRAP_STR_TO_CHAT_MESSAGE,

40 ConversionStrategy.UNWRAP_CHAT_MESSAGE_TO_STR,

41 ConversionStrategy.UNWRAP_STR_TO_CHAT_MESSAGE,

42)

45def _resolve_parameter_types(target: Callable) -> dict[str, Any]:

46 """

47 Map the parameter names of a callable to their type annotations, resolving postponed annotations.

49 A callable defined in a module using `from __future__ import annotations` stores its annotations as strings, which

50 never match the types they refer to. Only string annotations are looked up in the resolved type hints: for the

51 others the annotation from the signature is kept.

53 :param target: The callable to inspect.

54 :returns: A dict mapping parameter names to their type annotations. Annotations that cannot be resolved, and

55 parameters without an annotation, are returned as they appear in the signature.

56 """

57 parameters = inspect.signature(target).parameters

58 if any(isinstance(param.annotation, str) for param in parameters.values()):

59 try:

60 hints = get_type_hints(target)

61 except Exception:

62 # TypeError is raised for objects that cannot carry annotations, NameError for names that are not

63 # importable at runtime. Either way we fall back to the unresolved annotations.

64 hints = {}

65 # Non-string annotations are kept as they are written: on Python 3.10 `get_type_hints` widens the annotation

66 # of a parameter defaulting to `None` into an optional. This was changed in Python 3.11, see

67 # https://docs.python.org/3/whatsnew/3.11.html#typing.

68 return {

69 name: hints.get(name, param.annotation) if isinstance(param.annotation, str) else param.annotation

70 for name, param in parameters.items()

71 }

72 return {name: param.annotation for name, param in parameters.items()}

75def _type_name(type_: Any) -> str:

76 """

77 Util methods to get a nice readable representation of a type.

79 Handles Optional and Literal in a special way to make it more readable.

80 """

81 # Literal args are strings, so we wrap them in quotes to make it clear

82 if isinstance(type_, str):

83 return f"'{type_}'"

85 if type_ is NoneType:

86 return "None"

88 args = get_args(type_)

90 if isinstance(type_, UnionType):

91 return " | ".join([_type_name(a) for a in args])

93 name = getattr(type_, "__name__", str(type_))

94 if name.startswith("typing."):

95 name = name[7:]

96 if "[" in name:

97 name = name.split("[")[0]

99 if name == "Union" and NoneType in args and len(args) == 2:

100 # Optional is technically a Union of type and None

101 # but we want to display it as Optional

102 name = "Optional"

103

104 if args:

105 args_str = ", ".join([_type_name(a) for a in args if a is not NoneType])

106 return f"{name}[{args_str}]"

107

108 return f"{name}"

109

110

111def _safe_get_origin(_type: type | UnionType) -> Any:

112 """

113 Safely retrieves the origin type of a generic alias or returns the type itself if it's a built-in.

114

115 This function extends the behavior of `typing.get_origin()` by also handling plain built-in types

116 like `list`, `dict`, etc., which `get_origin()` would normally return `None` for.

117

118 :param _type: A type or generic alias (e.g., `list`, `list[int]`, `dict[str, int]`).

119

120 :returns: The origin type (e.g., `list`, `dict`), or `None` if the input is not a type.

121 """

122 origin = get_origin(_type) or (_type if isinstance(_type, type) else None)

123 # We want to treat typing.Union and UnionType as the same for compatibility checks.

124 # So we convert UnionType to Union if it is detected.

125 if origin is UnionType:

126 origin = Union

127 return origin

128

129

130def _contains_type(container: Any, target: Any) -> bool:

131 """Checks if the container type includes the target type"""

132 if container == target:

133 return True

134 return _safe_get_origin(container) is Union and target in get_args(container)

135

136

137def _strict_types_are_compatible(sender: Any, receiver: Any) -> bool: # noqa: PLR0911

138 """

139 Checks whether the sender type is equal to or a subtype of the receiver type under strict validation.

140

141 Note: this method has no pretense to perform complete type matching.

142 Consider simplifying the typing of your components if you observe unexpected errors during component connection.

143

144 :param sender: The sender type.

145 :param receiver: The receiver type.

146 :return: True if the sender type is strictly compatible with the receiver type, False otherwise.

147 """

148 if sender == receiver or receiver is Any:

149 return True

150

151 if sender is Any:

152 return False

153

154 try:

155 if issubclass(sender, receiver):

156 return True

157 except TypeError: # typing classes can't be used with issubclass, so we deal with them below

158 pass

159

160 sender_origin = _safe_get_origin(sender)

161 receiver_origin = _safe_get_origin(receiver)

162

163 # Special case to reject bare-Union types

164 if (sender_origin is Union and not get_args(sender)) or (receiver_origin is Union and not get_args(receiver)):

165 return False

166

167 if sender_origin is not Union and receiver_origin is Union:

168 return any(_strict_types_are_compatible(sender, union_arg) for union_arg in get_args(receiver))

169

170 # Both must have origins and they must be equal

171 if not (sender_origin and receiver_origin and sender_origin == receiver_origin):

172 return False

173

174 # Compare generic type arguments

175 sender_args = get_args(sender)

176 receiver_args = get_args(receiver)

177

178 # Handle Callable types

179 if sender_origin == receiver_origin == collections.abc.Callable:

180 return _check_callable_compatibility(sender_args, receiver_args)

181

182 # Handle bare types

183 if not sender_args and sender_origin:

184 sender_args = (Any,)

185 if not receiver_args and receiver_origin:

186 receiver_args = (Any,) * (len(sender_args) if sender_args else 1)

187

188 return not (len(sender_args) > len(receiver_args)) and all(

189 _strict_types_are_compatible(*args) for args in zip(sender_args, receiver_args, strict=False)

190 )

191

192

193def _check_callable_compatibility(sender_args: tuple[Any, ...], receiver_args: tuple[Any, ...]) -> bool:

194 """Helper function to check compatibility of Callable types"""

195 if not receiver_args:

196 return True

197 if not sender_args:

198 sender_args = ([Any] * len(receiver_args[0]), Any)

199 # Standard Callable has two elements in args: argument list and return type

200 if len(sender_args) != 2 or len(receiver_args) != 2:

201 return False

202 # Return types must be compatible

203 if not _strict_types_are_compatible(sender_args[1], receiver_args[1]):

204 return False

205 # Input Arguments must be of same length

206 if len(sender_args[0]) != len(receiver_args[0]):

207 return False

208 return all(_strict_types_are_compatible(sender_args[0][i], receiver_args[0][i]) for i in range(len(sender_args[0])))

209

210

211def _get_conversion_strategy(sender: Any, receiver: Any) -> ConversionStrategyType: # noqa: PLR0911

212 """

213 Determines whether a conversion exists from sender to receiver.

214

215 :param sender: The sender type.

216 :param receiver: The receiver type.

217

218 :returns: The ConversionStrategy if conversion is required and supported, otherwise None.

219 """

220 # If sender is a Union, it's only compatible if ALL its types are compatible with the same strategy

221 if _safe_get_origin(sender) is Union:

222 strategies = {_get_conversion_strategy(arg, receiver) for arg in get_args(sender)}

223 if len(strategies) == 1:

224 return strategies.pop()

225 return None

226

227 # If receiver is a Union, it's compatible if ANY of its types are compatible.

228 # When several members admit different conversions, decide based on _STRATEGY_PRIORITY.

229 if _safe_get_origin(receiver) is Union:

230 strategies = {_get_conversion_strategy(sender, arg) for arg in get_args(receiver)}

231 for preferred in _STRATEGY_PRIORITY:

232 if preferred in strategies:

233 return preferred

234 return None

235

236 # ChatMessage -> str

237 if sender is ChatMessage and receiver is str:

238 return ConversionStrategy.CHAT_MESSAGE_TO_STR

239

240 # str -> ChatMessage

241 if sender is str and receiver is ChatMessage:

242 return ConversionStrategy.STR_TO_CHAT_MESSAGE

243

244 # Wrap: T -> List[T]

245 if _safe_get_origin(receiver) is list and (args := get_args(receiver)):

246 inner = args[0]

247 if _strict_types_are_compatible(sender, inner):

248 return ConversionStrategy.WRAP

249 # Wrap + conversion

250 if _contains_type(sender, ChatMessage) and _contains_type(inner, str):

251 return ConversionStrategy.WRAP_CHAT_MESSAGE_TO_STR

252 if _contains_type(sender, str) and _contains_type(inner, ChatMessage):

253 return ConversionStrategy.WRAP_STR_TO_CHAT_MESSAGE

254

255 # Unwrap: list[T] -> T, restricted to str / ChatMessage to avoid silent drop of list[1:].

256 if _safe_get_origin(sender) is list and (args := get_args(sender)):

257 inner = args[0]

258 # Guard against multi-level unwrap (e.g. list[list[str]] -> list[str])

259 if (

260 _safe_get_origin(receiver) is not list

261 and inner in (str, ChatMessage)

262 and _strict_types_are_compatible(inner, receiver)

263 ):

264 return ConversionStrategy.UNWRAP

265 # Unwrap + conversion

266 # Check that all possible types in the sender list can be converted to the receiver type

267 # using the same strategy by recursively calling _get_conversion_strategy on each inner element type.

268 inner_strategy = _get_conversion_strategy(inner, receiver)

269 if inner_strategy == ConversionStrategy.STR_TO_CHAT_MESSAGE:

270 return ConversionStrategy.UNWRAP_STR_TO_CHAT_MESSAGE

271 if inner_strategy == ConversionStrategy.CHAT_MESSAGE_TO_STR:

272 return ConversionStrategy.UNWRAP_CHAT_MESSAGE_TO_STR

273

274 return None

275

276

277def _types_are_compatible(

278 sender: Any, receiver: Any, type_validation: bool = True

279) -> tuple[bool, ConversionStrategyType]:

280 """

281 Determines whether two types are compatible, optionally allowing conversion.

282

283 :param sender: The sender type.

284 :param receiver: The receiver type.

285 :param type_validation: If False, all types are considered compatible.

286

287 :returns: A tuple of (is_compatible, conversion_strategy) where:

288 - is_compatible is True if the types are strictly compatible or can be converted.

289 - conversion_strategy is a ConversionStrategy if conversion is required, otherwise None

290 (including when types are strictly compatible, incompatible, or type validation is disabled).

291 """

292 if not type_validation:

293 return True, None

294

295 if _strict_types_are_compatible(sender, receiver):

296 return True, None

297

298 conversion_strategy = _get_conversion_strategy(sender, receiver)

299 if conversion_strategy:

300 return True, conversion_strategy

301

302 return False, None

303

304

305def _chat_message_to_str(value: Any) -> str:

306 if value.text is None:

307 raise ValueError("Cannot convert `ChatMessage` to `str` because it has no text. ")

308 return value.text

309

310

311def _get_first_item(value: list[Any]) -> Any:

312 """Returns the only element of a one-element list. Raises on empty or multi-element input."""

313 if not value:

314 raise ValueError("Cannot get first item of an empty list. ")

315 if len(value) > 1:

316 raise ValueError(

317 f"Cannot unwrap a list of {len(value)} items to a single value: "

318 "a list-to-scalar connection only accepts one-element lists; otherwise items would be silently dropped."

319 )

320 return value[0]

321

322

323def _convert_value(value: Any, conversion_strategy: ConversionStrategy) -> Any: # noqa: PLR0911

324 """

325 Converts a value using the specified conversion strategy.

326 """

327 match conversion_strategy:

328 case ConversionStrategy.CHAT_MESSAGE_TO_STR:

329 return _chat_message_to_str(value)

330 case ConversionStrategy.STR_TO_CHAT_MESSAGE:

331 return ChatMessage.from_user(value)

332 case ConversionStrategy.WRAP:

333 return [value]

334 case ConversionStrategy.WRAP_CHAT_MESSAGE_TO_STR:

335 return [_chat_message_to_str(value)]

336 case ConversionStrategy.WRAP_STR_TO_CHAT_MESSAGE:

337 return [ChatMessage.from_user(value)]

338 case ConversionStrategy.UNWRAP:

339 return _get_first_item(value)

340 case ConversionStrategy.UNWRAP_STR_TO_CHAT_MESSAGE:

341 return ChatMessage.from_user(_get_first_item(value))

342 case ConversionStrategy.UNWRAP_CHAT_MESSAGE_TO_STR:

343 return _chat_message_to_str(_get_first_item(value))

Coverage for haystack/core/type_utils.py: 96%

165 statements