| from typing import Any |
|
|
| from langflow.custom import Component |
| from langflow.io import DataInput, DropdownInput, MessageTextInput, Output |
| from langflow.schema import Data, dotdict |
|
|
|
|
| class DataConditionalRouterComponent(Component): |
| display_name = "Condition" |
| description = "Route Data object(s) based on a condition applied to a specified key, including boolean validation." |
| icon = "split" |
| name = "DataConditionalRouter" |
| legacy = True |
|
|
| inputs = [ |
| DataInput( |
| name="data_input", |
| display_name="Data Input", |
| info="The Data object or list of Data objects to process", |
| is_list=True, |
| ), |
| MessageTextInput( |
| name="key_name", |
| display_name="Key Name", |
| info="The name of the key in the Data object(s) to check", |
| ), |
| DropdownInput( |
| name="operator", |
| display_name="Operator", |
| options=["equals", "not equals", "contains", "starts with", "ends with", "boolean validator"], |
| info="The operator to apply for comparing the values. 'boolean validator' treats the value as a boolean.", |
| value="equals", |
| ), |
| MessageTextInput( |
| name="compare_value", |
| display_name="Match Text", |
| info="The value to compare against (not used for boolean validator)", |
| ), |
| ] |
|
|
| outputs = [ |
| Output(display_name="True Output", name="true_output", method="process_data"), |
| Output(display_name="False Output", name="false_output", method="process_data"), |
| ] |
|
|
| def compare_values(self, item_value: str, compare_value: str, operator: str) -> bool: |
| if operator == "equals": |
| return item_value == compare_value |
| if operator == "not equals": |
| return item_value != compare_value |
| if operator == "contains": |
| return compare_value in item_value |
| if operator == "starts with": |
| return item_value.startswith(compare_value) |
| if operator == "ends with": |
| return item_value.endswith(compare_value) |
| if operator == "boolean validator": |
| return self.parse_boolean(item_value) |
| return False |
|
|
| def parse_boolean(self, value): |
| if isinstance(value, bool): |
| return value |
| if isinstance(value, str): |
| return value.lower() in {"true", "1", "yes", "y", "on"} |
| return bool(value) |
|
|
| def validate_input(self, data_item: Data) -> bool: |
| if not isinstance(data_item, Data): |
| self.status = "Input is not a Data object" |
| return False |
| if self.key_name not in data_item.data: |
| self.status = f"Key '{self.key_name}' not found in Data" |
| return False |
| return True |
|
|
| def process_data(self) -> Data | list[Data]: |
| if isinstance(self.data_input, list): |
| true_output = [] |
| false_output = [] |
| for item in self.data_input: |
| if self.validate_input(item): |
| result = self.process_single_data(item) |
| if result: |
| true_output.append(item) |
| else: |
| false_output.append(item) |
| self.stop("false_output" if true_output else "true_output") |
| return true_output or false_output |
| if not self.validate_input(self.data_input): |
| return Data(data={"error": self.status}) |
| result = self.process_single_data(self.data_input) |
| self.stop("false_output" if result else "true_output") |
| return self.data_input |
|
|
| def process_single_data(self, data_item: Data) -> bool: |
| item_value = data_item.data[self.key_name] |
| operator = self.operator |
|
|
| if operator == "boolean validator": |
| condition_met = self.parse_boolean(item_value) |
| condition_description = f"Boolean validation of '{self.key_name}'" |
| else: |
| compare_value = self.compare_value |
| condition_met = self.compare_values(str(item_value), compare_value, operator) |
| condition_description = f"{self.key_name} {operator} {compare_value}" |
|
|
| if condition_met: |
| self.status = f"Condition met: {condition_description}" |
| return True |
| self.status = f"Condition not met: {condition_description}" |
| return False |
|
|
| def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None): |
| if field_name == "operator": |
| if field_value == "boolean validator": |
| build_config["compare_value"]["show"] = False |
| build_config["compare_value"]["advanced"] = True |
| build_config["compare_value"]["value"] = None |
| else: |
| build_config["compare_value"]["show"] = True |
| build_config["compare_value"]["advanced"] = False |
|
|
| return build_config |
|
|
