jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

Refactor directory_reader.py for improved readability and maintainability

Browse source
This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-02-29 19:27:22 -03:00
commit 6ae5a39f8d
1 changed files with 40 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

55
src/backend/langflow/interface/custom/directory_reader/directory_reader.py
View file

@ -80,9 +80,13 @@ class DirectoryReader:
except Exception as e: except Exception as e:
logger.error(f"Error while loading component: {e}") logger.error(f"Error while loading component: {e}")
continue continue
items.append({"name": menu["name"], "path": menu["path"], "components": components}) items.append(
{"name": menu["name"], "path": menu["path"], "components": components}
)
filtered = [menu for menu in items if menu["components"]] filtered = [menu for menu in items if menu["components"]]
logger.debug(f'Filtered components {"with errors" if with_errors else ""}: {len(filtered)}') logger.debug(
f'Filtered components {"with errors" if with_errors else ""}: {len(filtered)}'
)
return {"menu": filtered} return {"menu": filtered}
def validate_code(self, file_content): def validate_code(self, file_content):
@ -115,7 +119,9 @@ class DirectoryReader:
Walk through the directory path and return a list of all .py files. Walk through the directory path and return a list of all .py files.
""" """
if not (safe_path := self.get_safe_path()): if not (safe_path := self.get_safe_path()):
raise CustomComponentPathValueError(f"The path needs to start with '{self.base_path}'.") raise CustomComponentPathValueError(
f"The path needs to start with '{self.base_path}'."
)
file_list = [] file_list = []
safe_path_obj = Path(safe_path) safe_path_obj = Path(safe_path)
@ -125,7 +131,11 @@ class DirectoryReader:
# any folders below [folder] will be ignored # any folders below [folder] will be ignored
# basically the parent folder of the file should be a # basically the parent folder of the file should be a
# folder in the safe_path # folder in the safe_path
if file_path.is_file() and file_path.parent.parent == safe_path_obj and not file_path.name.startswith("__"): if (
file_path.is_file()
and file_path.parent.parent == safe_path_obj
and not file_path.name.startswith("__")
):
file_list.append(str(file_path)) file_list.append(str(file_path))
return file_list return file_list
@ -163,7 +173,9 @@ class DirectoryReader:
for node in ast.walk(module): for node in ast.walk(module):
if isinstance(node, ast.FunctionDef): if isinstance(node, ast.FunctionDef):
for arg in node.args.args: for arg in node.args.args:
if self._is_type_hint_in_arg_annotation(arg.annotation, type_hint_name): if self._is_type_hint_in_arg_annotation(
arg.annotation, type_hint_name
):
return True return True
except SyntaxError: except SyntaxError:
# Returns False if the code is not valid Python # Returns False if the code is not valid Python
@ -181,14 +193,16 @@ class DirectoryReader:
and annotation.value.id == type_hint_name and annotation.value.id == type_hint_name
) )
def is_type_hint_used_but_not_imported(self, type_hint_name: str, code: str) -> bool: def is_type_hint_used_but_not_imported(
self, type_hint_name: str, code: str
) -> bool:
""" """
Check if a type hint is used but not imported in the given code. Check if a type hint is used but not imported in the given code.
""" """
try: try:
return self._is_type_hint_used_in_args(type_hint_name, code) and not self._is_type_hint_imported( return self._is_type_hint_used_in_args(
type_hint_name, code type_hint_name, code
) ) and not self._is_type_hint_imported(type_hint_name, code)
except SyntaxError: except SyntaxError:
# Returns True if there's something wrong with the code # Returns True if there's something wrong with the code
# TODO : Find a better way to handle this # TODO : Find a better way to handle this
@ -209,9 +223,9 @@ class DirectoryReader:
return False, "Syntax error" return False, "Syntax error"
elif not self.validate_build(file_content): elif not self.validate_build(file_content):
return False, "Missing build function" return False, "Missing build function"
elif self._is_type_hint_used_in_args("Optional", file_content) and not self._is_type_hint_imported( elif self._is_type_hint_used_in_args(
"Optional", file_content "Optional", file_content
): ) and not self._is_type_hint_imported("Optional", file_content):
return ( return (
False, False,
"Type hint 'Optional' is used but not imported in the code.", "Type hint 'Optional' is used but not imported in the code.",
@ -227,14 +241,18 @@ class DirectoryReader:
from the .py files in the directory. from the .py files in the directory.
""" """
response = {"menu": []} response = {"menu": []}
logger.debug("-------------------- Building component menu list --------------------") logger.debug(
"-------------------- Building component menu list --------------------"
)
for file_path in file_paths: for file_path in file_paths:
menu_name = os.path.basename(os.path.dirname(file_path)) menu_name = os.path.basename(os.path.dirname(file_path))
filename = os.path.basename(file_path) filename = os.path.basename(file_path)
validation_result, result_content = self.process_file(file_path) validation_result, result_content = self.process_file(file_path)
if not validation_result: if not validation_result:
logger.error(f"Error while processing file {file_path}: {result_content}") logger.error(
f"Error while processing file {file_path}: {result_content}"
)
menu_result = self.find_menu(response, menu_name) or { menu_result = self.find_menu(response, menu_name) or {
"name": menu_name, "name": menu_name,
@ -247,7 +265,9 @@ class DirectoryReader:
# first check if it's already CamelCase # first check if it's already CamelCase
if "_" in component_name: if "_" in component_name:
component_name_camelcase = " ".join(word.title() for word in component_name.split("_")) component_name_camelcase = " ".join(
word.title() for word in component_name.split("_")
)
else: else:
component_name_camelcase = component_name component_name_camelcase = component_name
@ -255,7 +275,9 @@ class DirectoryReader:
try: try:
output_types = self.get_output_types_from_code(result_content) output_types = self.get_output_types_from_code(result_content)
except Exception as exc: except Exception as exc:
logger.exception(f"Error while getting output types from code: {str(exc)}") logger.exception(
f"Error while getting output types from code: {str(exc)}"
)
output_types = [component_name_camelcase] output_types = [component_name_camelcase]
else: else:
output_types = [component_name_camelcase] output_types = [component_name_camelcase]
@ -271,7 +293,9 @@ class DirectoryReader:
if menu_result not in response["menu"]: if menu_result not in response["menu"]:
response["menu"].append(menu_result) response["menu"].append(menu_result)
logger.debug("-------------------- Component menu list built --------------------") logger.debug(
"-------------------- Component menu list built --------------------"
)
return response return response
@staticmethod @staticmethod
@ -281,5 +305,6 @@ class DirectoryReader:
""" """
custom_component = CustomComponent(code=code) custom_component = CustomComponent(code=code)
types_list = custom_component.get_function_entrypoint_return_type types_list = custom_component.get_function_entrypoint_return_type
# Get the name of types classes # Get the name of types classes
return [type_.__name__ for type_ in types_list if hasattr(type_, "__name__")] return [type_.__name__ for type_ in types_list if hasattr(type_, "__name__")]