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

94.2

haystack/core/type_utils.py

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

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