13972131258

Committed 20 Mar 2025 02:43PM UTC coverage: 90.021% (-0.03%) from 90.054%

Build # 13972131258

Build Type

Pull #9069

github

Committed by

web-flow

Commit Message

Merge 8371761b0 into 67ab3788e

Pull Request Pull Request #9069: refactor!: `ChatMessage` serialization-deserialization updates

Run Details

9833 of 10923 relevant lines covered (90.02%)

0.9 hits per line

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

95.65

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

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)

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

deepset-ai / haystack / 13972131258

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