| from typing import Any |
|
|
| from langflow.custom import Component |
| from langflow.field_typing.range_spec import RangeSpec |
| from langflow.inputs.inputs import BoolInput, DictInput, IntInput, MessageTextInput |
| from langflow.io import Output |
| from langflow.schema import Data |
| from langflow.schema.dotdict import dotdict |
|
|
|
|
| class CreateDataComponent(Component): |
| display_name: str = "Create Data" |
| description: str = "Dynamically create a Data with a specified number of fields." |
| name: str = "CreateData" |
| MAX_FIELDS = 15 |
| legacy = True |
| icon = "ListFilter" |
|
|
| inputs = [ |
| IntInput( |
| name="number_of_fields", |
| display_name="Number of Fields", |
| info="Number of fields to be added to the record.", |
| real_time_refresh=True, |
| value=1, |
| range_spec=RangeSpec(min=1, max=MAX_FIELDS, step=1, step_type="int"), |
| ), |
| MessageTextInput( |
| name="text_key", |
| display_name="Text Key", |
| info="Key that identifies the field to be used as the text content.", |
| advanced=True, |
| ), |
| BoolInput( |
| name="text_key_validator", |
| display_name="Text Key Validator", |
| advanced=True, |
| info="If enabled, checks if the given 'Text Key' is present in the given 'Data'.", |
| ), |
| ] |
|
|
| outputs = [ |
| Output(display_name="Data", name="data", method="build_data"), |
| ] |
|
|
| def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None): |
| if field_name == "number_of_fields": |
| default_keys = ["code", "_type", "number_of_fields", "text_key", "text_key_validator"] |
| try: |
| field_value_int = int(field_value) |
| except ValueError: |
| return build_config |
| existing_fields = {} |
| if field_value_int > self.MAX_FIELDS: |
| build_config["number_of_fields"]["value"] = self.MAX_FIELDS |
| msg = ( |
| f"Number of fields cannot exceed {self.MAX_FIELDS}. " |
| "Please adjust the number of fields to be within the allowed limit." |
| ) |
| raise ValueError(msg) |
| if len(build_config) > len(default_keys): |
| |
| for key in build_config.copy(): |
| if key not in default_keys: |
| existing_fields[key] = build_config.pop(key) |
|
|
| for i in range(1, field_value_int + 1): |
| key = f"field_{i}_key" |
| if key in existing_fields: |
| field = existing_fields[key] |
| build_config[key] = field |
| else: |
| field = DictInput( |
| display_name=f"Field {i}", |
| name=key, |
| info=f"Key for field {i}.", |
| input_types=["Text", "Data"], |
| ) |
| build_config[field.name] = field.to_dict() |
|
|
| build_config["number_of_fields"]["value"] = field_value_int |
| return build_config |
|
|
| async def build_data(self) -> Data: |
| data = self.get_data() |
| return_data = Data(data=data, text_key=self.text_key) |
| self.status = return_data |
| if self.text_key_validator: |
| self.validate_text_key() |
| return return_data |
|
|
| def get_data(self): |
| """Function to get the Data from the attributes.""" |
| data = {} |
| for value_dict in self._attributes.values(): |
| if isinstance(value_dict, dict): |
| |
| value_dict_ = { |
| key: value.get_text() if isinstance(value, Data) else value for key, value in value_dict.items() |
| } |
| data.update(value_dict_) |
| return data |
|
|
| def validate_text_key(self) -> None: |
| """This function validates that the Text Key is one of the keys in the Data.""" |
| data_keys = self.get_data().keys() |
| if self.text_key not in data_keys and self.text_key != "": |
| formatted_data_keys = ", ".join(data_keys) |
| msg = f"Text Key: '{self.text_key}' not found in the Data keys: '{formatted_data_keys}'" |
| raise ValueError(msg) |
|
|
