|
|
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 |
|
|
|
