15049844454

Committed 15 May 2025 04:07PM UTC coverage: 90.446% (+0.04%) from 90.41%

Build # 15049844454

Build Type

Pull #9345

github

Committed by

web-flow

Commit Message

Merge 9e4071f83 into 2a64cd4e9

Pull Request Pull Request #9345: feat: add serialization to `State` / move `State` to utils

Run Details

10981 of 12141 relevant lines covered (90.45%)

0.9 hits per line

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

97.22

haystack/tools/serde_utils.py

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

from typing import Any, Dict, List, Union

from haystack.core.errors import DeserializationError
from haystack.core.serialization import import_class_by_name
from haystack.tools.tool import Tool
from haystack.tools.toolset import Toolset


def serialize_tools_or_toolset(
    tools: Union[Toolset, List[Tool], None],
) -> Union[Dict[str, Any], List[Dict[str, Any]], None]:
    """
    Serialize a Toolset or a list of Tools to a dictionary or a list of tool dictionaries.

    :param tools: A Toolset, a list of Tools, or None
    :returns: A dictionary, a list of tool dictionaries, or None if tools is None
    """
    if tools is None:
        return None
    if isinstance(tools, Toolset):
        return tools.to_dict()
    return [tool.to_dict() for tool in tools]


def deserialize_tools_or_toolset_inplace(data: Dict[str, Any], key: str = "tools"):
    """
    Deserialize a list of Tools or a Toolset in a dictionary inplace.

    :param data:
        The dictionary with the serialized data.
    :param key:
        The key in the dictionary where the list of Tools or Toolset is stored.
    """
    if key in data:
        serialized_tools = data[key]

        if serialized_tools is None:
            return

        # Check if it's a serialized Toolset (a dict with "type" and "data" keys)
        if isinstance(serialized_tools, dict) and all(k in serialized_tools for k in ["type", "data"]):
            toolset_class_name = serialized_tools.get("type")
            if not toolset_class_name:
                raise DeserializationError("The 'type' key is missing or None in the serialized toolset data")

            toolset_class = import_class_by_name(toolset_class_name)

            if not issubclass(toolset_class, Toolset):
                raise TypeError(f"Class '{toolset_class}' is not a subclass of Toolset")

            data[key] = toolset_class.from_dict(serialized_tools)
            return

        if not isinstance(serialized_tools, list):
            raise TypeError(f"The value of '{key}' is not a list or a dictionary")

        deserialized_tools = []
        for tool in serialized_tools:
            if not isinstance(tool, dict):
                raise TypeError(f"Serialized tool '{tool}' is not a dictionary")

            # different classes are allowed: Tool, ComponentTool, etc.
            tool_class = import_class_by_name(tool["type"])
            if not issubclass(tool_class, Tool):
                raise TypeError(f"Class '{tool_class}' is not a subclass of Tool")

            deserialized_tools.append(tool_class.from_dict(tool))

        data[key] = deserialized_tools

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	from typing import Any, Dict, List, Union	1✔
6
7	from haystack.core.errors import DeserializationError	1✔
8	from haystack.core.serialization import import_class_by_name	1✔
9	from haystack.tools.tool import Tool	1✔
10	from haystack.tools.toolset import Toolset	1✔
11
12
13	def serialize_tools_or_toolset(	1✔
14	tools: Union[Toolset, List[Tool], None],
15	) -> Union[Dict[str, Any], List[Dict[str, Any]], None]:
16	"""
17	Serialize a Toolset or a list of Tools to a dictionary or a list of tool dictionaries.
18
19	:param tools: A Toolset, a list of Tools, or None
20	:returns: A dictionary, a list of tool dictionaries, or None if tools is None
21	"""
22	if tools is None:	1✔
23	return None	1✔
24	if isinstance(tools, Toolset):	1✔
25	return tools.to_dict()	1✔
26	return [tool.to_dict() for tool in tools]	1✔
27
28
29	def deserialize_tools_or_toolset_inplace(data: Dict[str, Any], key: str = "tools"):	1✔
30	"""
31	Deserialize a list of Tools or a Toolset in a dictionary inplace.
32
33	:param data:
34	The dictionary with the serialized data.
35	:param key:
36	The key in the dictionary where the list of Tools or Toolset is stored.
37	"""
38	if key in data:	1✔
39	serialized_tools = data[key]	1✔
40
41	if serialized_tools is None:	1✔
42	return	1✔
43
44	# Check if it's a serialized Toolset (a dict with "type" and "data" keys)
45	if isinstance(serialized_tools, dict) and all(k in serialized_tools for k in ["type", "data"]):	1✔
46	toolset_class_name = serialized_tools.get("type")	1✔
47	if not toolset_class_name:	1✔
48	raise DeserializationError("The 'type' key is missing or None in the serialized toolset data")	×
49
50	toolset_class = import_class_by_name(toolset_class_name)	1✔
51
52	if not issubclass(toolset_class, Toolset):	1✔
53	raise TypeError(f"Class '{toolset_class}' is not a subclass of Toolset")	1✔
54
55	data[key] = toolset_class.from_dict(serialized_tools)	1✔
56	return	1✔
57
58	if not isinstance(serialized_tools, list):	1✔
59	raise TypeError(f"The value of '{key}' is not a list or a dictionary")	1✔
60
61	deserialized_tools = []	1✔
62	for tool in serialized_tools:	1✔
63	if not isinstance(tool, dict):	1✔
64	raise TypeError(f"Serialized tool '{tool}' is not a dictionary")	1✔
65
66	# different classes are allowed: Tool, ComponentTool, etc.
67	tool_class = import_class_by_name(tool["type"])	1✔
68	if not issubclass(tool_class, Tool):	1✔
69	raise TypeError(f"Class '{tool_class}' is not a subclass of Tool")	1✔
70
71	deserialized_tools.append(tool_class.from_dict(tool))	1✔
72
73	data[key] = deserialized_tools	1✔

deepset-ai / haystack / 15049844454

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