16754475749

Committed 05 Aug 2025 03:26PM UTC coverage: 91.946% (+0.05%) from 91.901%

Build # 16754475749

Build Type

Pull #9678

github

Committed by

web-flow

Commit Message

Merge 31abaf9ce into 323274e17

Pull Request Pull Request #9678: Chore/pep585 type hints

Run Details

12774 of 13893 relevant lines covered (91.95%)

0.92 hits per line

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

91.89

haystack/core/component/types.py

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

from dataclasses import dataclass, field
from typing import Annotated, Any, Iterable, TypedDict, TypeVar, get_args

from typing_extensions import TypeAlias  # Python 3.9 compatibility

HAYSTACK_VARIADIC_ANNOTATION = "__haystack__variadic_t"
HAYSTACK_GREEDY_VARIADIC_ANNOTATION = "__haystack__greedy_variadic_t"

# # Generic type variable used in the Variadic container
T = TypeVar("T")


# Variadic is a custom annotation type we use to mark input types.
# This type doesn't do anything else than "marking" the contained
# type so it can be used in the `InputSocket` creation where we
# check that its annotation equals to HAYSTACK_VARIADIC_ANNOTATION
Variadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_VARIADIC_ANNOTATION]

# GreedyVariadic type is similar to Variadic.
# The only difference is the way it's treated by the Pipeline when input is received
# in a socket with this type.
# Instead of waiting for other inputs to be received, Components that have a GreedyVariadic
# input will be run right after receiving the first input.
# Even if there are multiple connections to that socket.
GreedyVariadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_GREEDY_VARIADIC_ANNOTATION]


class _empty:
    """Custom object for marking InputSocket.default_value as not set."""


@dataclass
class InputSocket:
    """
    Represents an input of a `Component`.

    :param name:
        The name of the input.
    :param type:
        The type of the input.
    :param default_value:
        The default value of the input. If not set, the input is mandatory.
    :param is_variadic:
        Whether the input is variadic or not.
    :param is_greedy
        Whether the input is a greedy variadic or not.
    :param senders:
        The list of components that send data to this input.
    """

    name: str
    type: type
    default_value: Any = _empty
    is_variadic: bool = field(init=False)
    is_greedy: bool = field(init=False)
    senders: list[str] = field(default_factory=list)

    @property
    def is_mandatory(self):
        """Check if the input is mandatory."""
        return self.default_value == _empty

    def __post_init__(self):
        try:
            # __metadata__ is a tuple
            self.is_variadic = hasattr(self.type, "__metadata__") and self.type.__metadata__[0] in [
                HAYSTACK_VARIADIC_ANNOTATION,
                HAYSTACK_GREEDY_VARIADIC_ANNOTATION,
            ]
            self.is_greedy = (
                hasattr(self.type, "__metadata__") and self.type.__metadata__[0] == HAYSTACK_GREEDY_VARIADIC_ANNOTATION
            )
        except AttributeError:
            self.is_variadic = False
            self.is_greedy = False
        if self.is_variadic:
            # We need to "unpack" the type inside the Variadic annotation,
            # otherwise the pipeline connection api will try to match
            # `Annotated[type, HAYSTACK_VARIADIC_ANNOTATION]`.
            #
            # Note1: Variadic is expressed as an annotation of one single type,
            # so the return value of get_args will always be a one-item tuple.
            #
            # Note2: a pipeline always passes a list of items when a component
            # input is declared as Variadic, so the type itself always wraps
            # an iterable of the declared type. For example, Variadic[int]
            # is eventually an alias for Iterable[int]. Since we're interested
            # in getting the inner type `int`, we call `get_args` twice: the
            # first time to get `List[int]` out of `Variadic`, the second time
            # to get `int` out of `List[int]`.
            self.type = get_args(get_args(self.type)[0])[0]


class InputSocketTypeDescriptor(TypedDict):
    """
    Describes the type of an `InputSocket`.
    """

    type: type
    is_mandatory: bool


@dataclass
class OutputSocket:
    """
    Represents an output of a `Component`.

    :param name:
        The name of the output.
    :param type:
        The type of the output.
    :param receivers:
        The list of components that receive the output of this component.
    """

    name: str
    type: type
    receivers: list[str] = field(default_factory=list)

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	from dataclasses import dataclass, field	1✔
6	from typing import Annotated, Any, Iterable, TypedDict, TypeVar, get_args	1✔
7
8	from typing_extensions import TypeAlias # Python 3.9 compatibility	1✔
9
10	HAYSTACK_VARIADIC_ANNOTATION = "__haystack__variadic_t"	1✔
11	HAYSTACK_GREEDY_VARIADIC_ANNOTATION = "__haystack__greedy_variadic_t"	1✔
12
13	# # Generic type variable used in the Variadic container
14	T = TypeVar("T")	1✔
15
16
17	# Variadic is a custom annotation type we use to mark input types.
18	# This type doesn't do anything else than "marking" the contained
19	# type so it can be used in the `InputSocket` creation where we
20	# check that its annotation equals to HAYSTACK_VARIADIC_ANNOTATION
21	Variadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_VARIADIC_ANNOTATION]	1✔
22
23	# GreedyVariadic type is similar to Variadic.
24	# The only difference is the way it's treated by the Pipeline when input is received
25	# in a socket with this type.
26	# Instead of waiting for other inputs to be received, Components that have a GreedyVariadic
27	# input will be run right after receiving the first input.
28	# Even if there are multiple connections to that socket.
29	GreedyVariadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_GREEDY_VARIADIC_ANNOTATION]	1✔
30
31
32	class _empty:	1✔
33	"""Custom object for marking InputSocket.default_value as not set."""
34
35
36	@dataclass	1✔
37	class InputSocket:	1✔
38	"""
39	Represents an input of a `Component`.
40
41	:param name:
42	The name of the input.
43	:param type:
44	The type of the input.
45	:param default_value:
46	The default value of the input. If not set, the input is mandatory.
47	:param is_variadic:
48	Whether the input is variadic or not.
49	:param is_greedy
50	Whether the input is a greedy variadic or not.
51	:param senders:
52	The list of components that send data to this input.
53	"""
54
55	name: str	1✔
56	type: type	1✔
57	default_value: Any = _empty	1✔
58	is_variadic: bool = field(init=False)	1✔
59	is_greedy: bool = field(init=False)	1✔
60	senders: list[str] = field(default_factory=list)	1✔
61
62	@property	1✔
63	def is_mandatory(self):	1✔
64	"""Check if the input is mandatory."""
65	return self.default_value == _empty	1✔
66
67	def __post_init__(self):	1✔
68	try:	1✔
69	# __metadata__ is a tuple
70	self.is_variadic = hasattr(self.type, "__metadata__") and self.type.__metadata__[0] in [	1✔
71	HAYSTACK_VARIADIC_ANNOTATION,
72	HAYSTACK_GREEDY_VARIADIC_ANNOTATION,
73	]
74	self.is_greedy = (	1✔
75	hasattr(self.type, "__metadata__") and self.type.__metadata__[0] == HAYSTACK_GREEDY_VARIADIC_ANNOTATION
76	)
77	except AttributeError:	×
78	self.is_variadic = False	×
79	self.is_greedy = False	×
80	if self.is_variadic:	1✔
81	# We need to "unpack" the type inside the Variadic annotation,
82	# otherwise the pipeline connection api will try to match
83	# `Annotated[type, HAYSTACK_VARIADIC_ANNOTATION]`.
84	#
85	# Note1: Variadic is expressed as an annotation of one single type,
86	# so the return value of get_args will always be a one-item tuple.
87	#
88	# Note2: a pipeline always passes a list of items when a component
89	# input is declared as Variadic, so the type itself always wraps
90	# an iterable of the declared type. For example, Variadic[int]
91	# is eventually an alias for Iterable[int]. Since we're interested
92	# in getting the inner type `int`, we call `get_args` twice: the
93	# first time to get `List[int]` out of `Variadic`, the second time
94	# to get `int` out of `List[int]`.
95	self.type = get_args(get_args(self.type)[0])[0]	1✔
96
97
98	class InputSocketTypeDescriptor(TypedDict):	1✔
99	"""
100	Describes the type of an `InputSocket`.
101	"""
102
103	type: type	1✔
104	is_mandatory: bool	1✔
105
106
107	@dataclass	1✔
108	class OutputSocket:	1✔
109	"""
110	Represents an output of a `Component`.
111
112	:param name:
113	The name of the output.
114	:param type:
115	The type of the output.
116	:param receivers:
117	The list of components that receive the output of this component.
118	"""
119
120	name: str	1✔
121	type: type	1✔
122	receivers: list[str] = field(default_factory=list)	1✔

deepset-ai / haystack / 16754475749

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