Coverage for haystack/utils/type_serialization.py: 96%
144 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import ast
6import builtins
7import importlib
8import inspect
9import typing
10from threading import Lock
11from types import GenericAlias, ModuleType, NoneType, UnionType
12from typing import Any, Union, get_args
14from haystack import logging
15from haystack.core.errors import DeserializationError
16from haystack.core.serialization_security import (
17 _check_builtin_is_type,
18 _check_module_allowed,
19 _check_not_deserialization_internal,
20 _check_resolved_module_allowed,
21 _check_traversable_attribute,
22 mark_deserialization_internal,
23)
25logger = logging.getLogger(__name__)
27_import_lock = Lock()
29# Literal singletons that are not types and must be resolved before the generic/module-path handling in
30# deserialize_type: the None singleton ("None"), NoneType, and Ellipsis ("..."). Ellipsis appears in
31# variadic tuples (tuple[int, ...]) and Callable[..., X]. "Ellipsis" is also accepted so that types
32# serialized by older Haystack versions (which emitted the literal "Ellipsis") can still be read back.
33_SPECIAL_LITERALS: dict[str, Any] = {"None": None, "NoneType": NoneType, "...": ..., "Ellipsis": ...}
36def _is_union_type(target: Any) -> bool:
37 """
38 Check if target is a Union type.
40 This handles both `typing.Union[X, Y]` and `X | Y` syntax from PEP 604,
41 including parameterized types like `Optional[str]`.
42 """
43 if target is Union or target is UnionType:
44 return True
45 origin = typing.get_origin(target)
46 return origin is Union or origin is UnionType
49def _build_pep604_union_type(types: list[type | UnionType]) -> type | UnionType:
50 """Build a union type from a list of types using PEP 604 syntax (X | Y)."""
51 result = types[0]
52 for t in types[1:]:
53 result = result | t
54 return result
57def _serialize_type_arg(arg: Any) -> str:
58 """
59 Serialize a single generic argument.
61 Almost all arguments are plain types and are handed off to ``serialize_type``. The exception is the
62 parameter list of a ``Callable[[X, Y], R]``, which ``typing.get_args`` returns as a Python ``list``
63 (e.g. ``[X, Y]``) rather than a type. Such a list is rendered as ``[X, Y]`` so it round-trips back to the
64 same ``Callable`` on deserialization. Without this, the list would fall through to ``serialize_type`` and
65 be dropped (its ``str()`` starts with ``[``, which the name handling truncates to an empty string).
66 """
67 if isinstance(arg, list):
68 return f"[{', '.join(serialize_type(a) for a in arg)}]"
69 return serialize_type(arg)
72def serialize_type(target: Any) -> str:
73 """
74 Serializes a type or an instance to its string representation, including the module name.
76 This function handles types, instances of types, and special typing objects.
77 It assumes that non-typing objects will have a '__name__' attribute.
79 :param target:
80 The object to serialize, can be an instance or a type.
81 :return:
82 The string representation of the type.
83 """
84 if target is NoneType:
85 return "None"
87 # `...` (Ellipsis) shows up as an argument in variadic tuples (tuple[int, ...]) and in
88 # Callable[..., X]. It is a singleton constant rather than a type, so serialize it to the same
89 # literal Python uses when rendering these types. Without this it would fall through to
90 # str(Ellipsis) == "Ellipsis", which deserialize_type then rejects as a non-type builtin.
91 if target is Ellipsis:
92 return "..."
94 # Literal holds values (e.g. Literal["yes", "no"]), not types. Serialize each value with repr() so
95 # strings keep their quotes.
96 if typing.get_origin(target) is typing.Literal:
97 return f"typing.Literal[{', '.join(repr(a) for a in get_args(target))}]"
99 args = get_args(target)
101 if isinstance(target, UnionType):
102 return " | ".join([serialize_type(a) for a in args])
104 name = getattr(target, "__name__", str(target))
105 if name.startswith("typing."):
106 name = name[7:]
107 if "[" in name:
108 name = name.split("[")[0]
110 # Get module name
111 module = inspect.getmodule(target)
112 module_name = ""
113 # We omit the module name for builtins to not clutter the output
114 if module and hasattr(module, "__name__") and module.__name__ != "builtins":
115 module_name = f"{module.__name__}"
117 if args:
118 # For typing generics, convert PEP 604 union types (X | Y) to typing.Union when serializing.
119 # This avoids issues with Python's internal cache, where List[Union[str, int]] and List[str | int] are treated
120 # as the same key. GenericAlias (builtins like list[...]) can keep the PEP 604 syntax.
121 is_typing_generic = not isinstance(target, GenericAlias)
122 # Optional[X] is normalized by Python to Union[X, None]; the trailing None is already implied by the
123 # "Optional" name, so we drop it. For any other generic (e.g. Dict[str, None], Tuple[int, None] or a
124 # Union with more than two members) NoneType is a regular argument and must be kept.
125 skip_nonetype = name == "Optional"
126 args_str = ", ".join(
127 _serialize_type_arg(Union[tuple(get_args(a))] if is_typing_generic and isinstance(a, UnionType) else a) # noqa: UP007
128 for a in args
129 if not (skip_nonetype and a is NoneType)
130 )
131 return f"{module_name}.{name}[{args_str}]" if module_name else f"{name}[{args_str}]"
133 return f"{module_name}.{name}" if module_name else f"{name}"
136def _parse_generic_args(args_str: str) -> list[str]:
137 args = []
138 bracket_count = 0
139 current_arg = ""
141 for char in args_str:
142 if char == "[":
143 bracket_count += 1
144 elif char == "]":
145 bracket_count -= 1
147 if char == "," and bracket_count == 0:
148 args.append(current_arg.strip())
149 current_arg = ""
150 else:
151 current_arg += char
153 if current_arg:
154 args.append(current_arg.strip())
156 return args
159def _parse_pep604_union_args(union_str: str) -> list[str]:
160 """
161 Parse a PEP 604 union string (e.g., "str | int | None") into individual type strings.
163 Handles nested generics properly, e.g., "list[str] | dict[str, int] | None".
165 :param union_str: The union string to parse
166 :returns: A list of individual type strings
167 """
168 args = []
169 bracket_count = 0
170 current_arg = ""
172 for char in union_str:
173 if char == "[":
174 bracket_count += 1
175 elif char == "]":
176 bracket_count -= 1
178 if char == "|" and bracket_count == 0:
179 args.append(current_arg.strip())
180 current_arg = ""
181 else:
182 current_arg += char
184 if current_arg.strip():
185 args.append(current_arg.strip())
187 return args
190def _deserialize_type_arg(arg_str: str) -> Any:
191 """
192 Deserialize a single generic argument produced by ``_parse_generic_args``.
194 Mirrors ``_serialize_type_arg``: a ``Callable`` parameter list is emitted as ``[X, Y]`` and must be read
195 back as a Python ``list`` of types (so ``Callable`` can be re-subscripted as ``Callable[[X, Y], R]``),
196 not as a single type. An empty parameter list ``[]`` becomes ``[]``. Every other argument is a normal
197 type and is delegated to ``deserialize_type``.
198 """
199 stripped = arg_str.strip()
200 if stripped.startswith("[") and stripped.endswith("]"):
201 inner = stripped[1:-1].strip()
202 if not inner:
203 return []
204 return [deserialize_type(a) for a in _parse_generic_args(inner)]
205 return deserialize_type(arg_str)
208@mark_deserialization_internal
209def deserialize_type(type_str: str) -> Any:
210 """
211 Deserializes a type given its full import path as a string, including nested generic types.
213 This function will dynamically import the module if it's not already imported
214 and then retrieve the type object from it. It also handles nested generic types like
215 `list[dict[int, str]]`.
217 Every module path with a `.` prefix is checked against the deserialization
218 allowlist (see `haystack.core.serialization_security`) before being imported. Modules outside
219 the allowlist are rejected with a `DeserializationError`. Builtin and `typing`/`collections`
220 names without a module prefix bypass this check.
222 :param type_str:
223 The string representation of the type's full import path.
224 :returns:
225 The deserialized type object.
226 :raises DeserializationError:
227 If the module is not on the deserialization allowlist, or if the type cannot be
228 deserialized due to a missing module or type.
229 """
230 # Resolve the non-type literal singletons (see _SPECIAL_LITERALS) up front, before the generic and
231 # module-path handling below: the dots in "..." would otherwise be read as a module path, and the
232 # None/Ellipsis singletons would be refused by the builtin type gate.
233 if type_str in _SPECIAL_LITERALS:
234 return _SPECIAL_LITERALS[type_str]
236 # Handle PEP 604 union syntax at the top level (e.g., "str | int", "str | None")
237 pep604_union_args = _parse_pep604_union_args(type_str)
238 if len(pep604_union_args) > 1:
239 deserialized_args = [deserialize_type(arg) for arg in pep604_union_args]
240 return _build_pep604_union_type(deserialized_args)
242 # Handle generics (including Union[X, Y])
243 if "[" in type_str and type_str.endswith("]"):
244 main_type_str, generics_str = type_str.split("[", 1)
245 generics_str = generics_str[:-1]
247 main_type = deserialize_type(main_type_str)
249 # Parse literal args with ast.literal_eval, which safely handles
250 # str/int/bool/None/bytes and is quote-aware, so a comma inside a string value does not split the
251 # arguments.
252 if main_type is typing.Literal:
253 return typing.Literal[ast.literal_eval(f"({generics_str},)")]
255 generic_args = [_deserialize_type_arg(arg) for arg in _parse_generic_args(generics_str)]
257 # Reconstruct
258 try:
259 return main_type[tuple(generic_args) if len(generic_args) > 1 else generic_args[0]]
260 except (TypeError, AttributeError) as e:
261 raise DeserializationError(f"Could not apply arguments {generic_args} to type {main_type}") from e
263 # Handle non-generic types
264 # First, check if there's a module prefix
265 if "." in type_str:
266 try:
267 return _import_class_by_name(type_str)
268 except ImportError as e:
269 raise DeserializationError(str(e)) from e
271 # No module prefix, check builtins and typing.
272 # (None / NoneType / Ellipsis are handled at the top of this function, before they can reach the
273 # builtin type gate below which would refuse them for not being types.)
274 if hasattr(builtins, type_str):
275 resolved = getattr(builtins, type_str)
276 # This bare-name path never consults the allowlist. A type annotation must resolve to an
277 # actual type, so builtin functions like `eval`/`exec` are rejected while types pass.
278 _check_builtin_is_type(resolved, type_str)
279 return resolved
281 # Then check typing
282 if hasattr(typing, type_str):
283 return getattr(typing, type_str)
285 raise DeserializationError(f"Could not deserialize type: {type_str}")
288def thread_safe_import(module_name: str) -> ModuleType:
289 """
290 Import a module in a thread-safe manner.
292 Importing modules in a multi-threaded environment can lead to race conditions.
293 This function ensures that the module is imported in a thread-safe manner without having impact
294 on the performance of the import for single-threaded environments.
296 :param module_name: the module to import
297 """
298 with _import_lock:
299 return importlib.import_module(module_name)
302@mark_deserialization_internal
303def _import_class_by_name(fully_qualified_name: str) -> Any:
304 """
305 Imports an attribute (typically a class) given its fully qualified name.
307 Checks the module against the deserialization allowlist (see
308 `haystack.core.serialization_security`) and, for `builtins`, that the resolved attribute is
309 an actual type — rejecting dangerous builtins like `eval`/`compile`.
311 :param fully_qualified_name: the fully qualified name, e.g. "my_package.MyClass"
312 :returns: the imported attribute.
313 :raises ImportError: If the module cannot be imported or the attribute doesn't exist on it.
314 :raises DeserializationError: If the module is not on the deserialization allowlist, or the
315 resolved builtin is not a type.
316 """
317 module_path, attr_name = fully_qualified_name.rsplit(".", 1)
318 _check_module_allowed(module_path)
319 # A class reference names a public attribute of a module, never an object-internals attribute
320 # (`__dict__`, `__globals__`, ...) that would expose a module namespace or escape gadget.
321 _check_traversable_attribute(attr_name, fully_qualified_name)
322 try:
323 logger.debug(
324 "Attempting to import '{attr_name}' from module '{module_path}'",
325 attr_name=attr_name,
326 module_path=module_path,
327 )
328 module = thread_safe_import(module_path)
329 resolved = getattr(module, attr_name)
330 # `_check_module_allowed` above gates the declared module path, which the caller controls.
331 # A class/module re-exported as an attribute of an allowlisted module (e.g.
332 # `haystack.utils.auth.os` -> the `os` module, or a re-exported `subprocess.Popen`) would
333 # otherwise slip through. Re-check the module the object actually belongs to; `module_path`
334 # is the allowlisted module it was resolved from, so a private C accelerator backing it
335 # (e.g. `io.StringIO` -> `_io`) is still accepted.
336 _check_resolved_module_allowed(resolved, declared_module=module_path)
337 # Refuse the deserializer's own machinery (the allowlist-administration function and the
338 # resolution helpers) on the class-resolution path too, so it cannot be reached as a
339 # component `type` or nested class reference. See `mark_deserialization_internal`.
340 _check_not_deserialization_internal(resolved, fully_qualified_name)
341 if module_path == "builtins":
342 _check_builtin_is_type(resolved, fully_qualified_name)
343 return resolved
344 except (ImportError, AttributeError) as error:
345 logger.exception("Failed to import '{full_name}'", full_name=fully_qualified_name)
346 raise ImportError(f"Could not import '{fully_qualified_name}'") from error