16493136922

Committed 24 Jul 2025 09:29AM UTC coverage: 90.796% (-0.02%) from 90.813%

Build # 16493136922

Build Type

Pull #9650

github

Committed by

web-flow

Commit Message

Merge b8da7dd3b into d059cf2c2

Pull Request Pull Request #9650: feat: Add support for the union operator `|` added in python 3.10

Run Details

12864 of 14168 relevant lines covered (90.8%)

0.91 hits per line

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

92.86

haystack/core/type_utils.py

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

import collections.abc
import types
from typing import Any, Type, TypeVar, Union, get_args, get_origin

from haystack.utils.type_serialization import _UnionType

T = TypeVar("T")


def _types_are_compatible(sender: Type, receiver: Type, type_validation: bool = True) -> bool:
    """
    Determines if two types are compatible based on the specified validation mode.

    :param sender: The sender type.
    :param receiver: The receiver type.
    :param type_validation: Whether to perform strict type validation.
    :return: True if the types are compatible, False otherwise.
    """
    if type_validation:
        return _strict_types_are_compatible(sender, receiver)
    else:
        return True


def _safe_get_origin(_type: Type[T]) -> Union[Type[T], None]:
    """
    Safely retrieves the origin type of a generic alias or returns the type itself if it's a built-in.

    This function extends the behavior of `typing.get_origin()` by also handling plain built-in types
    like `list`, `dict`, etc., which `get_origin()` would normally return `None` for.

    :param _type: A type or generic alias (e.g., `list`, `list[int]`, `dict[str, int]`).

    :returns: The origin type (e.g., `list`, `dict`), or `None` if the input is not a type.
    """
    origin = get_origin(_type) or (_type if isinstance(_type, type) else None)
    # We want to treat typing.Union and UnionType as the same for compatibility checks.
    # So we convert UnionType to Union if it is detected.
    if origin is _UnionType:
        origin = Union
    return origin


def _strict_types_are_compatible(sender, receiver):  # pylint: disable=too-many-return-statements
    """
    Checks whether the sender type is equal to or a subtype of the receiver type under strict validation.

    Note: this method has no pretense to perform proper type matching. It especially does not deal with aliasing of
    typing classes such as `List` or `Dict` to their runtime counterparts `list` and `dict`. It also does not deal well
    with "bare" types, so `List` is treated differently from `List[Any]`, even though they should be the same.
    Consider simplifying the typing of your components if you observe unexpected errors during component connection.

    :param sender: The sender type.
    :param receiver: The receiver type.
    :return: True if the sender type is strictly compatible with the receiver type, False otherwise.
    """
    if sender == receiver or receiver is Any:
        return True

    if sender is Any:
        return False

    try:
        if issubclass(sender, receiver):
            return True
    except TypeError:  # typing classes can't be used with issubclass, so we deal with them below
        pass

    sender_origin = _safe_get_origin(sender)
    receiver_origin = _safe_get_origin(receiver)

    if sender_origin is not Union and receiver_origin is Union:
        return any(_strict_types_are_compatible(sender, union_arg) for union_arg in get_args(receiver))

    # Both must have origins and they must be equal
    if not (sender_origin and receiver_origin and sender_origin == receiver_origin):
        return False

    # Compare generic type arguments
    sender_args = get_args(sender)
    receiver_args = get_args(receiver)

    # Handle Callable types
    if sender_origin == receiver_origin == collections.abc.Callable:
        return _check_callable_compatibility(sender_args, receiver_args)

    # Handle bare types
    if not sender_args and sender_origin:
        sender_args = (Any,)
    if not receiver_args and receiver_origin:
        receiver_args = (Any,) * (len(sender_args) if sender_args else 1)

    return not (len(sender_args) > len(receiver_args)) and all(
        _strict_types_are_compatible(*args) for args in zip(sender_args, receiver_args)
    )


def _check_callable_compatibility(sender_args, receiver_args):
    """Helper function to check compatibility of Callable types"""
    if not receiver_args:
        return True
    if not sender_args:
        sender_args = ([Any] * len(receiver_args[0]), Any)
    # Standard Callable has two elements in args: argument list and return type
    if len(sender_args) != 2 or len(receiver_args) != 2:
        return False
    # Return types must be compatible
    if not _strict_types_are_compatible(sender_args[1], receiver_args[1]):
        return False
    # Input Arguments must be of same length
    if len(sender_args[0]) != len(receiver_args[0]):
        return False
    return all(_strict_types_are_compatible(sender_args[0][i], receiver_args[0][i]) for i in range(len(sender_args[0])))


def _type_name(type_: Any) -> str:
    """
    Util methods to get a nice readable representation of a type.

    Handles Optional and Literal in a special way to make it more readable.
    """
    # Literal args are strings, so we wrap them in quotes to make it clear
    if isinstance(type_, str):
        return f"'{type_}'"

    if type_ is type(None):
        return "None"

    args = get_args(type_)

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

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

    if name == "Union" and type(None) in args and len(args) == 2:
        # Optional is technically a Union of type and None
        # but we want to display it as Optional
        name = "Optional"

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

    return f"{name}"

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	import collections.abc	1✔
6	import types	1✔
7	from typing import Any, Type, TypeVar, Union, get_args, get_origin	1✔
8
9	from haystack.utils.type_serialization import _UnionType	1✔
10
11	T = TypeVar("T")	1✔
12
13
14	def _types_are_compatible(sender: Type, receiver: Type, type_validation: bool = True) -> bool:	1✔
15	"""
16	Determines if two types are compatible based on the specified validation mode.
17
18	:param sender: The sender type.
19	:param receiver: The receiver type.
20	:param type_validation: Whether to perform strict type validation.
21	:return: True if the types are compatible, False otherwise.
22	"""
23	if type_validation:	1✔
24	return _strict_types_are_compatible(sender, receiver)	1✔
25	else:
26	return True	×
27
28
29	def _safe_get_origin(_type: Type[T]) -> Union[Type[T], None]:	1✔
30	"""
31	Safely retrieves the origin type of a generic alias or returns the type itself if it's a built-in.
32
33	This function extends the behavior of `typing.get_origin()` by also handling plain built-in types
34	like `list`, `dict`, etc., which `get_origin()` would normally return `None` for.
35
36	:param _type: A type or generic alias (e.g., `list`, `list[int]`, `dict[str, int]`).
37
38	:returns: The origin type (e.g., `list`, `dict`), or `None` if the input is not a type.
39	"""
40	origin = get_origin(_type) or (_type if isinstance(_type, type) else None)	1✔
41	# We want to treat typing.Union and UnionType as the same for compatibility checks.
42	# So we convert UnionType to Union if it is detected.
43	if origin is _UnionType:	1✔
44	origin = Union	×
45	return origin	1✔
46
47
48	def _strict_types_are_compatible(sender, receiver): # pylint: disable=too-many-return-statements	1✔
49	"""
50	Checks whether the sender type is equal to or a subtype of the receiver type under strict validation.
51
52	Note: this method has no pretense to perform proper type matching. It especially does not deal with aliasing of
53	typing classes such as `List` or `Dict` to their runtime counterparts `list` and `dict`. It also does not deal well
54	with "bare" types, so `List` is treated differently from `List[Any]`, even though they should be the same.
55	Consider simplifying the typing of your components if you observe unexpected errors during component connection.
56
57	:param sender: The sender type.
58	:param receiver: The receiver type.
59	:return: True if the sender type is strictly compatible with the receiver type, False otherwise.
60	"""
61	if sender == receiver or receiver is Any:	1✔
62	return True	1✔
63
64	if sender is Any:	1✔
65	return False	1✔
66
67	try:	1✔
68	if issubclass(sender, receiver):	1✔
69	return True	1✔
70	except TypeError: # typing classes can't be used with issubclass, so we deal with them below	1✔
71	pass	1✔
72
73	sender_origin = _safe_get_origin(sender)	1✔
74	receiver_origin = _safe_get_origin(receiver)	1✔
75
76	if sender_origin is not Union and receiver_origin is Union:	1✔
77	return any(_strict_types_are_compatible(sender, union_arg) for union_arg in get_args(receiver))	1✔
78
79	# Both must have origins and they must be equal
80	if not (sender_origin and receiver_origin and sender_origin == receiver_origin):	1✔
81	return False	1✔
82
83	# Compare generic type arguments
84	sender_args = get_args(sender)	1✔
85	receiver_args = get_args(receiver)	1✔
86
87	# Handle Callable types
88	if sender_origin == receiver_origin == collections.abc.Callable:	1✔
89	return _check_callable_compatibility(sender_args, receiver_args)	1✔
90
91	# Handle bare types
92	if not sender_args and sender_origin:	1✔
93	sender_args = (Any,)	1✔
94	if not receiver_args and receiver_origin:	1✔
95	receiver_args = (Any,) * (len(sender_args) if sender_args else 1)	1✔
96
97	return not (len(sender_args) > len(receiver_args)) and all(	1✔
98	_strict_types_are_compatible(*args) for args in zip(sender_args, receiver_args)
99	)
100
101
102	def _check_callable_compatibility(sender_args, receiver_args):	1✔
103	"""Helper function to check compatibility of Callable types"""
104	if not receiver_args:	1✔
105	return True	1✔
106	if not sender_args:	1✔
107	sender_args = ([Any] * len(receiver_args[0]), Any)	1✔
108	# Standard Callable has two elements in args: argument list and return type
109	if len(sender_args) != 2 or len(receiver_args) != 2:	1✔
110	return False	×
111	# Return types must be compatible
112	if not _strict_types_are_compatible(sender_args[1], receiver_args[1]):	1✔
113	return False	1✔
114	# Input Arguments must be of same length
115	if len(sender_args[0]) != len(receiver_args[0]):	1✔
116	return False	1✔
117	return all(_strict_types_are_compatible(sender_args[0][i], receiver_args[0][i]) for i in range(len(sender_args[0])))	1✔
118
119
120	def _type_name(type_: Any) -> str:	1✔
121	"""
122	Util methods to get a nice readable representation of a type.
123
124	Handles Optional and Literal in a special way to make it more readable.
125	"""
126	# Literal args are strings, so we wrap them in quotes to make it clear
127	if isinstance(type_, str):	1✔
128	return f"'{type_}'"	1✔
129
130	if type_ is type(None):	1✔
131	return "None"	1✔
132
133	args = get_args(type_)	1✔
134
135	if isinstance(type_, _UnionType):	1✔
136	return " \| ".join([_type_name(a) for a in args])	×
137
138	name = getattr(type_, "__name__", str(type_))	1✔
139	if name.startswith("typing."):	1✔
140	name = name[7:]	1✔
141	if "[" in name:	1✔
142	name = name.split("[")[0]	1✔
143
144	if name == "Union" and type(None) in args and len(args) == 2:	1✔
145	# Optional is technically a Union of type and None
146	# but we want to display it as Optional
147	name = "Optional"	×
148
149	if args:	1✔
150	args_str = ", ".join([_type_name(a) for a in args if a is not type(None)])	1✔
151	return f"{name}[{args_str}]"	1✔
152
153	return f"{name}"	1✔

deepset-ai / haystack / 16493136922

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