Underground-Digital/Workflow-Engine

1from datetime import datetime, timezone2 3from flask_login import current_user4from flask_restful import Resource, inputs, marshal_with, reqparse5from sqlalchemy import and_6from werkzeug.exceptions import BadRequest, Forbidden, NotFound7 8from controllers.console import api9from controllers.console.explore.wraps import InstalledAppResource10from controllers.console.wraps import account_initialization_required, cloud_edition_billing_resource_check11from extensions.ext_database import db12from fields.installed_app_fields import installed_app_list_fields13from libs.login import login_required14from models import App, InstalledApp, RecommendedApp15from services.account_service import TenantService16 17 18class InstalledAppsListApi(Resource):19    @login_required20    @account_initialization_required21    @marshal_with(installed_app_list_fields)22    def get(self):23        current_tenant_id = current_user.current_tenant_id24        installed_apps = db.session.query(InstalledApp).filter(InstalledApp.tenant_id == current_tenant_id).all()25 26        current_user.role = TenantService.get_user_role(current_user, current_user.current_tenant)27        installed_apps = [28            {29                "id": installed_app.id,30                "app": installed_app.app,31                "app_owner_tenant_id": installed_app.app_owner_tenant_id,32                "is_pinned": installed_app.is_pinned,33                "last_used_at": installed_app.last_used_at,34                "editable": current_user.role in {"owner", "admin"},35                "uninstallable": current_tenant_id == installed_app.app_owner_tenant_id,36            }37            for installed_app in installed_apps38            if installed_app.app is not None39        ]40        installed_apps.sort(41            key=lambda app: (42                -app["is_pinned"],43                app["last_used_at"] is None,44                -app["last_used_at"].timestamp() if app["last_used_at"] is not None else 0,45            )46        )47 48        return {"installed_apps": installed_apps}49 50    @login_required51    @account_initialization_required52    @cloud_edition_billing_resource_check("apps")53    def post(self):54        parser = reqparse.RequestParser()55        parser.add_argument("app_id", type=str, required=True, help="Invalid app_id")56        args = parser.parse_args()57 58        recommended_app = RecommendedApp.query.filter(RecommendedApp.app_id == args["app_id"]).first()59        if recommended_app is None:60            raise NotFound("App not found")61 62        current_tenant_id = current_user.current_tenant_id63        app = db.session.query(App).filter(App.id == args["app_id"]).first()64 65        if app is None:66            raise NotFound("App not found")67 68        if not app.is_public:69            raise Forbidden("You can't install a non-public app")70 71        installed_app = InstalledApp.query.filter(72            and_(InstalledApp.app_id == args["app_id"], InstalledApp.tenant_id == current_tenant_id)73        ).first()74 75        if installed_app is None:76            # todo: position77            recommended_app.install_count += 178 79            new_installed_app = InstalledApp(80                app_id=args["app_id"],81                tenant_id=current_tenant_id,82                app_owner_tenant_id=app.tenant_id,83                is_pinned=False,84                last_used_at=datetime.now(timezone.utc).replace(tzinfo=None),85            )86            db.session.add(new_installed_app)87            db.session.commit()88 89        return {"message": "App installed successfully"}90 91 92class InstalledAppApi(InstalledAppResource):93    """94    update and delete an installed app95    use InstalledAppResource to apply default decorators and get installed_app96    """97 98    def delete(self, installed_app):99        if installed_app.app_owner_tenant_id == current_user.current_tenant_id:100            raise BadRequest("You can't uninstall an app owned by the current tenant")101 102        db.session.delete(installed_app)103        db.session.commit()104 105        return {"result": "success", "message": "App uninstalled successfully"}106 107    def patch(self, installed_app):108        parser = reqparse.RequestParser()109        parser.add_argument("is_pinned", type=inputs.boolean)110        args = parser.parse_args()111 112        commit_args = False113        if "is_pinned" in args:114            installed_app.is_pinned = args["is_pinned"]115            commit_args = True116 117        if commit_args:118            db.session.commit()119 120        return {"result": "success", "message": "App info updated successfully"}121 122 123api.add_resource(InstalledAppsListApi, "/installed-apps")124api.add_resource(InstalledAppApi, "/installed-apps/<uuid:installed_app_id>")125