20241601392

Committed 15 Dec 2025 05:34PM UTC coverage: 92.121% (-0.01%) from 92.133%

Build # 20241601392

Build Type

Pull #10244

github

Committed by

web-flow

Commit Message

Merge 5f2f7fd60 into fd989fecc

Pull Request Pull Request #10244: feat!: drop Python 3.9 support due to EOL

Run Details

14123 of 15331 relevant lines covered (92.12%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.71

haystack/utils/type_serialization.py

# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

import builtins
import importlib
import inspect
import sys
import types
import typing
from threading import Lock
from types import ModuleType
from typing import Any, get_args

from haystack.core.errors import DeserializationError

_import_lock = Lock()


# Sentinel for Python < 3.10
class _Missing:
    pass


_UnionType = getattr(types, "UnionType", _Missing)


def serialize_type(target: Any) -> str:
    """
    Serializes a type or an instance to its string representation, including the module name.

    This function handles types, instances of types, and special typing objects.
    It assumes that non-typing objects will have a '__name__' attribute.

    :param target:
        The object to serialize, can be an instance or a type.
    :return:
        The string representation of the type.
    """
    if target is type(None):
        return "None"

    args = get_args(target)

    if isinstance(target, _UnionType):
        return " | ".join([serialize_type(a) for a in args])

    name = getattr(target, "__name__", str(target))
    if name.startswith("typing."):
        name = name[7:]
    if "[" in name:
        name = name.split("[")[0]

    # Get module name
    module = inspect.getmodule(target)
    module_name = ""
    # We omit the module name for builtins to not clutter the output
    if module and hasattr(module, "__name__") and module.__name__ != "builtins":
        module_name = f"{module.__name__}"

    if args:
        args_str = ", ".join([serialize_type(a) for a in args if a is not type(None)])
        return f"{module_name}.{name}[{args_str}]" if module_name else f"{name}[{args_str}]"

    return f"{module_name}.{name}" if module_name else f"{name}"


def _parse_generic_args(args_str):
    args = []
    bracket_count = 0
    current_arg = ""

    for char in args_str:
        if char == "[":
            bracket_count += 1
        elif char == "]":
            bracket_count -= 1

        if char == "," and bracket_count == 0:
            args.append(current_arg.strip())
            current_arg = ""
        else:
            current_arg += char

    if current_arg:
        args.append(current_arg.strip())

    return args


def deserialize_type(type_str: str) -> Any:  # pylint: disable=too-many-return-statements
    """
    Deserializes a type given its full import path as a string, including nested generic types.

    This function will dynamically import the module if it's not already imported
    and then retrieve the type object from it. It also handles nested generic types like
    `list[dict[int, str]]`.

    :param type_str:
        The string representation of the type's full import path.
    :returns:
        The deserialized type object.
    :raises DeserializationError:
        If the type cannot be deserialized due to missing module or type.
    """

    # Handle generics
    if "[" in type_str and type_str.endswith("]"):
        main_type_str, generics_str = type_str.split("[", 1)
        generics_str = generics_str[:-1]

        main_type = deserialize_type(main_type_str)
        generic_args = [deserialize_type(arg) for arg in _parse_generic_args(generics_str)]

        # Reconstruct
        try:
            return main_type[tuple(generic_args) if len(generic_args) > 1 else generic_args[0]]
        except (TypeError, AttributeError) as e:
            raise DeserializationError(f"Could not apply arguments {generic_args} to type {main_type}") from e

    # Handle non-generic types
    # First, check if there's a module prefix
    if "." in type_str:
        parts = type_str.split(".")
        module_name = ".".join(parts[:-1])
        type_name = parts[-1]

        module = sys.modules.get(module_name)
        if module is None:
            try:
                module = thread_safe_import(module_name)
            except ImportError as e:
                raise DeserializationError(f"Could not import the module: {module_name}") from e

        # Get the class from the module
        if hasattr(module, type_name):
            return getattr(module, type_name)

        raise DeserializationError(f"Could not locate the type: {type_name} in the module: {module_name}")

    # No module prefix, check builtins and typing
    # First check builtins
    if hasattr(builtins, type_str):
        return getattr(builtins, type_str)

    # Then check typing
    if hasattr(typing, type_str):
        return getattr(typing, type_str)

    # Special case for NoneType
    if type_str == "NoneType":
        return type(None)

    # Special case for None
    if type_str == "None":
        return None

    raise DeserializationError(f"Could not deserialize type: {type_str}")


def thread_safe_import(module_name: str) -> ModuleType:
    """
    Import a module in a thread-safe manner.

    Importing modules in a multi-threaded environment can lead to race conditions.
    This function ensures that the module is imported in a thread-safe manner without having impact
    on the performance of the import for single-threaded environments.

    :param module_name: the module to import
    """
    with _import_lock:
        return importlib.import_module(module_name)

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	import builtins	1✔
6	import importlib	1✔
7	import inspect	1✔
8	import sys	1✔
9	import types	1✔
10	import typing	1✔
11	from threading import Lock	1✔
12	from types import ModuleType	1✔
13	from typing import Any, get_args	1✔
14
15	from haystack.core.errors import DeserializationError	1✔
16
17	_import_lock = Lock()	1✔
18
19
20	# Sentinel for Python < 3.10
21	class _Missing:	1✔
22	pass	1✔
23
24
25	_UnionType = getattr(types, "UnionType", _Missing)	1✔
26
27
28	def serialize_type(target: Any) -> str:	1✔
29	"""
30	Serializes a type or an instance to its string representation, including the module name.
31
32	This function handles types, instances of types, and special typing objects.
33	It assumes that non-typing objects will have a '__name__' attribute.
34
35	:param target:
36	The object to serialize, can be an instance or a type.
37	:return:
38	The string representation of the type.
39	"""
40	if target is type(None):	1✔
41	return "None"	1✔
42
43	args = get_args(target)	1✔
44
45	if isinstance(target, _UnionType):	1✔
46	return " \| ".join([serialize_type(a) for a in args])	1✔
47
48	name = getattr(target, "__name__", str(target))	1✔
49	if name.startswith("typing."):	1✔
50	name = name[7:]	×
51	if "[" in name:	1✔
52	name = name.split("[")[0]	×
53
54	# Get module name
55	module = inspect.getmodule(target)	1✔
56	module_name = ""	1✔
57	# We omit the module name for builtins to not clutter the output
58	if module and hasattr(module, "__name__") and module.__name__ != "builtins":	1✔
59	module_name = f"{module.__name__}"	1✔
60
61	if args:	1✔
62	args_str = ", ".join([serialize_type(a) for a in args if a is not type(None)])	1✔
63	return f"{module_name}.{name}[{args_str}]" if module_name else f"{name}[{args_str}]"	1✔
64
65	return f"{module_name}.{name}" if module_name else f"{name}"	1✔
66
67
68	def _parse_generic_args(args_str):	1✔
69	args = []	1✔
70	bracket_count = 0	1✔
71	current_arg = ""	1✔
72
73	for char in args_str:	1✔
74	if char == "[":	1✔
75	bracket_count += 1	1✔
76	elif char == "]":	1✔
77	bracket_count -= 1	1✔
78
79	if char == "," and bracket_count == 0:	1✔
80	args.append(current_arg.strip())	1✔
81	current_arg = ""	1✔
82	else:
83	current_arg += char	1✔
84
85	if current_arg:	1✔
86	args.append(current_arg.strip())	1✔
87
88	return args	1✔
89
90
91	def deserialize_type(type_str: str) -> Any: # pylint: disable=too-many-return-statements	1✔
92	"""
93	Deserializes a type given its full import path as a string, including nested generic types.
94
95	This function will dynamically import the module if it's not already imported
96	and then retrieve the type object from it. It also handles nested generic types like
97	`list[dict[int, str]]`.
98
99	:param type_str:
100	The string representation of the type's full import path.
101	:returns:
102	The deserialized type object.
103	:raises DeserializationError:
104	If the type cannot be deserialized due to missing module or type.
105	"""
106
107	# Handle generics
108	if "[" in type_str and type_str.endswith("]"):	1✔
109	main_type_str, generics_str = type_str.split("[", 1)	1✔
110	generics_str = generics_str[:-1]	1✔
111
112	main_type = deserialize_type(main_type_str)	1✔
113	generic_args = [deserialize_type(arg) for arg in _parse_generic_args(generics_str)]	1✔
114
115	# Reconstruct
116	try:	1✔
117	return main_type[tuple(generic_args) if len(generic_args) > 1 else generic_args[0]]	1✔
118	except (TypeError, AttributeError) as e:	×
119	raise DeserializationError(f"Could not apply arguments {generic_args} to type {main_type}") from e	×
120
121	# Handle non-generic types
122	# First, check if there's a module prefix
123	if "." in type_str:	1✔
124	parts = type_str.split(".")	1✔
125	module_name = ".".join(parts[:-1])	1✔
126	type_name = parts[-1]	1✔
127
128	module = sys.modules.get(module_name)	1✔
129	if module is None:	1✔
130	try:	×
131	module = thread_safe_import(module_name)	×
132	except ImportError as e:	×
133	raise DeserializationError(f"Could not import the module: {module_name}") from e	×
134
135	# Get the class from the module
136	if hasattr(module, type_name):	1✔
137	return getattr(module, type_name)	1✔
138
139	raise DeserializationError(f"Could not locate the type: {type_name} in the module: {module_name}")	×
140
141	# No module prefix, check builtins and typing
142	# First check builtins
143	if hasattr(builtins, type_str):	1✔
144	return getattr(builtins, type_str)	1✔
145
146	# Then check typing
147	if hasattr(typing, type_str):	1✔
148	return getattr(typing, type_str)	1✔
149
150	# Special case for NoneType
151	if type_str == "NoneType":	1✔
152	return type(None)	1✔
153
154	# Special case for None
155	if type_str == "None":	×
156	return None	×
157
158	raise DeserializationError(f"Could not deserialize type: {type_str}")	×
159
160
161	def thread_safe_import(module_name: str) -> ModuleType:	1✔
162	"""
163	Import a module in a thread-safe manner.
164
165	Importing modules in a multi-threaded environment can lead to race conditions.
166	This function ensures that the module is imported in a thread-safe manner without having impact
167	on the performance of the import for single-threaded environments.
168
169	:param module_name: the module to import
170	"""
171	with _import_lock:	1✔
172	return importlib.import_module(module_name)	1✔

deepset-ai / haystack / 20241601392

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous