Underground-Digital/Workflow-Engine

1from typing import Optional2 3from core.extension.extensible import ExtensionModule4from extensions.ext_code_based_extension import code_based_extension5 6 7class ExternalDataToolFactory:8    def __init__(self, name: str, tenant_id: str, app_id: str, variable: str, config: dict) -> None:9        extension_class = code_based_extension.extension_class(ExtensionModule.EXTERNAL_DATA_TOOL, name)10        self.__extension_instance = extension_class(11            tenant_id=tenant_id, app_id=app_id, variable=variable, config=config12        )13 14    @classmethod15    def validate_config(cls, name: str, tenant_id: str, config: dict) -> None:16        """17        Validate the incoming form config data.18 19        :param name: the name of external data tool20        :param tenant_id: the id of workspace21        :param config: the form config data22        :return:23        """24        code_based_extension.validate_form_schema(ExtensionModule.EXTERNAL_DATA_TOOL, name, config)25        extension_class = code_based_extension.extension_class(ExtensionModule.EXTERNAL_DATA_TOOL, name)26        extension_class.validate_config(tenant_id, config)27 28    def query(self, inputs: dict, query: Optional[str] = None) -> str:29        """30        Query the external data tool.31 32        :param inputs: user inputs33        :param query: the query of chat app34        :return: the tool query result35        """36        return self.__extension_instance.query(inputs, query)37