15191527043

Committed 22 May 2025 04:08PM UTC coverage: 90.345% (-0.07%) from 90.411%

Build # 15191527043

Build Type

Pull #9426

github

Committed by

web-flow

Commit Message

Merge 212e60881 into 4a5e4d3e6

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

Run Details

11173 of 12367 relevant lines covered (90.35%)

0.9 hits per line

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

86.61

haystack/core/super_component/utils.py

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

from typing import Annotated, Any, Dict, List, Optional, Set, Tuple, TypeVar, Union, cast, get_args, get_origin

from haystack.core.component.types import HAYSTACK_GREEDY_VARIADIC_ANNOTATION, HAYSTACK_VARIADIC_ANNOTATION


class _delegate_default:
    """Custom object for delegating filling of default values to the underlying components."""


T = TypeVar("T")


def _is_compatible(type1: T, type2: T, unwrap_nested: bool = True) -> Tuple[bool, Optional[T]]:
    """
    Check if two types are compatible (bidirectional/symmetric check).

    :param type1: First type to compare
    :param type2: Second type to compare
    :param unwrap_nested: If True, recursively unwraps nested Optional and Variadic types.
        If False, only unwraps at the top level.
    :return: Tuple of (True if types are compatible, common type if compatible)
    """
    type1_unwrapped = _unwrap_all(type1, recursive=unwrap_nested)
    type2_unwrapped = _unwrap_all(type2, recursive=unwrap_nested)

    return _types_are_compatible(type1_unwrapped, type2_unwrapped)


def _types_are_compatible(type1: T, type2: T) -> Tuple[bool, Optional[T]]:
    """
    Core type compatibility check implementing symmetric matching.

    :param type1: First unwrapped type to compare
    :param type2: Second unwrapped type to compare
    :return: True if types are compatible, False otherwise
    """
    # Handle Any type
    if type1 is Any:
        return True, _convert_to_typing_type(type2)
    if type2 is Any:
        return True, _convert_to_typing_type(type1)

    # Direct equality
    if type1 == type2:
        return True, _convert_to_typing_type(type1)

    type1_origin = get_origin(type1)
    type2_origin = get_origin(type2)

    # Handle Union types
    if type1_origin is Union or type2_origin is Union:
        return _check_union_compatibility(type1, type2, type1_origin, type2_origin)

    # Handle non-Union types
    return _check_non_union_compatibility(type1, type2, type1_origin, type2_origin)


def _check_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> Tuple[bool, Optional[T]]:
    """Handle all Union type compatibility cases."""
    if type1_origin is Union and type2_origin is not Union:
        # Find all compatible types from the union
        compatible_types = []
        for union_arg in get_args(type1):
            is_compat, common = _types_are_compatible(union_arg, type2)
            if is_compat and common is not None:
                compatible_types.append(common)
        if compatible_types:
            # The constructed Union or single type must be cast to Optional[T]
            # to satisfy mypy, as T is specific to this function's call context.
            result_type = Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
            return True, cast(Optional[T], result_type)
        return False, None

    if type2_origin is Union and type1_origin is not Union:
        # Find all compatible types from the union
        compatible_types = []
        for union_arg in get_args(type2):
            is_compat, common = _types_are_compatible(type1, union_arg)
            if is_compat and common is not None:
                compatible_types.append(common)
        if compatible_types:
            # The constructed Union or single type must be cast to Optional[T]
            # to satisfy mypy, as T is specific to this function's call context.
            result_type = Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
            return True, cast(Optional[T], result_type)
        return False, None

    # Both are Union types
    compatible_types = []
    for arg1 in get_args(type1):
        for arg2 in get_args(type2):
            is_compat, common = _types_are_compatible(arg1, arg2)
            if is_compat and common is not None:
                compatible_types.append(common)

    if compatible_types:
        # The constructed Union or single type must be cast to Optional[T]
        # to satisfy mypy, as T is specific to this function's call context.
        result_type = Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
        return True, cast(Optional[T], result_type)
    return False, None


def _check_non_union_compatibility(
    type1: T, type2: T, type1_origin: Any, type2_origin: Any
) -> Tuple[bool, Optional[T]]:
    """Handle non-Union type compatibility cases."""
    # If no origin, compare types directly
    if not type1_origin and not type2_origin:
        if type1 == type2:
            return True, type1
        return False, None

    # Both must have origins and they must be equal
    if not (type1_origin and type2_origin and type1_origin == type2_origin):
        return False, None

    # Compare generic type arguments
    type1_args = get_args(type1)
    type2_args = get_args(type2)

    if len(type1_args) != len(type2_args):
        return False, None

    # Check if all arguments are compatible
    common_args = []
    for t1_arg, t2_arg in zip(type1_args, type2_args):
        is_compat, common = _types_are_compatible(t1_arg, t2_arg)
        if not is_compat:
            return False, None
        common_args.append(common)

    # Reconstruct the type with common arguments
    typing_type = _convert_to_typing_type(type1_origin)
    return True, cast(Optional[T], typing_type[tuple(common_args)])


def _unwrap_all(t: T, recursive: bool) -> T:
    """
    Unwrap a type until no more unwrapping is possible.

    :param t: Type to unwrap
    :param recursive: If True, recursively unwraps nested types
    :return: The fully unwrapped type
    """
    # First handle top-level Variadic/GreedyVariadic
    if _is_variadic_type(t):
        t = _unwrap_variadics(t, recursive=recursive)
    else:
        # If it's a generic type and we're unwrapping recursively
        origin = get_origin(t)
        if recursive and origin is not None and (args := get_args(t)):
            unwrapped_args = tuple(_unwrap_all(arg, recursive) for arg in args)
            t = origin[unwrapped_args]

    # Then handle top-level Optional
    if _is_optional_type(t):
        t = _unwrap_optionals(t, recursive=recursive)

    return t


def _is_variadic_type(t: T) -> bool:
    """Check if type is a Variadic or GreedyVariadic type."""
    origin = get_origin(t)
    if origin is Annotated:
        args = get_args(t)
        return len(args) >= 2 and args[1] in (HAYSTACK_VARIADIC_ANNOTATION, HAYSTACK_GREEDY_VARIADIC_ANNOTATION)  # noqa: PLR2004
    return False


def _is_optional_type(t: T) -> bool:
    """Check if type is an Optional type."""
    origin = get_origin(t)
    if origin is Union:
        args = get_args(t)
        return type(None) in args
    return False


def _unwrap_variadics(t: T, recursive: bool) -> T:
    """
    Unwrap Variadic or GreedyVariadic annotated types.

    :param t: Type to unwrap
    :param recursive: If True, recursively unwraps nested types
    :return: Unwrapped type if it was a variadic type, original type otherwise
    """
    if not _is_variadic_type(t):
        return t

    args = get_args(t)
    # Get the Iterable[X] type and extract X
    iterable_type = args[0]
    inner_type = get_args(iterable_type)[0]

    # Only recursively unwrap if requested
    if recursive:
        return _unwrap_all(inner_type, recursive)
    return inner_type


def _unwrap_optionals(t: T, recursive: bool) -> T:
    """
    Unwrap Optional[...] types (Union[X, None]).

    :param t: Type to unwrap
    :param recursive: If True, recursively unwraps nested types
    :return: Unwrapped type if it was an Optional, original type otherwise
    """
    if not _is_optional_type(t):
        return t

    args = list(get_args(t))
    args.remove(type(None))
    result = args[0] if len(args) == 1 else Union[tuple(args)]  # type: ignore

    # Only recursively unwrap if requested
    if recursive:
        return _unwrap_all(result, recursive)  # type: ignore
    return result  # type: ignore


def _convert_to_typing_type(t: Any) -> Any:
    """
    Convert built-in Python types to their typing equivalents.

    :param t: Type to convert
    :return: The type using typing module types
    """
    origin = get_origin(t)
    args = get_args(t)

    # Mapping of built-in types to their typing equivalents
    type_converters = {
        list: lambda: List if not args else List[Any],
        dict: lambda: Dict if not args else Dict[Any, Any],
        set: lambda: Set if not args else Set[Any],
        tuple: lambda: Tuple if not args else Tuple[Any, ...],
    }

    # Recursive argument handling
    if origin in type_converters:
        result = type_converters[origin]()
        if args:
            if origin == list:
                return List[_convert_to_typing_type(args[0])]  # type: ignore
            if origin == dict:
                return Dict[_convert_to_typing_type(args[0]), _convert_to_typing_type(args[1])]  # type: ignore
            if origin == set:
                return Set[_convert_to_typing_type(args[0])]  # type: ignore
            if origin == tuple:
                return Tuple[tuple(_convert_to_typing_type(arg) for arg in args)]
        return result
    return t

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	from typing import Annotated, Any, Dict, List, Optional, Set, Tuple, TypeVar, Union, cast, get_args, get_origin	1✔
6
7	from haystack.core.component.types import HAYSTACK_GREEDY_VARIADIC_ANNOTATION, HAYSTACK_VARIADIC_ANNOTATION	1✔
8
9
10	class _delegate_default:	1✔
11	"""Custom object for delegating filling of default values to the underlying components."""
12
13
14	T = TypeVar("T")	1✔
15
16
17	def _is_compatible(type1: T, type2: T, unwrap_nested: bool = True) -> Tuple[bool, Optional[T]]:	1✔
18	"""
19	Check if two types are compatible (bidirectional/symmetric check).
20
21	:param type1: First type to compare
22	:param type2: Second type to compare
23	:param unwrap_nested: If True, recursively unwraps nested Optional and Variadic types.
24	If False, only unwraps at the top level.
25	:return: Tuple of (True if types are compatible, common type if compatible)
26	"""
27	type1_unwrapped = _unwrap_all(type1, recursive=unwrap_nested)	1✔
28	type2_unwrapped = _unwrap_all(type2, recursive=unwrap_nested)	1✔
29
30	return _types_are_compatible(type1_unwrapped, type2_unwrapped)	1✔
31
32
33	def _types_are_compatible(type1: T, type2: T) -> Tuple[bool, Optional[T]]:	1✔
34	"""
35	Core type compatibility check implementing symmetric matching.
36
37	:param type1: First unwrapped type to compare
38	:param type2: Second unwrapped type to compare
39	:return: True if types are compatible, False otherwise
40	"""
41	# Handle Any type
42	if type1 is Any:	1✔
43	return True, _convert_to_typing_type(type2)	1✔
44	if type2 is Any:	1✔
45	return True, _convert_to_typing_type(type1)	1✔
46
47	# Direct equality
48	if type1 == type2:	1✔
49	return True, _convert_to_typing_type(type1)	1✔
50
51	type1_origin = get_origin(type1)	1✔
52	type2_origin = get_origin(type2)	1✔
53
54	# Handle Union types
55	if type1_origin is Union or type2_origin is Union:	1✔
56	return _check_union_compatibility(type1, type2, type1_origin, type2_origin)	1✔
57
58	# Handle non-Union types
59	return _check_non_union_compatibility(type1, type2, type1_origin, type2_origin)	1✔
60
61
62	def _check_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> Tuple[bool, Optional[T]]:	1✔
63	"""Handle all Union type compatibility cases."""
64	if type1_origin is Union and type2_origin is not Union:	1✔
65	# Find all compatible types from the union
66	compatible_types = []	1✔
67	for union_arg in get_args(type1):	1✔
68	is_compat, common = _types_are_compatible(union_arg, type2)	1✔
69	if is_compat and common is not None:	1✔
70	compatible_types.append(common)	1✔
71	if compatible_types:	1✔
72	# The constructed Union or single type must be cast to Optional[T]
73	# to satisfy mypy, as T is specific to this function's call context.
74	result_type = Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]	1✔
75	return True, cast(Optional[T], result_type)	1✔
76	return False, None	×
77
78	if type2_origin is Union and type1_origin is not Union:	1✔
79	# Find all compatible types from the union
80	compatible_types = []	1✔
81	for union_arg in get_args(type2):	1✔
82	is_compat, common = _types_are_compatible(type1, union_arg)	1✔
83	if is_compat and common is not None:	1✔
84	compatible_types.append(common)	1✔
85	if compatible_types:	1✔
86	# The constructed Union or single type must be cast to Optional[T]
87	# to satisfy mypy, as T is specific to this function's call context.
88	result_type = Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]	1✔
89	return True, cast(Optional[T], result_type)	1✔
90	return False, None	1✔
91
92	# Both are Union types
93	compatible_types = []	1✔
94	for arg1 in get_args(type1):	1✔
95	for arg2 in get_args(type2):	1✔
96	is_compat, common = _types_are_compatible(arg1, arg2)	1✔
97	if is_compat and common is not None:	1✔
98	compatible_types.append(common)	1✔
99
100	if compatible_types:	1✔
101	# The constructed Union or single type must be cast to Optional[T]
102	# to satisfy mypy, as T is specific to this function's call context.
103	result_type = Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]	1✔
104	return True, cast(Optional[T], result_type)	1✔
105	return False, None	×
106
107
108	def _check_non_union_compatibility(	1✔
109	type1: T, type2: T, type1_origin: Any, type2_origin: Any
110	) -> Tuple[bool, Optional[T]]:
111	"""Handle non-Union type compatibility cases."""
112	# If no origin, compare types directly
113	if not type1_origin and not type2_origin:	1✔
114	if type1 == type2:	1✔
115	return True, type1	×
116	return False, None	1✔
117
118	# Both must have origins and they must be equal
119	if not (type1_origin and type2_origin and type1_origin == type2_origin):	1✔
120	return False, None	1✔
121
122	# Compare generic type arguments
123	type1_args = get_args(type1)	1✔
124	type2_args = get_args(type2)	1✔
125
126	if len(type1_args) != len(type2_args):	1✔
127	return False, None	×
128
129	# Check if all arguments are compatible
130	common_args = []	1✔
131	for t1_arg, t2_arg in zip(type1_args, type2_args):	1✔
132	is_compat, common = _types_are_compatible(t1_arg, t2_arg)	1✔
133	if not is_compat:	1✔
134	return False, None	1✔
135	common_args.append(common)	×
136
137	# Reconstruct the type with common arguments
138	typing_type = _convert_to_typing_type(type1_origin)	×
139	return True, cast(Optional[T], typing_type[tuple(common_args)])	×
140
141
142	def _unwrap_all(t: T, recursive: bool) -> T:	1✔
143	"""
144	Unwrap a type until no more unwrapping is possible.
145
146	:param t: Type to unwrap
147	:param recursive: If True, recursively unwraps nested types
148	:return: The fully unwrapped type
149	"""
150	# First handle top-level Variadic/GreedyVariadic
151	if _is_variadic_type(t):	1✔
152	t = _unwrap_variadics(t, recursive=recursive)	1✔
153	else:
154	# If it's a generic type and we're unwrapping recursively
155	origin = get_origin(t)	1✔
156	if recursive and origin is not None and (args := get_args(t)):	1✔
157	unwrapped_args = tuple(_unwrap_all(arg, recursive) for arg in args)	1✔
158	t = origin[unwrapped_args]	1✔
159
160	# Then handle top-level Optional
161	if _is_optional_type(t):	1✔
162	t = _unwrap_optionals(t, recursive=recursive)	1✔
163
164	return t	1✔
165
166
167	def _is_variadic_type(t: T) -> bool:	1✔
168	"""Check if type is a Variadic or GreedyVariadic type."""
169	origin = get_origin(t)	1✔
170	if origin is Annotated:	1✔
171	args = get_args(t)	1✔
172	return len(args) >= 2 and args[1] in (HAYSTACK_VARIADIC_ANNOTATION, HAYSTACK_GREEDY_VARIADIC_ANNOTATION) # noqa: PLR2004	1✔
173	return False	1✔
174
175
176	def _is_optional_type(t: T) -> bool:	1✔
177	"""Check if type is an Optional type."""
178	origin = get_origin(t)	1✔
179	if origin is Union:	1✔
180	args = get_args(t)	1✔
181	return type(None) in args	1✔
182	return False	1✔
183
184
185	def _unwrap_variadics(t: T, recursive: bool) -> T:	1✔
186	"""
187	Unwrap Variadic or GreedyVariadic annotated types.
188
189	:param t: Type to unwrap
190	:param recursive: If True, recursively unwraps nested types
191	:return: Unwrapped type if it was a variadic type, original type otherwise
192	"""
193	if not _is_variadic_type(t):	1✔
194	return t	×
195
196	args = get_args(t)	1✔
197	# Get the Iterable[X] type and extract X
198	iterable_type = args[0]	1✔
199	inner_type = get_args(iterable_type)[0]	1✔
200
201	# Only recursively unwrap if requested
202	if recursive:	1✔
203	return _unwrap_all(inner_type, recursive)	1✔
204	return inner_type	1✔
205
206
207	def _unwrap_optionals(t: T, recursive: bool) -> T:	1✔
208	"""
209	Unwrap Optional[...] types (Union[X, None]).
210
211	:param t: Type to unwrap
212	:param recursive: If True, recursively unwraps nested types
213	:return: Unwrapped type if it was an Optional, original type otherwise
214	"""
215	if not _is_optional_type(t):	1✔
216	return t	×
217
218	args = list(get_args(t))	1✔
219	args.remove(type(None))	1✔
220	result = args[0] if len(args) == 1 else Union[tuple(args)] # type: ignore	1✔
221
222	# Only recursively unwrap if requested
223	if recursive:	1✔
224	return _unwrap_all(result, recursive) # type: ignore	1✔
225	return result # type: ignore	×
226
227
228	def _convert_to_typing_type(t: Any) -> Any:	1✔
229	"""
230	Convert built-in Python types to their typing equivalents.
231
232	:param t: Type to convert
233	:return: The type using typing module types
234	"""
235	origin = get_origin(t)	1✔
236	args = get_args(t)	1✔
237
238	# Mapping of built-in types to their typing equivalents
239	type_converters = {	1✔
240	list: lambda: List if not args else List[Any],
241	dict: lambda: Dict if not args else Dict[Any, Any],
242	set: lambda: Set if not args else Set[Any],
243	tuple: lambda: Tuple if not args else Tuple[Any, ...],
244	}
245
246	# Recursive argument handling
247	if origin in type_converters:	1✔
248	result = type_converters[origin]()	1✔
249	if args:	1✔
250	if origin == list:	1✔
251	return List[_convert_to_typing_type(args[0])] # type: ignore	1✔
252	if origin == dict:	×
253	return Dict[_convert_to_typing_type(args[0]), _convert_to_typing_type(args[1])] # type: ignore	×
254	if origin == set:	×
255	return Set[_convert_to_typing_type(args[0])] # type: ignore	×
256	if origin == tuple:	×
257	return Tuple[tuple(_convert_to_typing_type(arg) for arg in args)]	×
258	return result	×
259	return t	1✔

deepset-ai / haystack / 15191527043

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