Underground-Digital/Workflow-Engine

1import logging2 3from flask_restful import Resource, fields, marshal_with, reqparse4from flask_restful.inputs import int_range5from werkzeug.exceptions import InternalServerError6 7from controllers.service_api import api8from controllers.service_api.app.error import (9    CompletionRequestError,10    NotWorkflowAppError,11    ProviderModelCurrentlyNotSupportError,12    ProviderNotInitializeError,13    ProviderQuotaExceededError,14)15from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token16from core.app.apps.base_app_queue_manager import AppQueueManager17from core.app.entities.app_invoke_entities import InvokeFrom18from core.errors.error import (19    AppInvokeQuotaExceededError,20    ModelCurrentlyNotSupportError,21    ProviderTokenNotInitError,22    QuotaExceededError,23)24from core.model_runtime.errors.invoke import InvokeError25from extensions.ext_database import db26from fields.workflow_app_log_fields import workflow_app_log_pagination_fields27from libs import helper28from models.model import App, AppMode, EndUser29from models.workflow import WorkflowRun30from services.app_generate_service import AppGenerateService31from services.workflow_app_service import WorkflowAppService32 33logger = logging.getLogger(__name__)34 35workflow_run_fields = {36    "id": fields.String,37    "workflow_id": fields.String,38    "status": fields.String,39    "inputs": fields.Raw,40    "outputs": fields.Raw,41    "error": fields.String,42    "total_steps": fields.Integer,43    "total_tokens": fields.Integer,44    "created_at": fields.DateTime,45    "finished_at": fields.DateTime,46    "elapsed_time": fields.Float,47}48 49 50class WorkflowRunDetailApi(Resource):51    @validate_app_token52    @marshal_with(workflow_run_fields)53    def get(self, app_model: App, workflow_id: str):54        """55        Get a workflow task running detail56        """57        app_mode = AppMode.value_of(app_model.mode)58        if app_mode != AppMode.WORKFLOW:59            raise NotWorkflowAppError()60 61        workflow_run = db.session.query(WorkflowRun).filter(WorkflowRun.id == workflow_id).first()62        return workflow_run63 64 65class WorkflowRunApi(Resource):66    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))67    def post(self, app_model: App, end_user: EndUser):68        """69        Run workflow70        """71        app_mode = AppMode.value_of(app_model.mode)72        if app_mode != AppMode.WORKFLOW:73            raise NotWorkflowAppError()74 75        parser = reqparse.RequestParser()76        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")77        parser.add_argument("files", type=list, required=False, location="json")78        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")79        args = parser.parse_args()80 81        streaming = args.get("response_mode") == "streaming"82 83        try:84            response = AppGenerateService.generate(85                app_model=app_model, user=end_user, args=args, invoke_from=InvokeFrom.SERVICE_API, streaming=streaming86            )87 88            return helper.compact_generate_response(response)89        except ProviderTokenNotInitError as ex:90            raise ProviderNotInitializeError(ex.description)91        except QuotaExceededError:92            raise ProviderQuotaExceededError()93        except ModelCurrentlyNotSupportError:94            raise ProviderModelCurrentlyNotSupportError()95        except InvokeError as e:96            raise CompletionRequestError(e.description)97        except (ValueError, AppInvokeQuotaExceededError) as e:98            raise e99        except Exception as e:100            logging.exception("internal server error.")101            raise InternalServerError()102 103 104class WorkflowTaskStopApi(Resource):105    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))106    def post(self, app_model: App, end_user: EndUser, task_id: str):107        """108        Stop workflow task109        """110        app_mode = AppMode.value_of(app_model.mode)111        if app_mode != AppMode.WORKFLOW:112            raise NotWorkflowAppError()113 114        AppQueueManager.set_stop_flag(task_id, InvokeFrom.SERVICE_API, end_user.id)115 116        return {"result": "success"}117 118 119class WorkflowAppLogApi(Resource):120    @validate_app_token121    @marshal_with(workflow_app_log_pagination_fields)122    def get(self, app_model: App):123        """124        Get workflow app logs125        """126        parser = reqparse.RequestParser()127        parser.add_argument("keyword", type=str, location="args")128        parser.add_argument("status", type=str, choices=["succeeded", "failed", "stopped"], location="args")129        parser.add_argument("page", type=int_range(1, 99999), default=1, location="args")130        parser.add_argument("limit", type=int_range(1, 100), default=20, location="args")131        args = parser.parse_args()132 133        # get paginate workflow app logs134        workflow_app_service = WorkflowAppService()135        workflow_app_log_pagination = workflow_app_service.get_paginate_workflow_app_logs(136            app_model=app_model, args=args137        )138 139        return workflow_app_log_pagination140 141 142api.add_resource(WorkflowRunApi, "/workflows/run")143api.add_resource(WorkflowRunDetailApi, "/workflows/run/<string:workflow_id>")144api.add_resource(WorkflowTaskStopApi, "/workflows/tasks/<string:task_id>/stop")145api.add_resource(WorkflowAppLogApi, "/workflows/logs")146