Spaces:
Running
Running
| from typing import TYPE_CHECKING, Literal | |
| 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, | |
| FieldTypes.OTHER: str, | |
| } | |
| if TYPE_CHECKING: | |
| from langflow.inputs.inputs import InputTypes | |
| def create_input_schema(inputs: list["InputTypes"]) -> type[BaseModel]: | |
| if not isinstance(inputs, list): | |
| msg = "inputs must be a list of Inputs" | |
| raise TypeError(msg) | |
| 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] | |
| else: | |
| msg = f"Invalid field type: {field_type}" | |
| raise TypeError(msg) | |
| 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}) # noqa: S307 | |
| if hasattr(input_model, "is_list") and input_model.is_list: | |
| field_type = list[field_type] # type: ignore[valid-type] | |
| if input_model.name: | |
| name = input_model.name.replace("_", " ").title() | |
| elif input_model.display_name: | |
| name = input_model.display_name | |
| else: | |
| msg = "Input name or display_name is required" | |
| raise ValueError(msg) | |
| field_dict = { | |
| "title": name, | |
| "description": input_model.info or "", | |
| } | |
| if input_model.required is False: | |
| field_dict["default"] = input_model.value # type: ignore[assignment] | |
| 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 | |