14174011772

Committed 31 Mar 2025 02:27PM UTC coverage: 90.191%. Remained the same

Build # 14174011772

Build Type

Pull #9146

github

Committed by

web-flow

Commit Message

Merge 58fc2377e into adc3dfc5d

Pull Request Pull Request #9146: test: change model to test HF API embedders

Run Details

10243 of 11357 relevant lines covered (90.19%)

0.9 hits per line

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

50.0

haystack/core/pipeline/descriptions.py

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

from typing import Dict, List

import networkx  # type:ignore

from haystack.core.component.types import InputSocket, OutputSocket
from haystack.core.type_utils import _type_name


def find_pipeline_inputs(
    graph: networkx.MultiDiGraph, include_connected_sockets: bool = False
) -> Dict[str, List[InputSocket]]:
    """
    Collect components that have disconnected/connected input sockets.

    Note that this method returns *ALL* disconnected input sockets, including all such sockets with default values.
    """
    return {
        name: [
            socket
            for socket in data.get("input_sockets", {}).values()
            if socket.is_variadic or (include_connected_sockets or not socket.senders)
        ]
        for name, data in graph.nodes(data=True)
    }


def find_pipeline_outputs(
    graph: networkx.MultiDiGraph, include_connected_sockets: bool = False
) -> Dict[str, List[OutputSocket]]:
    """
    Collect components that have disconnected/connected output sockets. They define the pipeline output.
    """
    return {
        name: [
            socket
            for socket in data.get("output_sockets", {}).values()
            if (include_connected_sockets or not socket.receivers)
        ]
        for name, data in graph.nodes(data=True)
    }


def describe_pipeline_inputs(graph: networkx.MultiDiGraph):
    """
    Returns a dictionary with the input names and types that this pipeline accepts.
    """
    inputs = {
        comp: {socket.name: {"type": socket.type, "is_mandatory": socket.is_mandatory} for socket in data}
        for comp, data in find_pipeline_inputs(graph).items()
        if data
    }
    return inputs


def describe_pipeline_inputs_as_string(graph: networkx.MultiDiGraph):
    """
    Returns a string representation of the input names and types that this pipeline accepts.
    """
    inputs = describe_pipeline_inputs(graph)
    message = "This pipeline expects the following inputs:\n"
    for comp, sockets in inputs.items():
        if sockets:
            message += f"- {comp}:\n"
            for name, socket in sockets.items():
                message += f"    - {name}: {_type_name(socket['type'])}\n"
    return message

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	from typing import Dict, List	1✔
6
7	import networkx # type:ignore	1✔
8
9	from haystack.core.component.types import InputSocket, OutputSocket	1✔
10	from haystack.core.type_utils import _type_name	1✔
11
12
13	def find_pipeline_inputs(	1✔
14	graph: networkx.MultiDiGraph, include_connected_sockets: bool = False
15	) -> Dict[str, List[InputSocket]]:
16	"""
17	Collect components that have disconnected/connected input sockets.
18
19	Note that this method returns ALL disconnected input sockets, including all such sockets with default values.
20	"""
21	return {	1✔
22	name: [
23	socket
24	for socket in data.get("input_sockets", {}).values()
25	if socket.is_variadic or (include_connected_sockets or not socket.senders)
26	]
27	for name, data in graph.nodes(data=True)
28	}
29
30
31	def find_pipeline_outputs(	1✔
32	graph: networkx.MultiDiGraph, include_connected_sockets: bool = False
33	) -> Dict[str, List[OutputSocket]]:
34	"""
35	Collect components that have disconnected/connected output sockets. They define the pipeline output.
36	"""
37	return {	1✔
38	name: [
39	socket
40	for socket in data.get("output_sockets", {}).values()
41	if (include_connected_sockets or not socket.receivers)
42	]
43	for name, data in graph.nodes(data=True)
44	}
45
46
47	def describe_pipeline_inputs(graph: networkx.MultiDiGraph):	1✔
48	"""
49	Returns a dictionary with the input names and types that this pipeline accepts.
50	"""
51	inputs = {	×
52	comp: {socket.name: {"type": socket.type, "is_mandatory": socket.is_mandatory} for socket in data}
53	for comp, data in find_pipeline_inputs(graph).items()
54	if data
55	}
56	return inputs	×
57
58
59	def describe_pipeline_inputs_as_string(graph: networkx.MultiDiGraph):	1✔
60	"""
61	Returns a string representation of the input names and types that this pipeline accepts.
62	"""
63	inputs = describe_pipeline_inputs(graph)	×
64	message = "This pipeline expects the following inputs:\n"	×
65	for comp, sockets in inputs.items():	×
66	if sockets:	×
67	message += f"- {comp}:\n"	×
68	for name, socket in sockets.items():	×
69	message += f" - {name}: {_type_name(socket['type'])}\n"	×
70	return message	×

deepset-ai / haystack / 14174011772

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