import operator import re from typing import Any, ClassVar from uuid import UUID from cachetools import TTLCache, cachedmethod from fastapi import HTTPException from loguru import logger from langflow.custom.attributes import ATTR_FUNC_MAPPING from langflow.custom.code_parser import CodeParser from langflow.custom.eval import eval_custom_component_code from langflow.utils import validate class ComponentCodeNullError(HTTPException): pass class ComponentFunctionEntrypointNameNullError(HTTPException): pass class BaseComponent: ERROR_CODE_NULL: ClassVar[str] = "Python code must be provided." ERROR_FUNCTION_ENTRYPOINT_NAME_NULL: ClassVar[str] = "The name of the entrypoint function must be provided." _code: str | None = None """The code of the component. Defaults to None.""" _function_entrypoint_name: str = "build" field_config: dict = {} _user_id: str | UUID | None = None _template_config: dict = {} def __init__(self, **data) -> None: self.cache: TTLCache = TTLCache(maxsize=1024, ttl=60) for key, value in data.items(): if key == "user_id": self._user_id = value else: setattr(self, key, value) def __setattr__(self, key, value) -> None: if key == "_user_id" and self._user_id is not None: logger.warning("user_id is immutable and cannot be changed.") super().__setattr__(key, value) @cachedmethod(cache=operator.attrgetter("cache")) def get_code_tree(self, code: str): parser = CodeParser(code) return parser.parse_code() def get_function(self): if not self._code: raise ComponentCodeNullError( status_code=400, detail={"error": self.ERROR_CODE_NULL, "traceback": ""}, ) if not self._function_entrypoint_name: raise ComponentFunctionEntrypointNameNullError( status_code=400, detail={ "error": self.ERROR_FUNCTION_ENTRYPOINT_NAME_NULL, "traceback": "", }, ) return validate.create_function(self._code, self._function_entrypoint_name) @staticmethod def get_template_config(component): """Gets the template configuration for the custom component itself.""" template_config = {} for attribute, func in ATTR_FUNC_MAPPING.items(): if hasattr(component, attribute): value = getattr(component, attribute) if value is not None: template_config[attribute] = func(value=value) for key in template_config.copy(): if key not in ATTR_FUNC_MAPPING: template_config.pop(key, None) return template_config def build_template_config(self) -> dict: """Builds the template configuration for the custom component. Returns: A dictionary representing the template configuration. """ if not self._code: return {} try: cc_class = eval_custom_component_code(self._code) except AttributeError as e: pattern = r"module '.*?' has no attribute '.*?'" if re.search(pattern, str(e)): raise ImportError(e) from e raise component_instance = cc_class(_code=self._code) return self.get_template_config(component_instance) def build(self, *args: Any, **kwargs: Any) -> Any: raise NotImplementedError