Coverage for haystack/core/type

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

3# SPDX-License-Identifier: Apache-2.0

5import inspect

6from collections.abc import Callable, Iterable

7from enum import Enum

8from types import NoneType, UnionType

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

11from haystack.dataclasses import ChatMessage

14class ConversionStrategy(Enum):

15 """

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

17 """

19 CHAT_MESSAGE_TO_STR = "chat_message_to_str"

20 STR_TO_CHAT_MESSAGE = "str_to_chat_message"

21 WRAP = "wrap"

22 WRAP_CHAT_MESSAGE_TO_STR = "wrap_chat_message_to_str"

23 WRAP_STR_TO_CHAT_MESSAGE = "wrap_str_to_chat_message"

24 UNWRAP = "unwrap"

25 UNWRAP_STR_TO_CHAT_MESSAGE = "unwrap_str_to_chat_message"

26 UNWRAP_CHAT_MESSAGE_TO_STR = "unwrap_chat_message_to_str"

29ConversionStrategyType = ConversionStrategy | None

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

32_STRATEGY_PRIORITY = (

33 ConversionStrategy.WRAP,

34 ConversionStrategy.UNWRAP,

35 ConversionStrategy.CHAT_MESSAGE_TO_STR,

36 ConversionStrategy.STR_TO_CHAT_MESSAGE,

37 ConversionStrategy.WRAP_CHAT_MESSAGE_TO_STR,

38 ConversionStrategy.WRAP_STR_TO_CHAT_MESSAGE,

39 ConversionStrategy.UNWRAP_CHAT_MESSAGE_TO_STR,

40 ConversionStrategy.UNWRAP_STR_TO_CHAT_MESSAGE,

41)

44def _resolve_parameter_types(target: Callable, *, include_extras: bool = False) -> dict[str, Any]:

45 """

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

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

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

50 others the annotation from the signature is kept.

52 :param target: The callable to inspect.

53 :param include_extras: If `True`, resolved `Annotated` types keep their metadata instead of being unwrapped.

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, include_extras=include_extras)

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 sender_args = get_args(sender)

163 receiver_args = get_args(receiver)

164

165 # Special case to reject bare-Union types

166 if (sender_origin is Union and not sender_args) or (receiver_origin is Union and not receiver_args):

167 return False

168

169 if sender_origin is not Union and receiver_origin is Union:

170 return any(_strict_types_are_compatible(sender, union_arg) for union_arg in receiver_args)

171

172 # Special case to allow list[T] -> Iterable[T] and list[T] -> Iterable[Any]

173 if sender_origin is list and receiver_origin is Iterable:

174 # If the receiver is a bare Iterable, we accept any list.

175 if not receiver_args:

176 return True

177 # If the receiver is Iterable[T], we require the sender to be list[T] for the same T.

178 if len(sender_args) != 1 or len(receiver_args) != 1:

179 return False

180 return _strict_types_are_compatible(sender_args[0], receiver_args[0])

181

182 # Both must have origins and they must be equal

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

184 return False

185

186 # Handle Callable types

187 if sender_origin == receiver_origin == Callable:

188 return _check_callable_compatibility(sender_args, receiver_args)

189

190 # Handle bare types

191 if not sender_args and sender_origin:

192 sender_args = (Any,)

193 if not receiver_args and receiver_origin:

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

195

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

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

198 )

199

200

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

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

203 if not receiver_args:

204 return True

205 if not sender_args:

206 receiver_params = receiver_args[0]

207 # `Callable[..., T]` spells its parameters as Ellipsis, which has no length to expand to

208 sender_args = ([Any] if receiver_params is Ellipsis else [Any] * len(receiver_params), Any)

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

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

211 return False

212 # Return types must be compatible

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

214 return False

215 # An Ellipsis parameter list stands for parameters of any signature, so there are no positions to compare

216 if sender_args[0] is Ellipsis or receiver_args[0] is Ellipsis:

217 return True

218 # Input Arguments must be of same length

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

220 return False

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

222

223

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

225 """

226 Determines whether a conversion exists from sender to receiver.

227

228 :param sender: The sender type.

229 :param receiver: The receiver type.

230

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

232 """

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

234 if _safe_get_origin(sender) is Union:

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

236 if len(strategies) == 1:

237 return strategies.pop()

238 return None

239

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

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

242 if _safe_get_origin(receiver) is Union:

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

244 for preferred in _STRATEGY_PRIORITY:

245 if preferred in strategies:

246 return preferred

247 return None

248

249 # ChatMessage -> str

250 if sender is ChatMessage and receiver is str:

251 return ConversionStrategy.CHAT_MESSAGE_TO_STR

252

253 # str -> ChatMessage

254 if sender is str and receiver is ChatMessage:

255 return ConversionStrategy.STR_TO_CHAT_MESSAGE

256

257 # Wrap: T -> List[T]

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

259 inner = args[0]

260 if _strict_types_are_compatible(sender, inner):

261 return ConversionStrategy.WRAP

262 # Wrap + conversion

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

264 return ConversionStrategy.WRAP_CHAT_MESSAGE_TO_STR

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

266 return ConversionStrategy.WRAP_STR_TO_CHAT_MESSAGE

267

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

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

270 inner = args[0]

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

272 if (

273 _safe_get_origin(receiver) is not list

274 and inner in (str, ChatMessage)

275 and _strict_types_are_compatible(inner, receiver)

276 ):

277 return ConversionStrategy.UNWRAP

278 # Unwrap + conversion

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

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

281 inner_strategy = _get_conversion_strategy(inner, receiver)

282 if inner_strategy == ConversionStrategy.STR_TO_CHAT_MESSAGE:

283 return ConversionStrategy.UNWRAP_STR_TO_CHAT_MESSAGE

284 if inner_strategy == ConversionStrategy.CHAT_MESSAGE_TO_STR:

285 return ConversionStrategy.UNWRAP_CHAT_MESSAGE_TO_STR

286

287 return None

288

289

290def _types_are_compatible(

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

292) -> tuple[bool, ConversionStrategyType]:

293 """

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

295

296 :param sender: The sender type.

297 :param receiver: The receiver type.

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

299

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

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

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

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

304 """

305 if not type_validation:

306 return True, None

307

308 if _strict_types_are_compatible(sender, receiver):

309 return True, None

310

311 conversion_strategy = _get_conversion_strategy(sender, receiver)

312 if conversion_strategy:

313 return True, conversion_strategy

314

315 return False, None

316

317

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

319 if value.text is None:

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

321 return value.text

322

323

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

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

326 if not value:

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

328 if len(value) > 1:

329 raise ValueError(

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

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

332 )

333 return value[0]

334

335

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

337 """

338 Converts a value using the specified conversion strategy.

339 """

340 match conversion_strategy:

341 case ConversionStrategy.CHAT_MESSAGE_TO_STR:

342 return _chat_message_to_str(value)

343 case ConversionStrategy.STR_TO_CHAT_MESSAGE:

344 return ChatMessage.from_user(value)

345 case ConversionStrategy.WRAP:

346 return [value]

347 case ConversionStrategy.WRAP_CHAT_MESSAGE_TO_STR:

348 return [_chat_message_to_str(value)]

349 case ConversionStrategy.WRAP_STR_TO_CHAT_MESSAGE:

350 return [ChatMessage.from_user(value)]

351 case ConversionStrategy.UNWRAP:

352 return _get_first_item(value)

353 case ConversionStrategy.UNWRAP_STR_TO_CHAT_MESSAGE:

354 return ChatMessage.from_user(_get_first_item(value))

355 case ConversionStrategy.UNWRAP_CHAT_MESSAGE_TO_STR:

356 return _chat_message_to_str(_get_first_item(value))

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

173 statements