diff --git a/src/backend/base/langflow/io/schema.py b/src/backend/base/langflow/io/schema.py new file mode 100644 index 00000000000..b5d1f8add7b --- /dev/null +++ b/src/backend/base/langflow/io/schema.py @@ -0,0 +1,58 @@ +from typing import TYPE_CHECKING, List, Literal, Type + +from pydantic import BaseModel, Field, create_model + +from langflow.inputs.inputs import FieldTypes + +_convert_field_type_to_type: dict[FieldTypes, Type] = { + FieldTypes.TEXT: str, + FieldTypes.INTEGER: int, + FieldTypes.FLOAT: float, + FieldTypes.BOOLEAN: bool, + FieldTypes.DICT: dict, + FieldTypes.NESTED_DICT: dict, + FieldTypes.TABLE: dict, + FieldTypes.FILE: str, + FieldTypes.PROMPT: str, +} + +if TYPE_CHECKING: + from langflow.inputs.inputs import InputTypes + + +def create_input_schema(inputs: list["InputTypes"]) -> Type[BaseModel]: + if not isinstance(inputs, list): + raise TypeError("inputs must be a list of Inputs") + fields = {} + for input_model in inputs: + # Create a Pydantic Field for each input field + field_type = input_model.field_type + if isinstance(field_type, FieldTypes): + field_type = _convert_field_type_to_type[field_type] + if hasattr(input_model, "options") and isinstance(input_model.options, list) and input_model.options: + literal_string = f"Literal{input_model.options}" + # validate that the literal_string is a valid literal + + field_type = eval(literal_string, {"Literal": Literal}) # type: ignore + if hasattr(input_model, "is_list") and input_model.is_list: + field_type = List[field_type] # type: ignore + if input_model.name: + name = input_model.name.replace("_", " ").title() + elif input_model.display_name: + name = input_model.display_name + else: + raise ValueError("Input name or display_name is required") + field_dict = { + "title": name, + "description": input_model.info or "", + } + if input_model.required is False: + field_dict["default"] = input_model.value # type: ignore + pydantic_field = Field(**field_dict) + + fields[input_model.name] = (field_type, pydantic_field) + + # Create and return the InputSchema model + model = create_model("InputSchema", **fields) + model.model_rebuild() + return model diff --git a/src/backend/tests/unit/io/test_io_schema.py b/src/backend/tests/unit/io/test_io_schema.py new file mode 100644 index 00000000000..63da9be0b07 --- /dev/null +++ b/src/backend/tests/unit/io/test_io_schema.py @@ -0,0 +1,251 @@ +from typing import List, Literal + +import pytest +from pydantic.fields import FieldInfo + +from langflow.components.inputs.ChatInput import ChatInput + + +@pytest.fixture +def client(): + pass + + +def test_create_input_schema(): + from langflow.io.schema import create_input_schema + + schema = create_input_schema(ChatInput.inputs) + assert schema.__name__ == "InputSchema" + + +class TestCreateInputSchema: + # Single input type is converted to list and processed correctly + def test_single_input_type_conversion(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field") + schema = create_input_schema([input_instance]) + assert schema.__name__ == "InputSchema" + assert "test_field" in schema.model_fields + + # Multiple input types are processed and included in the schema + def test_multiple_input_types(self): + from langflow.inputs.inputs import IntInput, StrInput + from langflow.io.schema import create_input_schema + + inputs = [StrInput(name="str_field"), IntInput(name="int_field")] + schema = create_input_schema(inputs) + assert schema.__name__ == "InputSchema" + assert "str_field" in schema.model_fields + assert "int_field" in schema.model_fields + + # Fields are correctly created with appropriate types and attributes + def test_fields_creation_with_correct_types_and_attributes(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field", info="Test Info", required=True) + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.description == "Test Info" + assert field_info.is_required() is True + + # Schema model is created and returned successfully + def test_schema_model_creation(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field") + schema = create_input_schema([input_instance]) + assert schema.__name__ == "InputSchema" + + # Default values are correctly assigned to fields + def test_default_values_assignment(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field", value="default_value") + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.default == "default_value" + + # Empty list of inputs is handled without errors + def test_empty_list_of_inputs(self): + from langflow.io.schema import create_input_schema + + schema = create_input_schema([]) + assert schema.__name__ == "InputSchema" + + # Input with missing optional attributes (e.g., display_name, info) is processed correctly + def test_missing_optional_attributes(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field") + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.title == "Test Field" + assert field_info.description == "" + + # Input with is_list attribute set to True is processed correctly + def test_is_list_attribute_processing(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field", is_list=True) + schema = create_input_schema([input_instance]) + field_info: FieldInfo = schema.model_fields["test_field"] + assert field_info.annotation == List[str] + + # Input with options attribute is processed correctly + def test_options_attribute_processing(self): + from langflow.inputs.inputs import DropdownInput + from langflow.io.schema import create_input_schema + + input_instance = DropdownInput(name="test_field", options=["option1", "option2"]) + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.annotation == Literal["option1", "option2"] + + # Non-standard field types are handled correctly + def test_non_standard_field_types_handling(self): + from langflow.inputs.inputs import FileInput + from langflow.io.schema import create_input_schema + + input_instance = FileInput(name="file_field") + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["file_field"] + assert field_info.annotation == str + + # Inputs with mixed required and optional fields are processed correctly + def test_mixed_required_optional_fields_processing(self): + from langflow.inputs.inputs import IntInput, StrInput + from langflow.io.schema import create_input_schema + + inputs = [ + StrInput(name="required_field", required=True), + IntInput(name="optional_field", required=False), + ] + schema = create_input_schema(inputs) + required_field_info = schema.model_fields["required_field"] + optional_field_info = schema.model_fields["optional_field"] + + assert required_field_info.is_required() is True + assert optional_field_info.is_required() is False + + # Inputs with complex nested structures are handled correctly + def test_complex_nested_structures_handling(self): + from langflow.inputs.inputs import NestedDictInput + from langflow.io.schema import create_input_schema + + nested_input = NestedDictInput(name="nested_field", value={"key": "value"}) + schema = create_input_schema([nested_input]) + + field_info = schema.model_fields["nested_field"] + + assert isinstance(field_info.default, dict) + assert field_info.default["key"] == "value" + + # Creating a schema from a single input type + def test_single_input_type_replica(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field") + schema = create_input_schema([input_instance]) + assert schema.__name__ == "InputSchema" + assert "test_field" in schema.model_fields + + # Creating a schema from a list of input types + def test_passing_input_type_directly(self): + from langflow.inputs.inputs import IntInput, StrInput + from langflow.io.schema import create_input_schema + + inputs = StrInput(name="str_field"), IntInput(name="int_field") + with pytest.raises(TypeError): + create_input_schema(inputs) + + # Handling input types with options correctly + def test_options_handling(self): + from langflow.inputs.inputs import DropdownInput + from langflow.io.schema import create_input_schema + + input_instance = DropdownInput(name="test_field", options=["option1", "option2"]) + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.annotation == Literal["option1", "option2"] + + # Handling input types with is_list attribute correctly + def test_is_list_handling(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field", is_list=True) + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.annotation == List[str] + + # Converting FieldTypes to corresponding Python types + def test_field_types_conversion(self): + from langflow.inputs.inputs import IntInput + from langflow.io.schema import create_input_schema + + input_instance = IntInput(name="int_field") + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["int_field"] + assert field_info.annotation == int + + # Setting default values for non-required fields + def test_default_values_for_non_required_fields(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field", value="default_value") + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.default == "default_value" + + # Handling input types with missing attributes + def test_missing_attributes_handling(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field") + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.title == "Test Field" + assert field_info.description == "" + + # Handling invalid field types + def test_invalid_field_types_handling(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + class InvalidFieldType: + pass + + input_instance = StrInput(name="test_field") + input_instance.field_type = InvalidFieldType() + + with pytest.raises(KeyError): + create_input_schema([input_instance]) + + # Handling input types with None as default value + def test_none_default_value_handling(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test_field", value=None) + schema = create_input_schema([input_instance]) + field_info = schema.model_fields["test_field"] + assert field_info.default is None + + # Handling input types with special characters in names + def test_special_characters_in_names_handling(self): + from langflow.inputs.inputs import StrInput + from langflow.io.schema import create_input_schema + + input_instance = StrInput(name="test@field#name") + schema = create_input_schema([input_instance]) + assert "test@field#name" in schema.model_fields