15254528833

Committed 26 May 2025 12:56PM UTC coverage: 90.146% (-0.3%) from 90.411%

Build # 15254528833

Build Type

Pull #9426

github

Committed by

web-flow

Commit Message

Merge 06c2b66b1 into 802328e29

Pull Request Pull Request #9426: feat: add component name and type to `StreamingChunk`

Run Details

11398 of 12644 relevant lines covered (90.15%)

0.9 hits per line

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

86.84

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 typing
from threading import Lock
from types import ModuleType
from typing import Any, get_args

from haystack.core.errors import DeserializationError

_import_lock = Lock()


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.
    """
    name = getattr(target, "__name__", str(target))

    # Remove the 'typing.' prefix when using python <3.9
    if name.startswith("typing."):
        name = name[7:]
    # Remove the arguments from the name when using python <3.9
    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__}"

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

deepset-ai / haystack / 15254528833

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