Underground-Digital/Workflow-Engine

1from core.extension.api_based_extension_requestor import APIBasedExtensionRequestor2from core.helper.encrypter import decrypt_token, encrypt_token3from extensions.ext_database import db4from models.api_based_extension import APIBasedExtension, APIBasedExtensionPoint5 6 7class APIBasedExtensionService:8    @staticmethod9    def get_all_by_tenant_id(tenant_id: str) -> list[APIBasedExtension]:10        extension_list = (11            db.session.query(APIBasedExtension)12            .filter_by(tenant_id=tenant_id)13            .order_by(APIBasedExtension.created_at.desc())14            .all()15        )16 17        for extension in extension_list:18            extension.api_key = decrypt_token(extension.tenant_id, extension.api_key)19 20        return extension_list21 22    @classmethod23    def save(cls, extension_data: APIBasedExtension) -> APIBasedExtension:24        cls._validation(extension_data)25 26        extension_data.api_key = encrypt_token(extension_data.tenant_id, extension_data.api_key)27 28        db.session.add(extension_data)29        db.session.commit()30        return extension_data31 32    @staticmethod33    def delete(extension_data: APIBasedExtension) -> None:34        db.session.delete(extension_data)35        db.session.commit()36 37    @staticmethod38    def get_with_tenant_id(tenant_id: str, api_based_extension_id: str) -> APIBasedExtension:39        extension = (40            db.session.query(APIBasedExtension)41            .filter_by(tenant_id=tenant_id)42            .filter_by(id=api_based_extension_id)43            .first()44        )45 46        if not extension:47            raise ValueError("API based extension is not found")48 49        extension.api_key = decrypt_token(extension.tenant_id, extension.api_key)50 51        return extension52 53    @classmethod54    def _validation(cls, extension_data: APIBasedExtension) -> None:55        # name56        if not extension_data.name:57            raise ValueError("name must not be empty")58 59        if not extension_data.id:60            # case one: check new data, name must be unique61            is_name_existed = (62                db.session.query(APIBasedExtension)63                .filter_by(tenant_id=extension_data.tenant_id)64                .filter_by(name=extension_data.name)65                .first()66            )67 68            if is_name_existed:69                raise ValueError("name must be unique, it is already existed")70        else:71            # case two: check existing data, name must be unique72            is_name_existed = (73                db.session.query(APIBasedExtension)74                .filter_by(tenant_id=extension_data.tenant_id)75                .filter_by(name=extension_data.name)76                .filter(APIBasedExtension.id != extension_data.id)77                .first()78            )79 80            if is_name_existed:81                raise ValueError("name must be unique, it is already existed")82 83        # api_endpoint84        if not extension_data.api_endpoint:85            raise ValueError("api_endpoint must not be empty")86 87        # api_key88        if not extension_data.api_key:89            raise ValueError("api_key must not be empty")90 91        if len(extension_data.api_key) < 5:92            raise ValueError("api_key must be at least 5 characters")93 94        # check endpoint95        cls._ping_connection(extension_data)96 97    @staticmethod98    def _ping_connection(extension_data: APIBasedExtension) -> None:99        try:100            client = APIBasedExtensionRequestor(extension_data.api_endpoint, extension_data.api_key)101            resp = client.request(point=APIBasedExtensionPoint.PING, params={})102            if resp.get("result") != "pong":103                raise ValueError(resp)104        except Exception as e:105            raise ValueError("connection error: {}".format(e))106