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

91.67

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, TypeAlias, TypedDict, TypeVar, get_args

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