20373472849

Committed 19 Dec 2025 02:47PM UTC coverage: 92.261% (+0.08%) from 92.183%

Build # 20373472849

Build Type

push

github

Committed by

web-flow

Commit Message

refactor: support PEP604 typing syntax (`X|Y` and `X|None`)  (#10255)

* adopt X|Y syntax: draft

* cast Union

* fix pylint + state testing

* use X|Y

* rm unused imports

* trigger e2e tests

* fix + simplification

* add compatibility tests

* rm e2e tests trigger

* fix

* add relnote

* simplify/fix pep604 union parsing

* fix comments

* test _is_optional_type

* introduce _build_pep604_union_type; make _is_union_type private

* try removing problematic test

Run Details

14186 of 15376 relevant lines covered (92.26%)

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 types import NoneType, UnionType
from typing import Any, TypeVar, Union, get_args, get_origin

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

deepset-ai / haystack / 20373472849

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