diff --git a/src/backend/tests/unit/components/processing/test_data_operations_component.py b/src/backend/tests/unit/components/processing/test_data_operations_component.py index b713668158e0..3e9fddf7c44c 100644 --- a/src/backend/tests/unit/components/processing/test_data_operations_component.py +++ b/src/backend/tests/unit/components/processing/test_data_operations_component.py @@ -1,8 +1,11 @@ +import copy import importlib import sys import pytest -from lfx.components.processing.data_operations import DataOperationsComponent +from lfx.components.processing.data_operations import ACTION_CONFIG, DataOperationsComponent +from lfx.custom.utils import build_custom_component_template +from lfx.graph.graph.base import Graph from lfx.schema import Data from tests.base import ComponentTestBaseWithoutClient @@ -243,3 +246,113 @@ def test_update_build_config_clears_input_fields_when_operation_removed(self): # Default fields (operations, data) should remain visible assert result["operations"]["show"] is True assert result["data"]["show"] is True + + +class TestUpdateFrontendNodeRestoresVisibility: + """Clicking Update on a saved JSON Operations node must keep its operation's fields visible. + + ``update_frontend_node`` rebuilds the template from class defaults, where every operation field + has ``show=False``. The generic merge restores only values, so before the fix the selected + operation's input disappeared and ``ParameterHandler`` dropped it from the run. + """ + + NODE_ID = "DataOperations-test" + + @staticmethod + def _saved_node(operation: str | None, **values) -> dict: + """Return a node as the canvas saves it after the user picks ``operation`` and fills ``values``.""" + node, _ = build_custom_component_template(DataOperationsComponent(), user_id="test-user") + template = node["template"] + component = DataOperationsComponent() + component.update_build_config(template, [{"name": operation}] if operation else [], "operations") + for field, value in values.items(): + template[field]["value"] = value + if field == "mapped_json_display": + component.update_build_config(template, value, field) + return node + + @staticmethod + async def _update(saved: dict) -> dict: + """Run the Update Component path: a fresh class-default node merged with the saved one.""" + fresh, _ = build_custom_component_template(DataOperationsComponent(), user_id="test-user") + return await DataOperationsComponent().update_frontend_node(fresh, copy.deepcopy(saved)) + + def _run(self, node: dict, data: Data) -> Data: + """Run ``node`` with the params a graph build hands the component, which skips hidden fields.""" + payload = { + "nodes": [{"id": self.NODE_ID, "data": {"id": self.NODE_ID, "type": "DataOperations", "node": node}}], + "edges": [], + } + graph = Graph.from_payload(payload, flow_id="test-flow", user_id="test-user") + params = {k: v for k, v in graph.get_vertex(self.NODE_ID).params.items() if k != "code"} + return DataOperationsComponent(data=data, **params).as_data() + + @staticmethod + def _visibility(template: dict) -> dict[str, bool]: + return {field: template[field]["show"] for field in DataOperationsComponent.ALL_OPERATION_FIELDS} + + @pytest.mark.parametrize("operation", list(DataOperationsComponent.actions_data)) + async def test_update_keeps_saved_visibility(self, operation): + saved = self._saved_node(operation) + template = (await self._update(saved))["template"] + + shown = {field for field, show in self._visibility(template).items() if show} + assert shown == set(DataOperationsComponent.actions_data[operation]) - {"operations"} + assert self._visibility(template) == self._visibility(saved["template"]) + assert template["data"]["show"] is True + assert template["operations"]["show"] is True + assert template["data"]["is_list"] is ACTION_CONFIG[operation]["is_list"] + assert template["operations"]["value"] == [{"name": operation}] + + async def test_saved_select_keys_value_survives(self): + template = (await self._update(self._saved_node("Select Keys", select_keys_input=["text"])))["template"] + assert template["select_keys_input"]["show"] is True + assert template["select_keys_input"]["value"] == ["text"] + assert template["data"]["is_list"] is False + + async def test_no_operation_hides_all_operation_fields(self): + template = (await self._update(self._saved_node(None)))["template"] + assert not any(self._visibility(template).values()) + assert template["data"]["show"] is True + + async def test_removed_operation_hides_all_operation_fields(self): + template = (await self._update(self._saved_node("Filter Values")))["template"] + assert not any(self._visibility(template).values()) + assert template["data"]["show"] is True + + async def test_path_selection_rebuilds_selected_key_options(self): + saved = self._saved_node("Path Selection", mapped_json_display='{"a": {"b": 1}}', selected_key=".a.b") + template = (await self._update(saved))["template"] + assert template["selected_key"]["show"] is True + assert template["selected_key"]["value"] == ".a.b" + assert template["selected_key"]["options"] == [".a", ".a.b"] + + async def test_path_selection_with_invalid_saved_json_hides_selected_key(self): + saved = self._saved_node("Path Selection", mapped_json_display="{not json") + template = (await self._update(saved))["template"] + assert template["mapped_json_display"]["show"] is True + assert template["selected_key"]["show"] is False + + @pytest.mark.parametrize( + ("operation", "values", "expected"), + [ + ("Select Keys", {"select_keys_input": ["text"]}, {"text": "hello"}), + ("Remove Keys", {"remove_keys_input": ["extra"]}, {"text": "hello"}), + ("Rename Keys", {"rename_keys_input": {"text": "body"}}, {"body": "hello", "extra": 1}), + ("Append or Update", {"append_update_data": {"tag": "new"}}, {"text": "hello", "extra": 1, "tag": "new"}), + ("JQ Expression", {"query": ".text"}, {"result": "hello"}), + ( + "Path Selection", + {"mapped_json_display": '{"text": "x", "extra": 0}', "selected_key": ".text"}, + {"result": "hello"}, + ), + ], + ) + async def test_run_output_unchanged_after_update(self, operation, values, expected): + """The smoke-test case: Select Keys ``['text']`` returned ``{}`` after Update.""" + saved = self._saved_node(operation, **values) + updated = await self._update(saved) + data = Data(data={"text": "hello", "extra": 1}) + + assert self._run(saved, data).data == expected + assert self._run(updated, data).data == expected diff --git a/src/lfx/src/lfx/_assets/component_index.json b/src/lfx/src/lfx/_assets/component_index.json index b65d30629d68..c44bc5f0d1ab 100644 --- a/src/lfx/src/lfx/_assets/component_index.json +++ b/src/lfx/src/lfx/_assets/component_index.json @@ -24328,7 +24328,7 @@ "icon": "file-json", "legacy": true, "metadata": { - "code_hash": "6712afedc055", + "code_hash": "1a352d7d953f", "component_class_name": "DataOperationsComponent", "dependencies": { "dependencies": [ @@ -24438,7 +24438,7 @@ "show": true, "title_case": false, "type": "code", - "value": "import ast\nimport json\nfrom typing import TYPE_CHECKING, Any\n\nfrom json_repair import repair_json\n\nfrom lfx.custom import Component\nfrom lfx.inputs import DictInput, DropdownInput, MessageTextInput, SortableListInput\nfrom lfx.io import DataInput, MultilineInput, Output\nfrom lfx.log.logger import logger\nfrom lfx.schema import Data\nfrom lfx.schema.dotdict import dotdict\nfrom lfx.utils.component_utils import set_current_fields, set_field_display\nfrom lfx.utils.jq_security import validate_jq_program\n\nif TYPE_CHECKING:\n from collections.abc import Callable\n\nACTION_CONFIG = {\n \"Select Keys\": {\"is_list\": False, \"log_msg\": \"setting filter fields\"},\n \"Literal Eval\": {\"is_list\": False, \"log_msg\": \"setting evaluate fields\"},\n \"Combine\": {\"is_list\": True, \"log_msg\": \"setting combine fields\"},\n \"Append or Update\": {\"is_list\": False, \"log_msg\": \"setting Append or Update fields\"},\n \"Remove Keys\": {\"is_list\": False, \"log_msg\": \"setting remove keys fields\"},\n \"Rename Keys\": {\"is_list\": False, \"log_msg\": \"setting rename keys fields\"},\n \"Path Selection\": {\"is_list\": False, \"log_msg\": \"setting mapped key extractor fields\"},\n \"JQ Expression\": {\"is_list\": False, \"log_msg\": \"setting parse json fields\"},\n}\n\n\nclass DataOperationsComponent(Component):\n display_name = \"JSON Operations\"\n description = \"Perform various operations on a JSON object.\"\n icon = \"file-json\"\n name = \"DataOperations\"\n legacy = True\n replacement = [\"processing.Operations\"]\n default_keys = [\"operations\", \"data\"]\n metadata = {\n \"keywords\": [\n \"data\",\n \"json\",\n \"operations\",\n \"Append or Update\",\n \"remove keys\",\n \"rename keys\",\n \"select keys\",\n \"literal eval\",\n \"combine\",\n \"append\",\n \"update\",\n \"remove\",\n \"rename\",\n \"data operations\",\n \"json operations\",\n \"data manipulation\",\n \"data transformation\",\n \"data filtering\",\n \"data selection\",\n \"data combination\",\n \"Parse JSON\",\n \"JSON Query\",\n \"JQ Query\",\n ],\n }\n actions_data = {\n \"Select Keys\": [\"select_keys_input\", \"operations\"],\n \"Literal Eval\": [],\n \"Combine\": [],\n \"Append or Update\": [\"append_update_data\", \"operations\"],\n \"Remove Keys\": [\"remove_keys_input\", \"operations\"],\n \"Rename Keys\": [\"rename_keys_input\", \"operations\"],\n \"Path Selection\": [\"mapped_json_display\", \"selected_key\", \"operations\"],\n \"JQ Expression\": [\"query\", \"operations\"],\n }\n\n # All operation-specific input fields (used to hide and reset when no operation selected).\n ALL_OPERATION_FIELDS = [\n \"select_keys_input\",\n \"append_update_data\",\n \"remove_keys_input\",\n \"rename_keys_input\",\n \"mapped_json_display\",\n \"selected_key\",\n \"query\",\n ]\n\n @staticmethod\n def extract_all_paths(obj, path=\"\"):\n paths = []\n if isinstance(obj, dict):\n for k, v in obj.items():\n new_path = f\"{path}.{k}\" if path else f\".{k}\"\n paths.append(new_path)\n paths.extend(DataOperationsComponent.extract_all_paths(v, new_path))\n elif isinstance(obj, list) and obj:\n new_path = f\"{path}[0]\"\n paths.append(new_path)\n paths.extend(DataOperationsComponent.extract_all_paths(obj[0], new_path))\n return paths\n\n @staticmethod\n def remove_keys_recursive(obj, keys_to_remove):\n if isinstance(obj, dict):\n return {\n k: DataOperationsComponent.remove_keys_recursive(v, keys_to_remove)\n for k, v in obj.items()\n if k not in keys_to_remove\n }\n if isinstance(obj, list):\n return [DataOperationsComponent.remove_keys_recursive(item, keys_to_remove) for item in obj]\n return obj\n\n @staticmethod\n def rename_keys_recursive(obj, rename_map):\n if isinstance(obj, dict):\n return {\n rename_map.get(k, k): DataOperationsComponent.rename_keys_recursive(v, rename_map)\n for k, v in obj.items()\n }\n if isinstance(obj, list):\n return [DataOperationsComponent.rename_keys_recursive(item, rename_map) for item in obj]\n return obj\n\n inputs = [\n DataInput(name=\"data\", display_name=\"JSON\", info=\"Data object to filter.\", required=True, is_list=True),\n SortableListInput(\n name=\"operations\",\n display_name=\"Operations\",\n placeholder=\"Select Operation\",\n info=\"List of operations to perform on the data.\",\n options=[\n {\"name\": \"Select Keys\", \"icon\": \"lasso-select\"},\n {\"name\": \"Literal Eval\", \"icon\": \"braces\"},\n {\"name\": \"Combine\", \"icon\": \"merge\"},\n {\"name\": \"Append or Update\", \"icon\": \"circle-plus\"},\n {\"name\": \"Remove Keys\", \"icon\": \"eraser\"},\n {\"name\": \"Rename Keys\", \"icon\": \"pencil-line\"},\n {\"name\": \"Path Selection\", \"icon\": \"mouse-pointer\"},\n {\"name\": \"JQ Expression\", \"icon\": \"terminal\"},\n ],\n real_time_refresh=True,\n limit=1,\n ),\n # select keys inputs\n MessageTextInput(\n name=\"select_keys_input\",\n display_name=\"Select Keys\",\n info=\"List of keys to select from the data. Only top-level keys can be selected.\",\n show=False,\n is_list=True,\n value=[],\n ),\n # update/ Append data inputs\n DictInput(\n name=\"append_update_data\",\n display_name=\"Append or Update\",\n info=\"Data to append or update the existing data with. Only top-level keys are checked.\",\n show=False,\n value={\"key\": \"value\"},\n is_list=True,\n ),\n # remove keys inputs\n MessageTextInput(\n name=\"remove_keys_input\",\n display_name=\"Remove Keys\",\n info=\"List of keys to remove from the data.\",\n show=False,\n is_list=True,\n value=[],\n ),\n # rename keys inputs\n DictInput(\n name=\"rename_keys_input\",\n display_name=\"Rename Keys\",\n info=\"List of keys to rename in the data.\",\n show=False,\n is_list=True,\n value={\"old_key\": \"new_key\"},\n ),\n MultilineInput(\n name=\"mapped_json_display\",\n display_name=\"JSON to Map\",\n info=\"Paste or preview your JSON here to explore its structure and select a path for extraction.\",\n required=False,\n refresh_button=True,\n real_time_refresh=True,\n placeholder=\"Add a JSON example.\",\n show=False,\n ),\n DropdownInput(\n name=\"selected_key\",\n display_name=\"Select Path\",\n options=[],\n required=False,\n dynamic=True,\n show=False,\n value=None,\n ),\n MessageTextInput(\n name=\"query\",\n display_name=\"JQ Expression\",\n info=\"JSON Query to filter the data. Used by Parse JSON operation.\",\n placeholder=\"e.g., .properties.id\",\n show=False,\n ),\n ]\n\n # Default values for operation fields when clearing (match input definitions)\n OPERATION_FIELD_DEFAULTS: dict[str, Any] = {\n \"select_keys_input\": [],\n \"append_update_data\": {\"key\": \"value\"},\n \"remove_keys_input\": [],\n \"rename_keys_input\": {\"old_key\": \"new_key\"},\n \"mapped_json_display\": \"\",\n \"selected_key\": None,\n \"query\": \"\",\n }\n\n outputs = [\n Output(display_name=\"JSON\", name=\"data_output\", method=\"as_data\"),\n ]\n\n # Helper methods for data operations\n def get_data_dict(self) -> dict:\n \"\"\"Extract data dictionary from Data object.\"\"\"\n data = self.data[0] if isinstance(self.data, list) and len(self.data) == 1 else self.data\n return data.model_dump()\n\n def json_query(self) -> Data:\n import json\n\n try:\n import jq\n except ImportError:\n msg = \"jq is required for JQ Expression. Install with: pip install jq\"\n raise ImportError(msg) from None\n\n if not self.query or not self.query.strip():\n msg = \"JSON Query is required and cannot be blank.\"\n raise ValueError(msg)\n raw_data = self.get_data_dict()\n try:\n input_str = json.dumps(raw_data)\n repaired = repair_json(input_str)\n data_json = json.loads(repaired)\n jq_input = data_json[\"data\"] if isinstance(data_json, dict) and \"data\" in data_json else data_json\n validate_jq_program(self.query)\n results = jq.compile(self.query).input(jq_input).all()\n if not results:\n msg = \"No result from JSON query.\"\n raise ValueError(msg)\n result = results[0] if len(results) == 1 else results\n if result is None or result == \"None\":\n msg = \"JSON query returned null/None. Check if the path exists in your data.\"\n raise ValueError(msg)\n if isinstance(result, dict):\n return Data(data=result)\n return Data(data={\"result\": result})\n except (ValueError, TypeError, KeyError, json.JSONDecodeError) as e:\n logger.error(f\"JSON Query failed: {e}\")\n msg = f\"JSON Query error: {e}\"\n raise ValueError(msg) from e\n\n def get_normalized_data(self) -> dict:\n \"\"\"Get normalized data dictionary, handling the 'data' key if present.\"\"\"\n data_dict = self.get_data_dict()\n return data_dict.get(\"data\", data_dict)\n\n def data_is_list(self) -> bool:\n \"\"\"Check if data contains multiple items.\"\"\"\n return isinstance(self.data, list) and len(self.data) > 1\n\n def validate_single_data(self, operation: str) -> None:\n \"\"\"Validate that the operation is being performed on a single data object.\"\"\"\n if self.data_is_list():\n msg = f\"{operation} operation is not supported for multiple data objects.\"\n raise ValueError(msg)\n\n def operation_exception(self, operations: list[str]) -> None:\n \"\"\"Raise exception for incompatible operations.\"\"\"\n msg = f\"{operations} operations are not supported in combination with each other.\"\n raise ValueError(msg)\n\n # Data transformation operations\n def select_keys(self, *, evaluate: bool | None = None) -> Data:\n \"\"\"Select specific keys from the data dictionary.\"\"\"\n self.validate_single_data(\"Select Keys\")\n data_dict = self.get_normalized_data()\n filter_criteria: list[str] = self.select_keys_input\n\n # Filter the data\n if len(filter_criteria) == 1 and filter_criteria[0] == \"data\":\n filtered = data_dict[\"data\"]\n else:\n if not all(key in data_dict for key in filter_criteria):\n msg = f\"Select key not found in data. Available keys: {list(data_dict.keys())}\"\n raise ValueError(msg)\n filtered = {key: value for key, value in data_dict.items() if key in filter_criteria}\n\n # Create a new Data object with the filtered data\n if evaluate:\n filtered = self.recursive_eval(filtered)\n\n # Return a new Data object with the filtered data directly in the data attribute\n return Data(data=filtered)\n\n def remove_keys(self) -> Data:\n \"\"\"Remove specified keys from the data dictionary, recursively.\"\"\"\n self.validate_single_data(\"Remove Keys\")\n data_dict = self.get_normalized_data()\n remove_keys_input: list[str] = self.remove_keys_input\n\n filtered = DataOperationsComponent.remove_keys_recursive(data_dict, set(remove_keys_input))\n return Data(data=filtered)\n\n def rename_keys(self) -> Data:\n \"\"\"Rename keys in the data dictionary, recursively.\"\"\"\n self.validate_single_data(\"Rename Keys\")\n data_dict = self.get_normalized_data()\n rename_keys_input: dict[str, str] = self.rename_keys_input\n\n renamed = DataOperationsComponent.rename_keys_recursive(data_dict, rename_keys_input)\n return Data(data=renamed)\n\n def recursive_eval(self, data: Any) -> Any:\n \"\"\"Recursively evaluate string values in a dictionary or list.\n\n If the value is a string that can be evaluated, it will be evaluated.\n Otherwise, the original value is returned.\n \"\"\"\n if isinstance(data, dict):\n return {k: self.recursive_eval(v) for k, v in data.items()}\n if isinstance(data, list):\n return [self.recursive_eval(item) for item in data]\n if isinstance(data, str):\n try:\n # Only attempt to evaluate strings that look like Python literals\n if (\n data.strip().startswith((\"{\", \"[\", \"(\", \"'\", '\"'))\n or data.strip().lower() in (\"true\", \"false\", \"none\")\n or data.strip().replace(\".\", \"\").isdigit()\n ):\n return ast.literal_eval(data)\n # return data\n except (ValueError, SyntaxError, TypeError, MemoryError):\n # If evaluation fails for any reason, return the original string\n return data\n else:\n return data\n return data\n\n def evaluate_data(self) -> Data:\n \"\"\"Evaluate string values in the data dictionary.\"\"\"\n self.validate_single_data(\"Literal Eval\")\n logger.info(\"evaluating data\")\n return Data(**self.recursive_eval(self.get_data_dict()))\n\n def combine_data(self, *, evaluate: bool | None = None) -> Data:\n \"\"\"Combine multiple data objects into one.\"\"\"\n logger.info(\"combining data\")\n if not self.data_is_list():\n return self.data[0] if self.data else Data(data={})\n\n if len(self.data) == 1:\n msg = \"Combine operation requires multiple data inputs.\"\n raise ValueError(msg)\n\n data_dicts = [data.model_dump().get(\"data\", data.model_dump()) for data in self.data]\n combined_data = {}\n\n for data_dict in data_dicts:\n for key, value in data_dict.items():\n if key not in combined_data:\n combined_data[key] = value\n elif isinstance(combined_data[key], list):\n if isinstance(value, list):\n combined_data[key].extend(value)\n else:\n combined_data[key].append(value)\n else:\n # If current value is not a list, convert it to list and add new value\n combined_data[key] = (\n [combined_data[key], value] if not isinstance(value, list) else [combined_data[key], *value]\n )\n\n if evaluate:\n combined_data = self.recursive_eval(combined_data)\n\n return Data(data=combined_data)\n\n def append_update(self) -> Data:\n \"\"\"Append or Update with new key-value pairs.\"\"\"\n self.validate_single_data(\"Append or Update\")\n data_filtered = self.get_normalized_data()\n\n for key, value in self.append_update_data.items():\n data_filtered[key] = value\n\n return Data(data=data_filtered)\n\n # Configuration and execution methods\n def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:\n if field_name == \"operations\":\n build_config[\"operations\"][\"value\"] = field_value\n # Mirror Text Operations: first hide all operation-specific fields and clear their values\n for field in self.ALL_OPERATION_FIELDS:\n if field in build_config:\n build_config[field][\"show\"] = False\n if field in self.OPERATION_FIELD_DEFAULTS:\n build_config[field][\"value\"] = self.OPERATION_FIELD_DEFAULTS[field]\n\n selected_actions = [\n action[\"name\"] for action in (field_value or []) if isinstance(action, dict) and \"name\" in action\n ]\n if len(selected_actions) == 1 and selected_actions[0] in ACTION_CONFIG:\n action = selected_actions[0]\n config = ACTION_CONFIG[action]\n build_config[\"data\"][\"is_list\"] = config[\"is_list\"]\n logger.info(config[\"log_msg\"])\n return set_current_fields(\n build_config=build_config,\n action_fields=self.actions_data,\n selected_action=action,\n default_fields=[\"operations\", \"data\"],\n func=set_field_display,\n )\n return build_config\n\n if field_name == \"mapped_json_display\":\n try:\n parsed_json = json.loads(field_value)\n keys = DataOperationsComponent.extract_all_paths(parsed_json)\n build_config[\"selected_key\"][\"options\"] = keys\n build_config[\"selected_key\"][\"show\"] = True\n except (json.JSONDecodeError, TypeError, ValueError) as e:\n logger.error(f\"Error parsing mapped JSON: {e}\")\n build_config[\"selected_key\"][\"show\"] = False\n\n return build_config\n\n def json_path(self) -> Data:\n try:\n import jq\n except ImportError:\n msg = \"jq is required for Path Selection. Install with: pip install jq\"\n raise ImportError(msg) from None\n\n try:\n if not self.data or not self.selected_key:\n msg = \"Missing input data or selected key.\"\n raise ValueError(msg)\n input_payload = self.data[0].data if isinstance(self.data, list) else self.data.data\n validate_jq_program(self.selected_key)\n compiled = jq.compile(self.selected_key)\n result = compiled.input(input_payload).first()\n if isinstance(result, dict):\n return Data(data=result)\n return Data(data={\"result\": result})\n except (ValueError, TypeError, KeyError) as e:\n self.status = f\"Error: {e!s}\"\n self.log(self.status)\n return Data(data={\"error\": str(e)})\n\n def as_data(self) -> Data:\n if not hasattr(self, \"operations\") or not self.operations:\n return Data(data={})\n\n selected_actions = [action[\"name\"] for action in self.operations]\n logger.info(f\"selected_actions: {selected_actions}\")\n if len(selected_actions) != 1:\n return Data(data={})\n\n action_map: dict[str, Callable[[], Data]] = {\n \"Select Keys\": self.select_keys,\n \"Literal Eval\": self.evaluate_data,\n \"Combine\": self.combine_data,\n \"Append or Update\": self.append_update,\n \"Remove Keys\": self.remove_keys,\n \"Rename Keys\": self.rename_keys,\n \"Path Selection\": self.json_path,\n \"JQ Expression\": self.json_query,\n }\n action_name = selected_actions[0]\n handler: Callable[[], Data] | None = action_map.get(action_name)\n if handler is None:\n # Fail fast instead of silently returning empty data. Persisted flows\n # may still reference a removed operation (e.g. \"Filter Values\").\n msg = (\n f\"The '{action_name}' operation is no longer supported by the JSON Operations component. \"\n \"Update this flow to use the Operations component.\"\n )\n raise ValueError(msg)\n try:\n return handler()\n except Exception as e:\n logger.error(f\"Error executing {action_name}: {e!s}\")\n raise\n" + "value": "import ast\nimport json\nfrom typing import TYPE_CHECKING, Any\n\nfrom json_repair import repair_json\n\nfrom lfx.custom import Component\nfrom lfx.inputs import DictInput, DropdownInput, MessageTextInput, SortableListInput\nfrom lfx.io import DataInput, MultilineInput, Output\nfrom lfx.log.logger import logger\nfrom lfx.schema import Data\nfrom lfx.schema.dotdict import dotdict\nfrom lfx.utils.component_utils import set_current_fields, set_field_display\nfrom lfx.utils.jq_security import validate_jq_program\n\nif TYPE_CHECKING:\n from collections.abc import Callable\n\nACTION_CONFIG = {\n \"Select Keys\": {\"is_list\": False, \"log_msg\": \"setting filter fields\"},\n \"Literal Eval\": {\"is_list\": False, \"log_msg\": \"setting evaluate fields\"},\n \"Combine\": {\"is_list\": True, \"log_msg\": \"setting combine fields\"},\n \"Append or Update\": {\"is_list\": False, \"log_msg\": \"setting Append or Update fields\"},\n \"Remove Keys\": {\"is_list\": False, \"log_msg\": \"setting remove keys fields\"},\n \"Rename Keys\": {\"is_list\": False, \"log_msg\": \"setting rename keys fields\"},\n \"Path Selection\": {\"is_list\": False, \"log_msg\": \"setting mapped key extractor fields\"},\n \"JQ Expression\": {\"is_list\": False, \"log_msg\": \"setting parse json fields\"},\n}\n\n\nclass DataOperationsComponent(Component):\n display_name = \"JSON Operations\"\n description = \"Perform various operations on a JSON object.\"\n icon = \"file-json\"\n name = \"DataOperations\"\n legacy = True\n replacement = [\"processing.Operations\"]\n default_keys = [\"operations\", \"data\"]\n metadata = {\n \"keywords\": [\n \"data\",\n \"json\",\n \"operations\",\n \"Append or Update\",\n \"remove keys\",\n \"rename keys\",\n \"select keys\",\n \"literal eval\",\n \"combine\",\n \"append\",\n \"update\",\n \"remove\",\n \"rename\",\n \"data operations\",\n \"json operations\",\n \"data manipulation\",\n \"data transformation\",\n \"data filtering\",\n \"data selection\",\n \"data combination\",\n \"Parse JSON\",\n \"JSON Query\",\n \"JQ Query\",\n ],\n }\n actions_data = {\n \"Select Keys\": [\"select_keys_input\", \"operations\"],\n \"Literal Eval\": [],\n \"Combine\": [],\n \"Append or Update\": [\"append_update_data\", \"operations\"],\n \"Remove Keys\": [\"remove_keys_input\", \"operations\"],\n \"Rename Keys\": [\"rename_keys_input\", \"operations\"],\n \"Path Selection\": [\"mapped_json_display\", \"selected_key\", \"operations\"],\n \"JQ Expression\": [\"query\", \"operations\"],\n }\n\n # All operation-specific input fields (used to hide and reset when no operation selected).\n ALL_OPERATION_FIELDS = [\n \"select_keys_input\",\n \"append_update_data\",\n \"remove_keys_input\",\n \"rename_keys_input\",\n \"mapped_json_display\",\n \"selected_key\",\n \"query\",\n ]\n\n @staticmethod\n def extract_all_paths(obj, path=\"\"):\n paths = []\n if isinstance(obj, dict):\n for k, v in obj.items():\n new_path = f\"{path}.{k}\" if path else f\".{k}\"\n paths.append(new_path)\n paths.extend(DataOperationsComponent.extract_all_paths(v, new_path))\n elif isinstance(obj, list) and obj:\n new_path = f\"{path}[0]\"\n paths.append(new_path)\n paths.extend(DataOperationsComponent.extract_all_paths(obj[0], new_path))\n return paths\n\n @staticmethod\n def remove_keys_recursive(obj, keys_to_remove):\n if isinstance(obj, dict):\n return {\n k: DataOperationsComponent.remove_keys_recursive(v, keys_to_remove)\n for k, v in obj.items()\n if k not in keys_to_remove\n }\n if isinstance(obj, list):\n return [DataOperationsComponent.remove_keys_recursive(item, keys_to_remove) for item in obj]\n return obj\n\n @staticmethod\n def rename_keys_recursive(obj, rename_map):\n if isinstance(obj, dict):\n return {\n rename_map.get(k, k): DataOperationsComponent.rename_keys_recursive(v, rename_map)\n for k, v in obj.items()\n }\n if isinstance(obj, list):\n return [DataOperationsComponent.rename_keys_recursive(item, rename_map) for item in obj]\n return obj\n\n inputs = [\n DataInput(name=\"data\", display_name=\"JSON\", info=\"Data object to filter.\", required=True, is_list=True),\n SortableListInput(\n name=\"operations\",\n display_name=\"Operations\",\n placeholder=\"Select Operation\",\n info=\"List of operations to perform on the data.\",\n options=[\n {\"name\": \"Select Keys\", \"icon\": \"lasso-select\"},\n {\"name\": \"Literal Eval\", \"icon\": \"braces\"},\n {\"name\": \"Combine\", \"icon\": \"merge\"},\n {\"name\": \"Append or Update\", \"icon\": \"circle-plus\"},\n {\"name\": \"Remove Keys\", \"icon\": \"eraser\"},\n {\"name\": \"Rename Keys\", \"icon\": \"pencil-line\"},\n {\"name\": \"Path Selection\", \"icon\": \"mouse-pointer\"},\n {\"name\": \"JQ Expression\", \"icon\": \"terminal\"},\n ],\n real_time_refresh=True,\n limit=1,\n ),\n # select keys inputs\n MessageTextInput(\n name=\"select_keys_input\",\n display_name=\"Select Keys\",\n info=\"List of keys to select from the data. Only top-level keys can be selected.\",\n show=False,\n is_list=True,\n value=[],\n ),\n # update/ Append data inputs\n DictInput(\n name=\"append_update_data\",\n display_name=\"Append or Update\",\n info=\"Data to append or update the existing data with. Only top-level keys are checked.\",\n show=False,\n value={\"key\": \"value\"},\n is_list=True,\n ),\n # remove keys inputs\n MessageTextInput(\n name=\"remove_keys_input\",\n display_name=\"Remove Keys\",\n info=\"List of keys to remove from the data.\",\n show=False,\n is_list=True,\n value=[],\n ),\n # rename keys inputs\n DictInput(\n name=\"rename_keys_input\",\n display_name=\"Rename Keys\",\n info=\"List of keys to rename in the data.\",\n show=False,\n is_list=True,\n value={\"old_key\": \"new_key\"},\n ),\n MultilineInput(\n name=\"mapped_json_display\",\n display_name=\"JSON to Map\",\n info=\"Paste or preview your JSON here to explore its structure and select a path for extraction.\",\n required=False,\n refresh_button=True,\n real_time_refresh=True,\n placeholder=\"Add a JSON example.\",\n show=False,\n ),\n DropdownInput(\n name=\"selected_key\",\n display_name=\"Select Path\",\n options=[],\n required=False,\n dynamic=True,\n show=False,\n value=None,\n ),\n MessageTextInput(\n name=\"query\",\n display_name=\"JQ Expression\",\n info=\"JSON Query to filter the data. Used by Parse JSON operation.\",\n placeholder=\"e.g., .properties.id\",\n show=False,\n ),\n ]\n\n # Default values for operation fields when clearing (match input definitions)\n OPERATION_FIELD_DEFAULTS: dict[str, Any] = {\n \"select_keys_input\": [],\n \"append_update_data\": {\"key\": \"value\"},\n \"remove_keys_input\": [],\n \"rename_keys_input\": {\"old_key\": \"new_key\"},\n \"mapped_json_display\": \"\",\n \"selected_key\": None,\n \"query\": \"\",\n }\n\n outputs = [\n Output(display_name=\"JSON\", name=\"data_output\", method=\"as_data\"),\n ]\n\n # Helper methods for data operations\n def get_data_dict(self) -> dict:\n \"\"\"Extract data dictionary from Data object.\"\"\"\n data = self.data[0] if isinstance(self.data, list) and len(self.data) == 1 else self.data\n return data.model_dump()\n\n def json_query(self) -> Data:\n import json\n\n try:\n import jq\n except ImportError:\n msg = \"jq is required for JQ Expression. Install with: pip install jq\"\n raise ImportError(msg) from None\n\n if not self.query or not self.query.strip():\n msg = \"JSON Query is required and cannot be blank.\"\n raise ValueError(msg)\n raw_data = self.get_data_dict()\n try:\n input_str = json.dumps(raw_data)\n repaired = repair_json(input_str)\n data_json = json.loads(repaired)\n jq_input = data_json[\"data\"] if isinstance(data_json, dict) and \"data\" in data_json else data_json\n validate_jq_program(self.query)\n results = jq.compile(self.query).input(jq_input).all()\n if not results:\n msg = \"No result from JSON query.\"\n raise ValueError(msg)\n result = results[0] if len(results) == 1 else results\n if result is None or result == \"None\":\n msg = \"JSON query returned null/None. Check if the path exists in your data.\"\n raise ValueError(msg)\n if isinstance(result, dict):\n return Data(data=result)\n return Data(data={\"result\": result})\n except (ValueError, TypeError, KeyError, json.JSONDecodeError) as e:\n logger.error(f\"JSON Query failed: {e}\")\n msg = f\"JSON Query error: {e}\"\n raise ValueError(msg) from e\n\n def get_normalized_data(self) -> dict:\n \"\"\"Get normalized data dictionary, handling the 'data' key if present.\"\"\"\n data_dict = self.get_data_dict()\n return data_dict.get(\"data\", data_dict)\n\n def data_is_list(self) -> bool:\n \"\"\"Check if data contains multiple items.\"\"\"\n return isinstance(self.data, list) and len(self.data) > 1\n\n def validate_single_data(self, operation: str) -> None:\n \"\"\"Validate that the operation is being performed on a single data object.\"\"\"\n if self.data_is_list():\n msg = f\"{operation} operation is not supported for multiple data objects.\"\n raise ValueError(msg)\n\n def operation_exception(self, operations: list[str]) -> None:\n \"\"\"Raise exception for incompatible operations.\"\"\"\n msg = f\"{operations} operations are not supported in combination with each other.\"\n raise ValueError(msg)\n\n # Data transformation operations\n def select_keys(self, *, evaluate: bool | None = None) -> Data:\n \"\"\"Select specific keys from the data dictionary.\"\"\"\n self.validate_single_data(\"Select Keys\")\n data_dict = self.get_normalized_data()\n filter_criteria: list[str] = self.select_keys_input\n\n # Filter the data\n if len(filter_criteria) == 1 and filter_criteria[0] == \"data\":\n filtered = data_dict[\"data\"]\n else:\n if not all(key in data_dict for key in filter_criteria):\n msg = f\"Select key not found in data. Available keys: {list(data_dict.keys())}\"\n raise ValueError(msg)\n filtered = {key: value for key, value in data_dict.items() if key in filter_criteria}\n\n # Create a new Data object with the filtered data\n if evaluate:\n filtered = self.recursive_eval(filtered)\n\n # Return a new Data object with the filtered data directly in the data attribute\n return Data(data=filtered)\n\n def remove_keys(self) -> Data:\n \"\"\"Remove specified keys from the data dictionary, recursively.\"\"\"\n self.validate_single_data(\"Remove Keys\")\n data_dict = self.get_normalized_data()\n remove_keys_input: list[str] = self.remove_keys_input\n\n filtered = DataOperationsComponent.remove_keys_recursive(data_dict, set(remove_keys_input))\n return Data(data=filtered)\n\n def rename_keys(self) -> Data:\n \"\"\"Rename keys in the data dictionary, recursively.\"\"\"\n self.validate_single_data(\"Rename Keys\")\n data_dict = self.get_normalized_data()\n rename_keys_input: dict[str, str] = self.rename_keys_input\n\n renamed = DataOperationsComponent.rename_keys_recursive(data_dict, rename_keys_input)\n return Data(data=renamed)\n\n def recursive_eval(self, data: Any) -> Any:\n \"\"\"Recursively evaluate string values in a dictionary or list.\n\n If the value is a string that can be evaluated, it will be evaluated.\n Otherwise, the original value is returned.\n \"\"\"\n if isinstance(data, dict):\n return {k: self.recursive_eval(v) for k, v in data.items()}\n if isinstance(data, list):\n return [self.recursive_eval(item) for item in data]\n if isinstance(data, str):\n try:\n # Only attempt to evaluate strings that look like Python literals\n if (\n data.strip().startswith((\"{\", \"[\", \"(\", \"'\", '\"'))\n or data.strip().lower() in (\"true\", \"false\", \"none\")\n or data.strip().replace(\".\", \"\").isdigit()\n ):\n return ast.literal_eval(data)\n # return data\n except (ValueError, SyntaxError, TypeError, MemoryError):\n # If evaluation fails for any reason, return the original string\n return data\n else:\n return data\n return data\n\n def evaluate_data(self) -> Data:\n \"\"\"Evaluate string values in the data dictionary.\"\"\"\n self.validate_single_data(\"Literal Eval\")\n logger.info(\"evaluating data\")\n return Data(**self.recursive_eval(self.get_data_dict()))\n\n def combine_data(self, *, evaluate: bool | None = None) -> Data:\n \"\"\"Combine multiple data objects into one.\"\"\"\n logger.info(\"combining data\")\n if not self.data_is_list():\n return self.data[0] if self.data else Data(data={})\n\n if len(self.data) == 1:\n msg = \"Combine operation requires multiple data inputs.\"\n raise ValueError(msg)\n\n data_dicts = [data.model_dump().get(\"data\", data.model_dump()) for data in self.data]\n combined_data = {}\n\n for data_dict in data_dicts:\n for key, value in data_dict.items():\n if key not in combined_data:\n combined_data[key] = value\n elif isinstance(combined_data[key], list):\n if isinstance(value, list):\n combined_data[key].extend(value)\n else:\n combined_data[key].append(value)\n else:\n # If current value is not a list, convert it to list and add new value\n combined_data[key] = (\n [combined_data[key], value] if not isinstance(value, list) else [combined_data[key], *value]\n )\n\n if evaluate:\n combined_data = self.recursive_eval(combined_data)\n\n return Data(data=combined_data)\n\n def append_update(self) -> Data:\n \"\"\"Append or Update with new key-value pairs.\"\"\"\n self.validate_single_data(\"Append or Update\")\n data_filtered = self.get_normalized_data()\n\n for key, value in self.append_update_data.items():\n data_filtered[key] = value\n\n return Data(data=data_filtered)\n\n # Configuration and execution methods\n def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:\n if field_name == \"operations\":\n build_config[\"operations\"][\"value\"] = field_value\n # Mirror Text Operations: first hide all operation-specific fields and clear their values\n for field in self.ALL_OPERATION_FIELDS:\n if field in build_config:\n build_config[field][\"show\"] = False\n if field in self.OPERATION_FIELD_DEFAULTS:\n build_config[field][\"value\"] = self.OPERATION_FIELD_DEFAULTS[field]\n\n selected_actions = [\n action[\"name\"] for action in (field_value or []) if isinstance(action, dict) and \"name\" in action\n ]\n if len(selected_actions) == 1 and selected_actions[0] in ACTION_CONFIG:\n action = selected_actions[0]\n config = ACTION_CONFIG[action]\n build_config[\"data\"][\"is_list\"] = config[\"is_list\"]\n logger.info(config[\"log_msg\"])\n return set_current_fields(\n build_config=build_config,\n action_fields=self.actions_data,\n selected_action=action,\n default_fields=[\"operations\", \"data\"],\n func=set_field_display,\n )\n return build_config\n\n if field_name == \"mapped_json_display\":\n try:\n parsed_json = json.loads(field_value)\n keys = DataOperationsComponent.extract_all_paths(parsed_json)\n build_config[\"selected_key\"][\"options\"] = keys\n build_config[\"selected_key\"][\"show\"] = True\n except (json.JSONDecodeError, TypeError, ValueError) as e:\n logger.error(f\"Error parsing mapped JSON: {e}\")\n build_config[\"selected_key\"][\"show\"] = False\n\n return build_config\n\n async def update_frontend_node(self, new_frontend_node: dict, current_frontend_node: dict) -> dict:\n \"\"\"Keep the saved operation's fields visible when the node is updated.\n\n The rebuilt template carries class-default ``show`` flags, which the generic merge does not\n restore, so visibility is re-derived here from the saved operation without touching saved values.\n \"\"\"\n await super().update_frontend_node(new_frontend_node, current_frontend_node)\n\n template = new_frontend_node.get(\"template\", {})\n operations_value = template.get(\"operations\", {}).get(\"value\") or []\n selected_actions = [\n action[\"name\"]\n for action in (operations_value if isinstance(operations_value, list) else [])\n if isinstance(action, dict) and \"name\" in action\n ]\n action = selected_actions[0] if len(selected_actions) == 1 else None\n if action not in ACTION_CONFIG:\n action = None\n\n # Re-apply visibility. Unlike update_build_config, keep saved values.\n visible_fields = self.actions_data.get(action, []) if action else []\n for field in self.ALL_OPERATION_FIELDS:\n if field in template:\n template[field][\"show\"] = field in visible_fields\n if action and \"data\" in template:\n template[\"data\"][\"is_list\"] = ACTION_CONFIG[action][\"is_list\"]\n\n # The merge restores only values, so rebuild Path Selection's dropdown from the saved JSON.\n if action == \"Path Selection\" and \"selected_key\" in template:\n mapped_json = template.get(\"mapped_json_display\", {}).get(\"value\")\n if mapped_json:\n try:\n template[\"selected_key\"][\"options\"] = self.extract_all_paths(json.loads(mapped_json))\n except (json.JSONDecodeError, TypeError, ValueError, RecursionError) as e:\n logger.warning(f\"Error parsing saved mapped JSON: {e}\")\n template[\"selected_key\"][\"show\"] = False\n\n return new_frontend_node\n\n def json_path(self) -> Data:\n try:\n import jq\n except ImportError:\n msg = \"jq is required for Path Selection. Install with: pip install jq\"\n raise ImportError(msg) from None\n\n try:\n if not self.data or not self.selected_key:\n msg = \"Missing input data or selected key.\"\n raise ValueError(msg)\n input_payload = self.data[0].data if isinstance(self.data, list) else self.data.data\n validate_jq_program(self.selected_key)\n compiled = jq.compile(self.selected_key)\n result = compiled.input(input_payload).first()\n if isinstance(result, dict):\n return Data(data=result)\n return Data(data={\"result\": result})\n except (ValueError, TypeError, KeyError) as e:\n self.status = f\"Error: {e!s}\"\n self.log(self.status)\n return Data(data={\"error\": str(e)})\n\n def as_data(self) -> Data:\n if not hasattr(self, \"operations\") or not self.operations:\n return Data(data={})\n\n selected_actions = [action[\"name\"] for action in self.operations]\n logger.info(f\"selected_actions: {selected_actions}\")\n if len(selected_actions) != 1:\n return Data(data={})\n\n action_map: dict[str, Callable[[], Data]] = {\n \"Select Keys\": self.select_keys,\n \"Literal Eval\": self.evaluate_data,\n \"Combine\": self.combine_data,\n \"Append or Update\": self.append_update,\n \"Remove Keys\": self.remove_keys,\n \"Rename Keys\": self.rename_keys,\n \"Path Selection\": self.json_path,\n \"JQ Expression\": self.json_query,\n }\n action_name = selected_actions[0]\n handler: Callable[[], Data] | None = action_map.get(action_name)\n if handler is None:\n # Fail fast instead of silently returning empty data. Persisted flows\n # may still reference a removed operation (e.g. \"Filter Values\").\n msg = (\n f\"The '{action_name}' operation is no longer supported by the JSON Operations component. \"\n \"Update this flow to use the Operations component.\"\n )\n raise ValueError(msg)\n try:\n return handler()\n except Exception as e:\n logger.error(f\"Error executing {action_name}: {e!s}\")\n raise\n" }, "data": { "_input_type": "JSONInput", @@ -32518,6 +32518,6 @@ "num_components": 130, "num_modules": 17 }, - "sha256": "37c7c46e7752936ec4a4f3ced88e1ec7df3db7fadcbf4171d320731b7beb8711", + "sha256": "8ff6a69241f4a1db537c4695ed91c65d587641965765b8a260affaf71f4287a5", "version": "1.13.0" } diff --git a/src/lfx/src/lfx/components/processing/data_operations.py b/src/lfx/src/lfx/components/processing/data_operations.py index b73006935e04..e8a7fb1c7c1e 100644 --- a/src/lfx/src/lfx/components/processing/data_operations.py +++ b/src/lfx/src/lfx/components/processing/data_operations.py @@ -439,6 +439,45 @@ def update_build_config(self, build_config: dotdict, field_value: Any, field_nam return build_config + async def update_frontend_node(self, new_frontend_node: dict, current_frontend_node: dict) -> dict: + """Keep the saved operation's fields visible when the node is updated. + + The rebuilt template carries class-default ``show`` flags, which the generic merge does not + restore, so visibility is re-derived here from the saved operation without touching saved values. + """ + await super().update_frontend_node(new_frontend_node, current_frontend_node) + + template = new_frontend_node.get("template", {}) + operations_value = template.get("operations", {}).get("value") or [] + selected_actions = [ + action["name"] + for action in (operations_value if isinstance(operations_value, list) else []) + if isinstance(action, dict) and "name" in action + ] + action = selected_actions[0] if len(selected_actions) == 1 else None + if action not in ACTION_CONFIG: + action = None + + # Re-apply visibility. Unlike update_build_config, keep saved values. + visible_fields = self.actions_data.get(action, []) if action else [] + for field in self.ALL_OPERATION_FIELDS: + if field in template: + template[field]["show"] = field in visible_fields + if action and "data" in template: + template["data"]["is_list"] = ACTION_CONFIG[action]["is_list"] + + # The merge restores only values, so rebuild Path Selection's dropdown from the saved JSON. + if action == "Path Selection" and "selected_key" in template: + mapped_json = template.get("mapped_json_display", {}).get("value") + if mapped_json: + try: + template["selected_key"]["options"] = self.extract_all_paths(json.loads(mapped_json)) + except (json.JSONDecodeError, TypeError, ValueError, RecursionError) as e: + logger.warning(f"Error parsing saved mapped JSON: {e}") + template["selected_key"]["show"] = False + + return new_frontend_node + def json_path(self) -> Data: try: import jq