chore: update type annotations in template, io and inputs modules (#3593)

* refactor(input_mixin.py): update typing annotations for variables to use union types for better clarity and compatibility with Python 3.10
refactor(inputs.py): update typing annotations for variables to use union types and import necessary modules for compatibility with Python 3.10

* refactor(base.py): remove unnecessary imports and update typing for fields in Input and Output classes
feat(base.py): add support for specifying field types more explicitly in Input and Output classes
feat(frontend_node/base.py): enhance typing and field definitions in FrontendNode class
feat(frontend_node/custom_components.py): improve typing and field definitions in CustomComponentFrontendNode and ComponentFrontendNode classes
feat(template/base.py): update typing for fields in Template class and remove unnecessary imports

* refactor(inputs): remove unnecessary Optional import from typing in input_mixin.py and inputs.py files to improve code readability and maintainability

* refactor(schema.py): change 'Type' to 'type' for consistency in type annotations
refactor(schema.py): change 'list' to 'List' and 'Literal' to 'literal' for correct type hinting in create_input_schema function

* test: fix tests

---------

Co-authored-by: italojohnny <italojohnnydosanjos@gmail.com>
This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-08-28 16:35:01 -03:00 • committed by GitHub
commit 1883710dd1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 63 additions and 63 deletions

View file

@ -1,5 +1,5 @@
from enum import Enum from enum import Enum
from typing import Annotated, Any, Optional from typing import Annotated, Any
from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, field_validator, model_serializer from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, field_validator, model_serializer
@ -50,27 +50,27 @@ class BaseInputMixin(BaseModel, validate_assignment=True): # type: ignore
value: Any = "" value: Any = ""
"""The value of the field. Default is an empty string.""" """The value of the field. Default is an empty string."""
display_name: Optional[str] = None display_name: str | None = None
"""Display name of the field. Defaults to None.""" """Display name of the field. Defaults to None."""
advanced: bool = False advanced: bool = False
"""Specifies if the field will an advanced parameter (hidden). Defaults to False.""" """Specifies if the field will an advanced parameter (hidden). Defaults to False."""
input_types: Optional[list[str]] = None input_types: list[str] | None = None
"""List of input types for the handle when the field has more than one type. Default is an empty list.""" """List of input types for the handle when the field has more than one type. Default is an empty list."""
dynamic: bool = False dynamic: bool = False
"""Specifies if the field is dynamic. Defaults to False.""" """Specifies if the field is dynamic. Defaults to False."""
info: Optional[str] = "" info: str | None = ""
"""Additional information about the field to be shown in the tooltip. Defaults to an empty string.""" """Additional information about the field to be shown in the tooltip. Defaults to an empty string."""
real_time_refresh: Optional[bool] = None real_time_refresh: bool | None = None
"""Specifies if the field should have real time refresh. `refresh_button` must be False. Defaults to None.""" """Specifies if the field should have real time refresh. `refresh_button` must be False. Defaults to None."""
refresh_button: Optional[bool] = None refresh_button: bool | None = None
"""Specifies if the field should have a refresh button. Defaults to False.""" """Specifies if the field should have a refresh button. Defaults to False."""
refresh_button_text: Optional[str] = None refresh_button_text: str | None = None
"""Specifies the text for the refresh button. Defaults to None.""" """Specifies the text for the refresh button. Defaults to None."""
title_case: bool = False title_case: bool = False
@ -116,7 +116,7 @@ class DatabaseLoadMixin(BaseModel):
# Specific mixin for fields needing file interaction # Specific mixin for fields needing file interaction
class FileMixin(BaseModel): class FileMixin(BaseModel):
file_path: Optional[str] = Field(default="") file_path: str | None = Field(default="")
file_types: list[str] = Field(default=[], alias="fileTypes") file_types: list[str] = Field(default=[], alias="fileTypes")
@field_validator("file_types") @field_validator("file_types")
@ -134,11 +134,11 @@ class FileMixin(BaseModel):
class RangeMixin(BaseModel): class RangeMixin(BaseModel):
range_spec: Optional[RangeSpec] = None range_spec: RangeSpec | None = None
class DropDownMixin(BaseModel): class DropDownMixin(BaseModel):
options: Optional[list[str]] = None options: list[str] | None = None
"""List of options for the field. Only used when is_list=True. Default is an empty list.""" """List of options for the field. Only used when is_list=True. Default is an empty list."""
combobox: CoalesceBool = False combobox: CoalesceBool = False
"""Variable that defines if the user can insert custom values in the dropdown.""" """Variable that defines if the user can insert custom values in the dropdown."""
@ -149,7 +149,7 @@ class MultilineMixin(BaseModel):
class TableMixin(BaseModel): class TableMixin(BaseModel):
table_schema: Optional[TableSchema | list[Column]] = None table_schema: TableSchema | list[Column] | None = None
@field_validator("table_schema") @field_validator("table_schema")
@classmethod @classmethod

View file

@ -1,5 +1,6 @@
import warnings import warnings
from typing import Any, AsyncIterator, Iterator, Optional, Union, get_args from typing import Any, Union, get_args
from collections.abc import AsyncIterator, Iterator
from pydantic import Field, field_validator from pydantic import Field, field_validator
@ -379,7 +380,7 @@ class NestedDictInput(BaseInputMixin, ListableInputMixin, MetadataTraceMixin, In
""" """
field_type: SerializableFieldTypes = FieldTypes.NESTED_DICT field_type: SerializableFieldTypes = FieldTypes.NESTED_DICT
value: Optional[dict | Data] = {} value: dict | Data | None = {}
class DictInput(BaseInputMixin, ListableInputMixin, InputTraceMixin): class DictInput(BaseInputMixin, ListableInputMixin, InputTraceMixin):
@ -395,7 +396,7 @@ class DictInput(BaseInputMixin, ListableInputMixin, InputTraceMixin):
""" """
field_type: SerializableFieldTypes = FieldTypes.DICT field_type: SerializableFieldTypes = FieldTypes.DICT
value: Optional[dict] = {} value: dict | None = {}
class DropdownInput(BaseInputMixin, DropDownMixin, MetadataTraceMixin): class DropdownInput(BaseInputMixin, DropDownMixin, MetadataTraceMixin):
@ -465,7 +466,7 @@ DEFAULT_PROMPT_INTUT_TYPES = ["Message", "Text"]
class DefaultPromptField(Input): class DefaultPromptField(Input):
name: str name: str
display_name: Optional[str] = None display_name: str | None = None
field_type: str = "str" field_type: str = "str"
advanced: bool = False advanced: bool = False

View file

@ -1,10 +1,10 @@
from typing import TYPE_CHECKING, List, Literal, Type from typing import TYPE_CHECKING, Literal
from pydantic import BaseModel, Field, create_model from pydantic import BaseModel, Field, create_model
from langflow.inputs.inputs import FieldTypes from langflow.inputs.inputs import FieldTypes
_convert_field_type_to_type: dict[FieldTypes, Type] = { _convert_field_type_to_type: dict[FieldTypes, type] = {
FieldTypes.TEXT: str, FieldTypes.TEXT: str,
FieldTypes.INTEGER: int, FieldTypes.INTEGER: int,
FieldTypes.FLOAT: float, FieldTypes.FLOAT: float,
@ -20,7 +20,7 @@ if TYPE_CHECKING:
from langflow.inputs.inputs import InputTypes from langflow.inputs.inputs import InputTypes
def create_input_schema(inputs: list["InputTypes"]) -> Type[BaseModel]: def create_input_schema(inputs: list["InputTypes"]) -> type[BaseModel]:
if not isinstance(inputs, list): if not isinstance(inputs, list):
raise TypeError("inputs must be a list of Inputs") raise TypeError("inputs must be a list of Inputs")
fields = {} fields = {}
@ -35,7 +35,7 @@ def create_input_schema(inputs: list["InputTypes"]) -> Type[BaseModel]:
field_type = eval(literal_string, {"Literal": Literal}) # type: ignore field_type = eval(literal_string, {"Literal": Literal}) # type: ignore
if hasattr(input_model, "is_list") and input_model.is_list: if hasattr(input_model, "is_list") and input_model.is_list:
field_type = List[field_type] # type: ignore field_type = list[field_type] # type: ignore
if input_model.name: if input_model.name:
name = input_model.name.replace("_", " ").title() name = input_model.name.replace("_", " ").title()
elif input_model.display_name: elif input_model.display_name:

View file

@ -2,7 +2,8 @@ from enum import Enum
from typing import GenericAlias # type: ignore from typing import GenericAlias # type: ignore
from typing import _GenericAlias # type: ignore from typing import _GenericAlias # type: ignore
from typing import _UnionGenericAlias # type: ignore from typing import _UnionGenericAlias # type: ignore
from typing import Any, Callable, Optional, Union from typing import Any
from collections.abc import Callable
from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator, model_serializer, model_validator from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator, model_serializer, model_validator
@ -46,42 +47,42 @@ class Input(BaseModel):
file_types: list[str] = Field(default=[], serialization_alias="fileTypes") file_types: list[str] = Field(default=[], serialization_alias="fileTypes")
"""List of file types associated with the field . Default is an empty list.""" """List of file types associated with the field . Default is an empty list."""
file_path: Optional[str] = "" file_path: str | None = ""
"""The file path of the field if it is a file. Defaults to None.""" """The file path of the field if it is a file. Defaults to None."""
password: bool = False password: bool = False
"""Specifies if the field is a password. Defaults to False.""" """Specifies if the field is a password. Defaults to False."""
options: Optional[Union[list[str], Callable]] = None options: list[str] | Callable | None = None
"""List of options for the field. Only used when is_list=True. Default is an empty list.""" """List of options for the field. Only used when is_list=True. Default is an empty list."""
name: Optional[str] = None name: str | None = None
"""Name of the field. Default is an empty string.""" """Name of the field. Default is an empty string."""
display_name: Optional[str] = None display_name: str | None = None
"""Display name of the field. Defaults to None.""" """Display name of the field. Defaults to None."""
advanced: bool = False advanced: bool = False
"""Specifies if the field will an advanced parameter (hidden). Defaults to False.""" """Specifies if the field will an advanced parameter (hidden). Defaults to False."""
input_types: Optional[list[str]] = None input_types: list[str] | None = None
"""List of input types for the handle when the field has more than one type. Default is an empty list.""" """List of input types for the handle when the field has more than one type. Default is an empty list."""
dynamic: bool = False dynamic: bool = False
"""Specifies if the field is dynamic. Defaults to False.""" """Specifies if the field is dynamic. Defaults to False."""
info: Optional[str] = "" info: str | None = ""
"""Additional information about the field to be shown in the tooltip. Defaults to an empty string.""" """Additional information about the field to be shown in the tooltip. Defaults to an empty string."""
real_time_refresh: Optional[bool] = None real_time_refresh: bool | None = None
"""Specifies if the field should have real time refresh. `refresh_button` must be False. Defaults to None.""" """Specifies if the field should have real time refresh. `refresh_button` must be False. Defaults to None."""
refresh_button: Optional[bool] = None refresh_button: bool | None = None
"""Specifies if the field should have a refresh button. Defaults to False.""" """Specifies if the field should have a refresh button. Defaults to False."""
refresh_button_text: Optional[str] = None refresh_button_text: str | None = None
"""Specifies the text for the refresh button. Defaults to None.""" """Specifies the text for the refresh button. Defaults to None."""
range_spec: Optional[RangeSpec] = Field(default=None, serialization_alias="rangeSpec") range_spec: RangeSpec | None = Field(default=None, serialization_alias="rangeSpec")
"""Range specification for the field. Defaults to None.""" """Range specification for the field. Defaults to None."""
load_from_db: bool = False load_from_db: bool = False
@ -161,22 +162,22 @@ class Output(BaseModel):
types: list[str] = Field(default=[]) types: list[str] = Field(default=[])
"""List of output types for the field.""" """List of output types for the field."""
selected: Optional[str] = Field(default=None) selected: str | None = Field(default=None)
"""The selected output type for the field.""" """The selected output type for the field."""
name: str = Field(description="The name of the field.") name: str = Field(description="The name of the field.")
"""The name of the field.""" """The name of the field."""
hidden: Optional[bool] = Field(default=None) hidden: bool | None = Field(default=None)
"""Dictates if the field is hidden.""" """Dictates if the field is hidden."""
display_name: Optional[str] = Field(default=None) display_name: str | None = Field(default=None)
"""The display name of the field.""" """The display name of the field."""
method: Optional[str] = Field(default=None) method: str | None = Field(default=None)
"""The method to use for the output.""" """The method to use for the output."""
value: Optional[Any] = Field(default=UNDEFINED) value: Any | None = Field(default=UNDEFINED)
cache: bool = Field(default=True) cache: bool = Field(default=True)

View file

@ -1,5 +1,4 @@
from collections import defaultdict from collections import defaultdict
from typing import Dict, List, Optional, Union
from pydantic import BaseModel, field_serializer, model_serializer from pydantic import BaseModel, field_serializer, model_serializer
@ -11,46 +10,46 @@ class FrontendNode(BaseModel):
_format_template: bool = True _format_template: bool = True
template: Template template: Template
"""Template for the frontend node.""" """Template for the frontend node."""
description: Optional[str] = None description: str | None = None
"""Description of the frontend node.""" """Description of the frontend node."""
icon: Optional[str] = None icon: str | None = None
"""Icon of the frontend node.""" """Icon of the frontend node."""
is_input: Optional[bool] = None is_input: bool | None = None
"""Whether the frontend node is used as an input when processing the Graph. """Whether the frontend node is used as an input when processing the Graph.
If True, there should be a field named 'input_value'.""" If True, there should be a field named 'input_value'."""
is_output: Optional[bool] = None is_output: bool | None = None
"""Whether the frontend node is used as an output when processing the Graph. """Whether the frontend node is used as an output when processing the Graph.
If True, there should be a field named 'input_value'.""" If True, there should be a field named 'input_value'."""
is_composition: Optional[bool] = None is_composition: bool | None = None
"""Whether the frontend node is used for composition.""" """Whether the frontend node is used for composition."""
base_classes: List[str] base_classes: list[str]
"""List of base classes for the frontend node.""" """List of base classes for the frontend node."""
name: str = "" name: str = ""
"""Name of the frontend node.""" """Name of the frontend node."""
display_name: Optional[str] = "" display_name: str | None = ""
"""Display name of the frontend node.""" """Display name of the frontend node."""
documentation: str = "" documentation: str = ""
"""Documentation of the frontend node.""" """Documentation of the frontend node."""
custom_fields: Optional[Dict] = defaultdict(list) custom_fields: dict | None = defaultdict(list)
"""Custom fields of the frontend node.""" """Custom fields of the frontend node."""
output_types: List[str] = [] output_types: list[str] = []
"""List of output types for the frontend node.""" """List of output types for the frontend node."""
full_path: Optional[str] = None full_path: str | None = None
"""Full path of the frontend node.""" """Full path of the frontend node."""
pinned: bool = False pinned: bool = False
"""Whether the frontend node is pinned.""" """Whether the frontend node is pinned."""
conditional_paths: List[str] = [] conditional_paths: list[str] = []
"""List of conditional paths for the frontend node.""" """List of conditional paths for the frontend node."""
frozen: bool = False frozen: bool = False
"""Whether the frontend node is frozen.""" """Whether the frontend node is frozen."""
outputs: List[Output] = [] outputs: list[Output] = []
"""List of output fields for the frontend node.""" """List of output fields for the frontend node."""
field_order: list[str] = [] field_order: list[str] = []
"""Order of the fields in the frontend node.""" """Order of the fields in the frontend node."""
beta: bool = False beta: bool = False
"""Whether the frontend node is in beta.""" """Whether the frontend node is in beta."""
error: Optional[str] = None error: str | None = None
"""Error message for the frontend node.""" """Error message for the frontend node."""
edited: bool = False edited: bool = False
"""Whether the frontend node has been edited.""" """Whether the frontend node has been edited."""
@ -60,7 +59,7 @@ class FrontendNode(BaseModel):
self.documentation = documentation self.documentation = documentation
@field_serializer("base_classes") @field_serializer("base_classes")
def process_base_classes(self, base_classes: List[str]) -> List[str]: def process_base_classes(self, base_classes: list[str]) -> list[str]:
"""Removes unwanted base classes from the list of base classes.""" """Removes unwanted base classes from the list of base classes."""
sorted_base_classes = sorted(list(set(base_classes)), key=lambda x: x.lower()) sorted_base_classes = sorted(list(set(base_classes)), key=lambda x: x.lower())
@ -156,14 +155,14 @@ class FrontendNode(BaseModel):
input_overlap_str = ", ".join(map(lambda x: f"'{x}'", input_overlap)) input_overlap_str = ", ".join(map(lambda x: f"'{x}'", input_overlap))
error_message += f"Input names {input_overlap_str} are reserved attributes." error_message += f"Input names {input_overlap_str} are reserved attributes."
def add_base_class(self, base_class: Union[str, List[str]]) -> None: def add_base_class(self, base_class: str | list[str]) -> None:
"""Adds a base class to the frontend node.""" """Adds a base class to the frontend node."""
if isinstance(base_class, str): if isinstance(base_class, str):
self.base_classes.append(base_class) self.base_classes.append(base_class)
elif isinstance(base_class, list): elif isinstance(base_class, list):
self.base_classes.extend(base_class) self.base_classes.extend(base_class)
def add_output_type(self, output_type: Union[str, List[str]]) -> None: def add_output_type(self, output_type: str | list[str]) -> None:
"""Adds an output type to the frontend node.""" """Adds an output type to the frontend node."""
if isinstance(output_type, str): if isinstance(output_type, str):
self.output_types.append(output_type) self.output_types.append(output_type)

View file

@ -1,5 +1,3 @@
from typing import Optional
from langflow.template.field.base import Input from langflow.template.field.base import Input
from langflow.template.frontend_node.base import FrontendNode from langflow.template.frontend_node.base import FrontendNode
from langflow.template.template.base import Template from langflow.template.template.base import Template
@ -47,7 +45,7 @@ class Component(CustomComponent):
class CustomComponentFrontendNode(FrontendNode): class CustomComponentFrontendNode(FrontendNode):
_format_template: bool = False _format_template: bool = False
name: str = "CustomComponent" name: str = "CustomComponent"
display_name: Optional[str] = "CustomComponent" display_name: str | None = "CustomComponent"
beta: bool = False beta: bool = False
template: Template = Template( template: Template = Template(
type_name="CustomComponent", type_name="CustomComponent",
@ -65,14 +63,14 @@ class CustomComponentFrontendNode(FrontendNode):
) )
], ],
) )
description: Optional[str] = None description: str | None = None
base_classes: list[str] = [] base_classes: list[str] = []
class ComponentFrontendNode(FrontendNode): class ComponentFrontendNode(FrontendNode):
_format_template: bool = False _format_template: bool = False
name: str = "Component" name: str = "Component"
display_name: Optional[str] = "Component" display_name: str | None = "Component"
beta: bool = False beta: bool = False
template: Template = Template( template: Template = Template(
type_name="Component", type_name="Component",
@ -90,5 +88,5 @@ class ComponentFrontendNode(FrontendNode):
) )
], ],
) )
description: Optional[str] = None description: str | None = None
base_classes: list[str] = [] base_classes: list[str] = []

View file

@ -1,4 +1,5 @@
from typing import Callable, Union, cast from typing import cast
from collections.abc import Callable
from pydantic import BaseModel, Field, model_serializer from pydantic import BaseModel, Field, model_serializer
@ -13,7 +14,7 @@ class Template(BaseModel):
def process_fields( def process_fields(
self, self,
format_field_func: Union[Callable, None] = None, format_field_func: Callable | None = None,
): ):
if format_field_func: if format_field_func:
for field in self.fields: for field in self.fields:

View file

@ -1,4 +1,4 @@
from typing import List, Literal from typing import Literal
import pytest import pytest
from pydantic.fields import FieldInfo from pydantic.fields import FieldInfo
@ -96,7 +96,7 @@ class TestCreateInputSchema:
input_instance = StrInput(name="test_field", is_list=True) input_instance = StrInput(name="test_field", is_list=True)
schema = create_input_schema([input_instance]) schema = create_input_schema([input_instance])
field_info: FieldInfo = schema.model_fields["test_field"] field_info: FieldInfo = schema.model_fields["test_field"]
assert field_info.annotation == List[str] assert field_info.annotation == list[str]
# Input with options attribute is processed correctly # Input with options attribute is processed correctly
def test_options_attribute_processing(self): def test_options_attribute_processing(self):
@ -184,7 +184,7 @@ class TestCreateInputSchema:
input_instance = StrInput(name="test_field", is_list=True) input_instance = StrInput(name="test_field", is_list=True)
schema = create_input_schema([input_instance]) schema = create_input_schema([input_instance])
field_info = schema.model_fields["test_field"] field_info = schema.model_fields["test_field"]
assert field_info.annotation == List[str] assert field_info.annotation == list[str]
# Converting FieldTypes to corresponding Python types # Converting FieldTypes to corresponding Python types
def test_field_types_conversion(self): def test_field_types_conversion(self):