13432090183

Committed 20 Feb 2025 09:30AM UTC coverage: 91.155% (+0.007%) from 91.148%

Build # 13432090183

Build Type

Pull #8875

github

Committed by

web-flow

Commit Message

Merge 013fa6d37 into 8cafcddb0

Pull Request Pull Request #8875: feat: Add Type Validation parameter for Pipeline Connections

Run Details

9430 of 10345 relevant lines covered (91.16%)

0.91 hits per line

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

95.45

haystack/core/type_utils.py

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

from typing import Any, TypeVar, Union, get_args, get_origin

from haystack import logging

logger = logging.getLogger(__name__)

T = TypeVar("T")


def _types_are_compatible(sender, receiver, 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 _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 = get_origin(sender)
    receiver_origin = 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)
    if len(sender_args) > len(receiver_args):
        return False

    return all(_strict_types_are_compatible(*args) for args in zip(sender_args, receiver_args))


def _type_name(type_):
    """
    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_}'"

    name = getattr(type_, "__name__", str(type_))

    if name.startswith("typing."):
        name = name[7:]
    if "[" in name:
        name = name.split("[")[0]
    args = get_args(type_)
    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 = ", ".join([_type_name(a) for a in args if a is not type(None)])
        return f"{name}[{args}]"

    return f"{name}"

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	from typing import Any, TypeVar, Union, get_args, get_origin	1✔
6
7	from haystack import logging	1✔
8
9	logger = logging.getLogger(__name__)	1✔
10
11	T = TypeVar("T")	1✔
12
13
14	def _types_are_compatible(sender, receiver, 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 _strict_types_are_compatible(sender, receiver): # pylint: disable=too-many-return-statements	1✔
30	"""
31	Checks whether the sender type is equal to or a subtype of the receiver type under strict validation.
32
33	Note: this method has no pretense to perform proper type matching. It especially does not deal with aliasing of
34	typing classes such as `List` or `Dict` to their runtime counterparts `list` and `dict`. It also does not deal well
35	with "bare" types, so `List` is treated differently from `List[Any]`, even though they should be the same.
36	Consider simplifying the typing of your components if you observe unexpected errors during component connection.
37
38	:param sender: The sender type.
39	:param receiver: The receiver type.
40	:return: True if the sender type is strictly compatible with the receiver type, False otherwise.
41	"""
42	if sender == receiver or receiver is Any:	1✔
43	return True	1✔
44
45	if sender is Any:	1✔
46	return False	1✔
47
48	try:	1✔
49	if issubclass(sender, receiver):	1✔
50	return True	1✔
51	except TypeError: # typing classes can't be used with issubclass, so we deal with them below	1✔
52	pass	1✔
53
54	sender_origin = get_origin(sender)	1✔
55	receiver_origin = get_origin(receiver)	1✔
56
57	if sender_origin is not Union and receiver_origin is Union:	1✔
58	return any(_strict_types_are_compatible(sender, union_arg) for union_arg in get_args(receiver))	1✔
59
60	# Both must have origins and they must be equal
61	if not (sender_origin and receiver_origin and sender_origin == receiver_origin):	1✔
62	return False	1✔
63
64	# Compare generic type arguments
65	sender_args = get_args(sender)	1✔
66	receiver_args = get_args(receiver)	1✔
67	if len(sender_args) > len(receiver_args):	1✔
68	return False	1✔
69
70	return all(_strict_types_are_compatible(*args) for args in zip(sender_args, receiver_args))	1✔
71
72
73	def _type_name(type_):	1✔
74	"""
75	Util methods to get a nice readable representation of a type.
76
77	Handles Optional and Literal in a special way to make it more readable.
78	"""
79	# Literal args are strings, so we wrap them in quotes to make it clear
80	if isinstance(type_, str):	1✔
81	return f"'{type_}'"	1✔
82
83	name = getattr(type_, "__name__", str(type_))	1✔
84
85	if name.startswith("typing."):	1✔
86	name = name[7:]	1✔
87	if "[" in name:	1✔
88	name = name.split("[")[0]	1✔
89	args = get_args(type_)	1✔
90	if name == "Union" and type(None) in args and len(args) == 2:	1✔
91	# Optional is technically a Union of type and None
92	# but we want to display it as Optional
93	name = "Optional"	×
94
95	if args:	1✔
96	args = ", ".join([_type_name(a) for a in args if a is not type(None)])	1✔
97	return f"{name}[{args}]"	1✔
98
99	return f"{name}"	1✔

deepset-ai / haystack / 13432090183

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