From 63cb3baca4491dde7d2e2135f48df43c13eb8394 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Sat, 22 Jun 2024 18:00:10 -0300 Subject: [PATCH] refactor: move prompt utils to template module --- src/backend/base/langflow/api/utils.py | 82 -------------------- src/backend/base/langflow/template/utils.py | 84 +++++++++++++++++++++ 2 files changed, 84 insertions(+), 82 deletions(-) create mode 100644 src/backend/base/langflow/template/utils.py diff --git a/src/backend/base/langflow/api/utils.py b/src/backend/base/langflow/api/utils.py index 99d4d3681..704b42c84 100644 --- a/src/backend/base/langflow/api/utils.py +++ b/src/backend/base/langflow/api/utils.py @@ -69,87 +69,6 @@ def build_input_keys_response(langchain_object, artifacts): return input_keys_response -def update_frontend_node_with_template_values(frontend_node, raw_frontend_node): - """ - Updates the given frontend node with values from the raw template data. - - :param frontend_node: A dict representing a built frontend node. - :param raw_template_data: A dict representing raw template data. - :return: Updated frontend node. - """ - if not is_valid_data(frontend_node, raw_frontend_node): - return frontend_node - - # Check if the display_name is different than "CustomComponent" - # if so, update the display_name in the frontend_node - if raw_frontend_node["display_name"] != "CustomComponent": - frontend_node["display_name"] = raw_frontend_node["display_name"] - - update_template_values(frontend_node["template"], raw_frontend_node["template"]) - - old_code = raw_frontend_node["template"]["code"]["value"] - new_code = frontend_node["template"]["code"]["value"] - frontend_node["edited"] = old_code != new_code - - return frontend_node - - -def raw_frontend_data_is_valid(raw_frontend_data): - """Check if the raw frontend data is valid for processing.""" - return "template" in raw_frontend_data and "display_name" in raw_frontend_data - - -def is_valid_data(frontend_node, raw_frontend_data): - """Check if the data is valid for processing.""" - - return frontend_node and "template" in frontend_node and raw_frontend_data_is_valid(raw_frontend_data) - - -def update_template_values(frontend_template, raw_template): - """Updates the frontend template with values from the raw template.""" - for key, value_dict in raw_template.items(): - if key == "code" or not isinstance(value_dict, dict): - continue - - update_template_field(frontend_template, key, value_dict) - - -def update_template_field(frontend_template, key, value_dict): - """Updates a specific field in the frontend template.""" - template_field = frontend_template.get(key) - if not template_field or template_field.get("type") != value_dict.get("type"): - return - - if "value" in value_dict and value_dict["value"]: - template_field["value"] = value_dict["value"] - - if "file_path" in value_dict and value_dict["file_path"]: - file_path_value = get_file_path_value(value_dict["file_path"]) - if not file_path_value: - # If the file does not exist, remove the value from the template_field["value"] - template_field["value"] = "" - template_field["file_path"] = file_path_value - - -def get_file_path_value(file_path): - """Get the file path value if the file exists, else return empty string.""" - try: - path = Path(file_path) - except TypeError: - return "" - - # Check for safety - # If the path is not in the cache dir, return empty string - # This is to prevent access to files outside the cache dir - # If the path is not a file, return empty string - if not str(path).startswith(user_cache_dir("langflow", "langflow")): - return "" - - if not path.exists(): - return "" - return file_path - - def validate_is_component(flows: list["Flow"]): for flow in flows: if not flow.data or flow.is_component is not None: @@ -326,4 +245,3 @@ def parse_exception(exc): if hasattr(exc, "body"): return exc.body["message"] return str(exc) - return str(exc) diff --git a/src/backend/base/langflow/template/utils.py b/src/backend/base/langflow/template/utils.py new file mode 100644 index 000000000..b436b8987 --- /dev/null +++ b/src/backend/base/langflow/template/utils.py @@ -0,0 +1,84 @@ +from pathlib import Path + +from platformdirs import user_cache_dir + + +def raw_frontend_data_is_valid(raw_frontend_data): + """Check if the raw frontend data is valid for processing.""" + return "template" in raw_frontend_data and "display_name" in raw_frontend_data + + +def get_file_path_value(file_path): + """Get the file path value if the file exists, else return empty string.""" + try: + path = Path(file_path) + except TypeError: + return "" + + # Check for safety + # If the path is not in the cache dir, return empty string + # This is to prevent access to files outside the cache dir + # If the path is not a file, return empty string + if not str(path).startswith(user_cache_dir("langflow", "langflow")): + return "" + + if not path.exists(): + return "" + return file_path + + +def update_template_field(frontend_template, key, value_dict): + """Updates a specific field in the frontend template.""" + template_field = frontend_template.get(key) + if not template_field or template_field.get("type") != value_dict.get("type"): + return + + if "value" in value_dict and value_dict["value"]: + template_field["value"] = value_dict["value"] + + if "file_path" in value_dict and value_dict["file_path"]: + file_path_value = get_file_path_value(value_dict["file_path"]) + if not file_path_value: + # If the file does not exist, remove the value from the template_field["value"] + template_field["value"] = "" + template_field["file_path"] = file_path_value + + +def is_valid_data(frontend_node, raw_frontend_data): + """Check if the data is valid for processing.""" + + return frontend_node and "template" in frontend_node and raw_frontend_data_is_valid(raw_frontend_data) + + +def update_template_values(frontend_template, raw_template): + """Updates the frontend template with values from the raw template.""" + for key, value_dict in raw_template.items(): + if key == "code" or not isinstance(value_dict, dict): + continue + + update_template_field(frontend_template, key, value_dict) + + +def update_frontend_node_with_template_values(frontend_node, raw_frontend_node): + """ + Updates the given frontend node with values from the raw template data. + + :param frontend_node: A dict representing a built frontend node. + :param raw_template_data: A dict representing raw template data. + :return: Updated frontend node. + """ + if not is_valid_data(frontend_node, raw_frontend_node): + return frontend_node + + # Check if the display_name is different than "CustomComponent" + # if so, update the display_name in the frontend_node + if raw_frontend_node["display_name"] != "CustomComponent": + frontend_node["display_name"] = raw_frontend_node["display_name"] + + update_template_values(frontend_node["template"], raw_frontend_node["template"]) + + old_code = raw_frontend_node["template"]["code"]["value"] + new_code = frontend_node["template"]["code"]["value"] + frontend_node["edited"] = old_code != new_code + + return frontend_node