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

98.44

haystack/components/converters/output_adapter.py

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

import ast
import contextlib
from typing import Any, Callable, Optional, TypeAlias

import jinja2.runtime
from jinja2 import TemplateSyntaxError
from jinja2.nativetypes import NativeEnvironment
from jinja2.sandbox import SandboxedEnvironment

from haystack import component, default_from_dict, default_to_dict, logging
from haystack.utils import deserialize_callable, deserialize_type, serialize_callable, serialize_type
from haystack.utils.jinja2_extensions import _extract_template_variables_and_assignments

logger = logging.getLogger(__name__)


class OutputAdaptationException(Exception):
    """Exception raised when there is an error during output adaptation."""


@component
class OutputAdapter:
    """
    Adapts output of a Component using Jinja templates.

    Usage example:
    ```python
    from haystack import Document
    from haystack.components.converters import OutputAdapter

    adapter = OutputAdapter(template="{{ documents[0].content }}", output_type=str)
    documents = [Document(content="Test content"]
    result = adapter.run(documents=documents)

    assert result["output"] == "Test content"
    ```
    """

    def __init__(
        self,
        template: str,
        output_type: TypeAlias,
        custom_filters: Optional[dict[str, Callable]] = None,
        unsafe: bool = False,
    ) -> None:
        """
        Create an OutputAdapter component.

        :param template:
            A Jinja template that defines how to adapt the input data.
            The variables in the template define the input of this instance.
            e.g.
            With this template:
            ```
            {{ documents[0].content }}
            ```
            The Component input will be `documents`.
        :param output_type:
            The type of output this instance will return.
        :param custom_filters:
            A dictionary of custom Jinja filters used in the template.
        :param unsafe:
            Enable execution of arbitrary code in the Jinja template.
            This should only be used if you trust the source of the template as it can be lead to remote code execution.
        """
        self.custom_filters = {**(custom_filters or {})}
        input_types: set[str] = set()

        self._unsafe = unsafe

        if self._unsafe:
            msg = (
                "Unsafe mode is enabled. This allows execution of arbitrary code in the Jinja template. "
                "Use this only if you trust the source of the template."
            )
            logger.warning(msg)
        self._env = (
            NativeEnvironment() if self._unsafe else SandboxedEnvironment(undefined=jinja2.runtime.StrictUndefined)
        )

        try:
            self._env.parse(template)  # Validate template syntax
            self.template = template
        except TemplateSyntaxError as e:
            raise ValueError(f"Invalid Jinja template '{template}': {e}") from e

        for name, filter_func in self.custom_filters.items():
            self._env.filters[name] = filter_func

        # b) extract variables in the template
        assigned_variables, template_variables = _extract_template_variables_and_assignments(
            env=self._env, template=self.template
        )
        route_input_names = template_variables - assigned_variables
        input_types.update(route_input_names)

        # the env is not needed, discarded automatically
        component.set_input_types(self, **dict.fromkeys(input_types, Any))
        component.set_output_types(self, **{"output": output_type})
        self.output_type = output_type

    def run(self, **kwargs):
        """
        Renders the Jinja template with the provided inputs.

        :param kwargs:
            Must contain all variables used in the `template` string.
        :returns:
            A dictionary with the following keys:
            - `output`: Rendered Jinja template.

        :raises OutputAdaptationException: If template rendering fails.
        """
        # check if kwargs are empty
        if not kwargs:
            raise ValueError("No input data provided for output adaptation")
        for name, filter_func in self.custom_filters.items():
            self._env.filters[name] = filter_func
        adapted_outputs = {}
        try:
            adapted_output_template = self._env.from_string(self.template)
            output_result = adapted_output_template.render(**kwargs)
            if isinstance(output_result, jinja2.runtime.Undefined):
                raise OutputAdaptationException(f"Undefined variable in the template {self.template}; kwargs: {kwargs}")

            # We suppress the exception in case the output is already a string, otherwise
            # we try to evaluate it and would fail.
            # This must be done cause the output could be different literal structures.
            # This doesn't support any user types.
            with contextlib.suppress(Exception):
                if not self._unsafe:
                    output_result = ast.literal_eval(output_result)

            adapted_outputs["output"] = output_result
        except Exception as e:
            raise OutputAdaptationException(f"Error adapting {self.template} with {kwargs}: {e}") from e
        return adapted_outputs

    def to_dict(self) -> dict[str, Any]:
        """
        Serializes the component to a dictionary.

        :returns:
            Dictionary with serialized data.
        """
        se_filters = {name: serialize_callable(filter_func) for name, filter_func in self.custom_filters.items()}
        return default_to_dict(
            self,
            template=self.template,
            output_type=serialize_type(self.output_type),
            custom_filters=se_filters,
            unsafe=self._unsafe,
        )

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "OutputAdapter":
        """
        Deserializes the component from a dictionary.

        :param data:
            The dictionary to deserialize from.
        :returns:
            The deserialized component.
        """
        init_params = data.get("init_parameters", {})
        init_params["output_type"] = deserialize_type(init_params["output_type"])

        custom_filters = init_params.get("custom_filters", {})
        if custom_filters:
            init_params["custom_filters"] = {
                name: deserialize_callable(filter_func) if filter_func else None
                for name, filter_func in custom_filters.items()
            }
        return default_from_dict(cls, data)

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	import ast	1✔
6	import contextlib	1✔
7	from typing import Any, Callable, Optional, TypeAlias	1✔
8
9	import jinja2.runtime	1✔
10	from jinja2 import TemplateSyntaxError	1✔
11	from jinja2.nativetypes import NativeEnvironment	1✔
12	from jinja2.sandbox import SandboxedEnvironment	1✔
13
14	from haystack import component, default_from_dict, default_to_dict, logging	1✔
15	from haystack.utils import deserialize_callable, deserialize_type, serialize_callable, serialize_type	1✔
16	from haystack.utils.jinja2_extensions import _extract_template_variables_and_assignments	1✔
17
18	logger = logging.getLogger(__name__)	1✔
19
20
21	class OutputAdaptationException(Exception):	1✔
22	"""Exception raised when there is an error during output adaptation."""
23
24
25	@component	1✔
26	class OutputAdapter:	1✔
27	"""
28	Adapts output of a Component using Jinja templates.
29
30	Usage example:
31	```python
32	from haystack import Document
33	from haystack.components.converters import OutputAdapter
34
35	adapter = OutputAdapter(template="{{ documents[0].content }}", output_type=str)
36	documents = [Document(content="Test content"]
37	result = adapter.run(documents=documents)
38
39	assert result["output"] == "Test content"
40	```
41	"""
42
43	def __init__(	1✔
44	self,
45	template: str,
46	output_type: TypeAlias,
47	custom_filters: Optional[dict[str, Callable]] = None,
48	unsafe: bool = False,
49	) -> None:
50	"""
51	Create an OutputAdapter component.
52
53	:param template:
54	A Jinja template that defines how to adapt the input data.
55	The variables in the template define the input of this instance.
56	e.g.
57	With this template:
58	```
59	{{ documents[0].content }}
60	```
61	The Component input will be `documents`.
62	:param output_type:
63	The type of output this instance will return.
64	:param custom_filters:
65	A dictionary of custom Jinja filters used in the template.
66	:param unsafe:
67	Enable execution of arbitrary code in the Jinja template.
68	This should only be used if you trust the source of the template as it can be lead to remote code execution.
69	"""
70	self.custom_filters = {**(custom_filters or {})}	1✔
71	input_types: set[str] = set()	1✔
72
73	self._unsafe = unsafe	1✔
74
75	if self._unsafe:	1✔
76	msg = (	1✔
77	"Unsafe mode is enabled. This allows execution of arbitrary code in the Jinja template. "
78	"Use this only if you trust the source of the template."
79	)
80	logger.warning(msg)	1✔
81	self._env = (	1✔
82	NativeEnvironment() if self._unsafe else SandboxedEnvironment(undefined=jinja2.runtime.StrictUndefined)
83	)
84
85	try:	1✔
86	self._env.parse(template) # Validate template syntax	1✔
87	self.template = template	1✔
88	except TemplateSyntaxError as e:	1✔
89	raise ValueError(f"Invalid Jinja template '{template}': {e}") from e	1✔
90
91	for name, filter_func in self.custom_filters.items():	1✔
92	self._env.filters[name] = filter_func	1✔
93
94	# b) extract variables in the template
95	assigned_variables, template_variables = _extract_template_variables_and_assignments(	1✔
96	env=self._env, template=self.template
97	)
98	route_input_names = template_variables - assigned_variables	1✔
99	input_types.update(route_input_names)	1✔
100
101	# the env is not needed, discarded automatically
102	component.set_input_types(self, **dict.fromkeys(input_types, Any))	1✔
103	component.set_output_types(self, **{"output": output_type})	1✔
104	self.output_type = output_type	1✔
105
106	def run(self, **kwargs):	1✔
107	"""
108	Renders the Jinja template with the provided inputs.
109
110	:param kwargs:
111	Must contain all variables used in the `template` string.
112	:returns:
113	A dictionary with the following keys:
114	- `output`: Rendered Jinja template.
115
116	:raises OutputAdaptationException: If template rendering fails.
117	"""
118	# check if kwargs are empty
119	if not kwargs:	1✔
120	raise ValueError("No input data provided for output adaptation")	1✔
121	for name, filter_func in self.custom_filters.items():	1✔
122	self._env.filters[name] = filter_func	1✔
123	adapted_outputs = {}	1✔
124	try:	1✔
125	adapted_output_template = self._env.from_string(self.template)	1✔
126	output_result = adapted_output_template.render(**kwargs)	1✔
127	if isinstance(output_result, jinja2.runtime.Undefined):	1✔
128	raise OutputAdaptationException(f"Undefined variable in the template {self.template}; kwargs: {kwargs}")	×
129
130	# We suppress the exception in case the output is already a string, otherwise
131	# we try to evaluate it and would fail.
132	# This must be done cause the output could be different literal structures.
133	# This doesn't support any user types.
134	with contextlib.suppress(Exception):	1✔
135	if not self._unsafe:	1✔
136	output_result = ast.literal_eval(output_result)	1✔
137
138	adapted_outputs["output"] = output_result	1✔
139	except Exception as e:	1✔
140	raise OutputAdaptationException(f"Error adapting {self.template} with {kwargs}: {e}") from e	1✔
141	return adapted_outputs	1✔
142
143	def to_dict(self) -> dict[str, Any]:	1✔
144	"""
145	Serializes the component to a dictionary.
146
147	:returns:
148	Dictionary with serialized data.
149	"""
150	se_filters = {name: serialize_callable(filter_func) for name, filter_func in self.custom_filters.items()}	1✔
151	return default_to_dict(	1✔
152	self,
153	template=self.template,
154	output_type=serialize_type(self.output_type),
155	custom_filters=se_filters,
156	unsafe=self._unsafe,
157	)
158
159	@classmethod	1✔
160	def from_dict(cls, data: dict[str, Any]) -> "OutputAdapter":	1✔
161	"""
162	Deserializes the component from a dictionary.
163
164	:param data:
165	The dictionary to deserialize from.
166	:returns:
167	The deserialized component.
168	"""
169	init_params = data.get("init_parameters", {})	1✔
170	init_params["output_type"] = deserialize_type(init_params["output_type"])	1✔
171
172	custom_filters = init_params.get("custom_filters", {})	1✔
173	if custom_filters:	1✔
174	init_params["custom_filters"] = {	1✔
175	name: deserialize_callable(filter_func) if filter_func else None
176	for name, filter_func in custom_filters.items()
177	}
178	return default_from_dict(cls, data)	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