|
|
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) |
|
|
|
