Underground-Digital/Workflow-Engine

1from abc import ABC, abstractmethod2from typing import Optional3 4from core.extension.extensible import Extensible, ExtensionModule5 6 7class ExternalDataTool(Extensible, ABC):8    """9    The base class of external data tool.10    """11 12    module: ExtensionModule = ExtensionModule.EXTERNAL_DATA_TOOL13 14    app_id: str15    """the id of app"""16    variable: str17    """the tool variable name of app tool"""18 19    def __init__(self, tenant_id: str, app_id: str, variable: str, config: Optional[dict] = None) -> None:20        super().__init__(tenant_id, config)21        self.app_id = app_id22        self.variable = variable23 24    @classmethod25    @abstractmethod26    def validate_config(cls, tenant_id: str, config: dict) -> None:27        """28        Validate the incoming form config data.29 30        :param tenant_id: the id of workspace31        :param config: the form config data32        :return:33        """34        raise NotImplementedError35 36    @abstractmethod37    def query(self, inputs: dict, query: Optional[str] = None) -> str:38        """39        Query the external data tool.40 41        :param inputs: user inputs42        :param query: the query of chat app43        :return: the tool query result44        """45        raise NotImplementedError46