import uuid from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqparse from sqlalchemy import select from sqlalchemy.orm import Session from werkzeug.exceptions import BadRequest, abort from controllers.console import console_ns from controllers.console.app.wraps import get_app_model from controllers.console.wraps import ( account_initialization_required, cloud_edition_billing_resource_check, edit_permission_required, enterprise_license_required, is_admin_or_owner_required, setup_required, ) from core.ops.ops_trace_manager import OpsTraceManager from core.workflow.enums import NodeType from extensions.ext_database import db from fields.app_fields import ( deleted_tool_fields, model_config_fields, model_config_partial_fields, site_fields, tag_fields, ) from fields.workflow_fields import workflow_partial_fields as _workflow_partial_fields_dict from libs.helper import AppIconUrlField, TimestampField from libs.login import current_account_with_tenant, login_required from libs.validators import validate_description_length from models import App, Workflow from services.app_dsl_service import AppDslService, ImportMode from services.app_service import AppService from services.enterprise.enterprise_service import EnterpriseService from services.feature_service import FeatureService ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "completion"] # Register models for flask_restx to avoid dict type issues in Swagger # Register base models first tag_model = console_ns.model("Tag", tag_fields) workflow_partial_model = console_ns.model("WorkflowPartial", _workflow_partial_fields_dict) model_config_model = console_ns.model("ModelConfig", model_config_fields) model_config_partial_model = console_ns.model("ModelConfigPartial", model_config_partial_fields) deleted_tool_model = console_ns.model("DeletedTool", deleted_tool_fields) site_model = console_ns.model("Site", site_fields) app_partial_model = console_ns.model( "AppPartial", { "id": fields.String, "name": fields.String, "max_active_requests": fields.Raw(), "description": fields.String(attribute="desc_or_prompt"), "mode": fields.String(attribute="mode_compatible_with_agent"), "icon_type": fields.String, "icon": fields.String, "icon_background": fields.String, "icon_url": AppIconUrlField, "model_config": fields.Nested(model_config_partial_model, attribute="app_model_config", allow_null=True), "workflow": fields.Nested(workflow_partial_model, allow_null=True), "use_icon_as_answer_icon": fields.Boolean, "created_by": fields.String, "created_at": TimestampField, "updated_by": fields.String, "updated_at": TimestampField, "tags": fields.List(fields.Nested(tag_model)), "access_mode": fields.String, "create_user_name": fields.String, "author_name": fields.String, "has_draft_trigger": fields.Boolean, }, ) app_detail_model = console_ns.model( "AppDetail", { "id": fields.String, "name": fields.String, "description": fields.String, "mode": fields.String(attribute="mode_compatible_with_agent"), "icon": fields.String, "icon_background": fields.String, "enable_site": fields.Boolean, "enable_api": fields.Boolean, "model_config": fields.Nested(model_config_model, attribute="app_model_config", allow_null=True), "workflow": fields.Nested(workflow_partial_model, allow_null=True), "tracing": fields.Raw, "use_icon_as_answer_icon": fields.Boolean, "created_by": fields.String, "created_at": TimestampField, "updated_by": fields.String, "updated_at": TimestampField, "access_mode": fields.String, "tags": fields.List(fields.Nested(tag_model)), }, ) app_detail_with_site_model = console_ns.model( "AppDetailWithSite", { "id": fields.String, "name": fields.String, "description": fields.String, "mode": fields.String(attribute="mode_compatible_with_agent"), "icon_type": fields.String, "icon": fields.String, "icon_background": fields.String, "icon_url": AppIconUrlField, "enable_site": fields.Boolean, "enable_api": fields.Boolean, "model_config": fields.Nested(model_config_model, attribute="app_model_config", allow_null=True), "workflow": fields.Nested(workflow_partial_model, allow_null=True), "api_base_url": fields.String, "use_icon_as_answer_icon": fields.Boolean, "max_active_requests": fields.Integer, "created_by": fields.String, "created_at": TimestampField, "updated_by": fields.String, "updated_at": TimestampField, "deleted_tools": fields.List(fields.Nested(deleted_tool_model)), "access_mode": fields.String, "tags": fields.List(fields.Nested(tag_model)), "site": fields.Nested(site_model), }, ) app_pagination_model = console_ns.model( "AppPagination", { "page": fields.Integer, "limit": fields.Integer(attribute="per_page"), "total": fields.Integer, "has_more": fields.Boolean(attribute="has_next"), "data": fields.List(fields.Nested(app_partial_model), attribute="items"), }, ) @console_ns.route("/apps") class AppListApi(Resource): @console_ns.doc("list_apps") @console_ns.doc(description="Get list of applications with pagination and filtering") @console_ns.expect( console_ns.parser() .add_argument("page", type=int, location="args", help="Page number (1-99999)", default=1) .add_argument("limit", type=int, location="args", help="Page size (1-100)", default=20) .add_argument( "mode", type=str, location="args", choices=["completion", "chat", "advanced-chat", "workflow", "agent-chat", "channel", "all"], default="all", help="App mode filter", ) .add_argument("name", type=str, location="args", help="Filter by app name") .add_argument("tag_ids", type=str, location="args", help="Comma-separated tag IDs") .add_argument("is_created_by_me", type=bool, location="args", help="Filter by creator") ) @console_ns.response(200, "Success", app_pagination_model) @setup_required @login_required @account_initialization_required @enterprise_license_required def get(self): """Get app list""" current_user, current_tenant_id = current_account_with_tenant() def uuid_list(value): try: return [str(uuid.UUID(v)) for v in value.split(",")] except ValueError: abort(400, message="Invalid UUID format in tag_ids.") parser = ( reqparse.RequestParser() .add_argument("page", type=inputs.int_range(1, 99999), required=False, default=1, location="args") .add_argument("limit", type=inputs.int_range(1, 100), required=False, default=20, location="args") .add_argument( "mode", type=str, choices=[ "completion", "chat", "advanced-chat", "workflow", "agent-chat", "channel", "all", ], default="all", location="args", required=False, ) .add_argument("name", type=str, location="args", required=False) .add_argument("tag_ids", type=uuid_list, location="args", required=False) .add_argument("is_created_by_me", type=inputs.boolean, location="args", required=False) ) args = parser.parse_args() # get app list app_service = AppService() app_pagination = app_service.get_paginate_apps(current_user.id, current_tenant_id, args) if not app_pagination: return {"data": [], "total": 0, "page": 1, "limit": 20, "has_more": False} if FeatureService.get_system_features().webapp_auth.enabled: app_ids = [str(app.id) for app in app_pagination.items] res = EnterpriseService.WebAppAuth.batch_get_app_access_mode_by_id(app_ids=app_ids) if len(res) != len(app_ids): raise BadRequest("Invalid app id in webapp auth") for app in app_pagination.items: if str(app.id) in res: app.access_mode = res[str(app.id)].access_mode workflow_capable_app_ids = [ str(app.id) for app in app_pagination.items if app.mode in {"workflow", "advanced-chat"} ] draft_trigger_app_ids: set[str] = set() if workflow_capable_app_ids: draft_workflows = ( db.session.execute( select(Workflow).where( Workflow.version == Workflow.VERSION_DRAFT, Workflow.app_id.in_(workflow_capable_app_ids), ) ) .scalars() .all() ) trigger_node_types = { NodeType.TRIGGER_WEBHOOK, NodeType.TRIGGER_SCHEDULE, NodeType.TRIGGER_PLUGIN, } for workflow in draft_workflows: for _, node_data in workflow.walk_nodes(): if node_data.get("type") in trigger_node_types: draft_trigger_app_ids.add(str(workflow.app_id)) break for app in app_pagination.items: app.has_draft_trigger = str(app.id) in draft_trigger_app_ids return marshal(app_pagination, app_pagination_model), 200 @console_ns.doc("create_app") @console_ns.doc(description="Create a new application") @console_ns.expect( console_ns.model( "CreateAppRequest", { "name": fields.String(required=True, description="App name"), "description": fields.String(description="App description (max 400 chars)"), "mode": fields.String(required=True, enum=ALLOW_CREATE_APP_MODES, description="App mode"), "icon_type": fields.String(description="Icon type"), "icon": fields.String(description="Icon"), "icon_background": fields.String(description="Icon background color"), }, ) ) @console_ns.response(201, "App created successfully", app_detail_model) @console_ns.response(403, "Insufficient permissions") @console_ns.response(400, "Invalid request parameters") @setup_required @login_required @account_initialization_required @marshal_with(app_detail_model) @cloud_edition_billing_resource_check("apps") @edit_permission_required def post(self): """Create app""" current_user, current_tenant_id = current_account_with_tenant() parser = ( reqparse.RequestParser() .add_argument("name", type=str, required=True, location="json") .add_argument("description", type=validate_description_length, location="json") .add_argument("mode", type=str, choices=ALLOW_CREATE_APP_MODES, location="json") .add_argument("icon_type", type=str, location="json") .add_argument("icon", type=str, location="json") .add_argument("icon_background", type=str, location="json") ) args = parser.parse_args() if "mode" not in args or args["mode"] is None: raise BadRequest("mode is required") app_service = AppService() app = app_service.create_app(current_tenant_id, args, current_user) return app, 201 @console_ns.route("/apps/") class AppApi(Resource): @console_ns.doc("get_app_detail") @console_ns.doc(description="Get application details") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.response(200, "Success", app_detail_with_site_model) @setup_required @login_required @account_initialization_required @enterprise_license_required @get_app_model @marshal_with(app_detail_with_site_model) def get(self, app_model): """Get app detail""" app_service = AppService() app_model = app_service.get_app(app_model) if FeatureService.get_system_features().webapp_auth.enabled: app_setting = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=str(app_model.id)) app_model.access_mode = app_setting.access_mode return app_model @console_ns.doc("update_app") @console_ns.doc(description="Update application details") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.expect( console_ns.model( "UpdateAppRequest", { "name": fields.String(required=True, description="App name"), "description": fields.String(description="App description (max 400 chars)"), "icon_type": fields.String(description="Icon type"), "icon": fields.String(description="Icon"), "icon_background": fields.String(description="Icon background color"), "use_icon_as_answer_icon": fields.Boolean(description="Use icon as answer icon"), "max_active_requests": fields.Integer(description="Maximum active requests"), }, ) ) @console_ns.response(200, "App updated successfully", app_detail_with_site_model) @console_ns.response(403, "Insufficient permissions") @console_ns.response(400, "Invalid request parameters") @setup_required @login_required @account_initialization_required @get_app_model @edit_permission_required @marshal_with(app_detail_with_site_model) def put(self, app_model): """Update app""" parser = ( reqparse.RequestParser() .add_argument("name", type=str, required=True, nullable=False, location="json") .add_argument("description", type=validate_description_length, location="json") .add_argument("icon_type", type=str, location="json") .add_argument("icon", type=str, location="json") .add_argument("icon_background", type=str, location="json") .add_argument("use_icon_as_answer_icon", type=bool, location="json") .add_argument("max_active_requests", type=int, location="json") ) args = parser.parse_args() app_service = AppService() args_dict: AppService.ArgsDict = { "name": args["name"], "description": args.get("description", ""), "icon_type": args.get("icon_type", ""), "icon": args.get("icon", ""), "icon_background": args.get("icon_background", ""), "use_icon_as_answer_icon": args.get("use_icon_as_answer_icon", False), "max_active_requests": args.get("max_active_requests", 0), } app_model = app_service.update_app(app_model, args_dict) return app_model @console_ns.doc("delete_app") @console_ns.doc(description="Delete application") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.response(204, "App deleted successfully") @console_ns.response(403, "Insufficient permissions") @get_app_model @setup_required @login_required @account_initialization_required @edit_permission_required def delete(self, app_model): """Delete app""" app_service = AppService() app_service.delete_app(app_model) return {"result": "success"}, 204 @console_ns.route("/apps//copy") class AppCopyApi(Resource): @console_ns.doc("copy_app") @console_ns.doc(description="Create a copy of an existing application") @console_ns.doc(params={"app_id": "Application ID to copy"}) @console_ns.expect( console_ns.model( "CopyAppRequest", { "name": fields.String(description="Name for the copied app"), "description": fields.String(description="Description for the copied app"), "icon_type": fields.String(description="Icon type"), "icon": fields.String(description="Icon"), "icon_background": fields.String(description="Icon background color"), }, ) ) @console_ns.response(201, "App copied successfully", app_detail_with_site_model) @console_ns.response(403, "Insufficient permissions") @setup_required @login_required @account_initialization_required @get_app_model @edit_permission_required @marshal_with(app_detail_with_site_model) def post(self, app_model): """Copy app""" # The role of the current user in the ta table must be admin, owner, or editor current_user, _ = current_account_with_tenant() parser = ( reqparse.RequestParser() .add_argument("name", type=str, location="json") .add_argument("description", type=validate_description_length, location="json") .add_argument("icon_type", type=str, location="json") .add_argument("icon", type=str, location="json") .add_argument("icon_background", type=str, location="json") ) args = parser.parse_args() with Session(db.engine) as session: import_service = AppDslService(session) yaml_content = import_service.export_dsl(app_model=app_model, include_secret=True) result = import_service.import_app( account=current_user, import_mode=ImportMode.YAML_CONTENT, yaml_content=yaml_content, name=args.get("name"), description=args.get("description"), icon_type=args.get("icon_type"), icon=args.get("icon"), icon_background=args.get("icon_background"), ) session.commit() stmt = select(App).where(App.id == result.app_id) app = session.scalar(stmt) return app, 201 @console_ns.route("/apps//export") class AppExportApi(Resource): @console_ns.doc("export_app") @console_ns.doc(description="Export application configuration as DSL") @console_ns.doc(params={"app_id": "Application ID to export"}) @console_ns.expect( console_ns.parser() .add_argument("include_secret", type=bool, location="args", default=False, help="Include secrets in export") .add_argument("workflow_id", type=str, location="args", help="Specific workflow ID to export") ) @console_ns.response( 200, "App exported successfully", console_ns.model("AppExportResponse", {"data": fields.String(description="DSL export data")}), ) @console_ns.response(403, "Insufficient permissions") @get_app_model @setup_required @login_required @account_initialization_required @edit_permission_required def get(self, app_model): """Export app""" # Add include_secret params parser = ( reqparse.RequestParser() .add_argument("include_secret", type=inputs.boolean, default=False, location="args") .add_argument("workflow_id", type=str, location="args") ) args = parser.parse_args() return { "data": AppDslService.export_dsl( app_model=app_model, include_secret=args["include_secret"], workflow_id=args.get("workflow_id") ) } parser = reqparse.RequestParser().add_argument("name", type=str, required=True, location="json", help="Name to check") @console_ns.route("/apps//name") class AppNameApi(Resource): @console_ns.doc("check_app_name") @console_ns.doc(description="Check if app name is available") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.expect(parser) @console_ns.response(200, "Name availability checked") @setup_required @login_required @account_initialization_required @get_app_model @marshal_with(app_detail_model) @edit_permission_required def post(self, app_model): args = parser.parse_args() app_service = AppService() app_model = app_service.update_app_name(app_model, args["name"]) return app_model @console_ns.route("/apps//icon") class AppIconApi(Resource): @console_ns.doc("update_app_icon") @console_ns.doc(description="Update application icon") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.expect( console_ns.model( "AppIconRequest", { "icon": fields.String(required=True, description="Icon data"), "icon_type": fields.String(description="Icon type"), "icon_background": fields.String(description="Icon background color"), }, ) ) @console_ns.response(200, "Icon updated successfully") @console_ns.response(403, "Insufficient permissions") @setup_required @login_required @account_initialization_required @get_app_model @marshal_with(app_detail_model) @edit_permission_required def post(self, app_model): parser = ( reqparse.RequestParser() .add_argument("icon", type=str, location="json") .add_argument("icon_background", type=str, location="json") ) args = parser.parse_args() app_service = AppService() app_model = app_service.update_app_icon(app_model, args.get("icon") or "", args.get("icon_background") or "") return app_model @console_ns.route("/apps//site-enable") class AppSiteStatus(Resource): @console_ns.doc("update_app_site_status") @console_ns.doc(description="Enable or disable app site") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.expect( console_ns.model( "AppSiteStatusRequest", {"enable_site": fields.Boolean(required=True, description="Enable or disable site")} ) ) @console_ns.response(200, "Site status updated successfully", app_detail_model) @console_ns.response(403, "Insufficient permissions") @setup_required @login_required @account_initialization_required @get_app_model @marshal_with(app_detail_model) @edit_permission_required def post(self, app_model): parser = reqparse.RequestParser().add_argument("enable_site", type=bool, required=True, location="json") args = parser.parse_args() app_service = AppService() app_model = app_service.update_app_site_status(app_model, args["enable_site"]) return app_model @console_ns.route("/apps//api-enable") class AppApiStatus(Resource): @console_ns.doc("update_app_api_status") @console_ns.doc(description="Enable or disable app API") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.expect( console_ns.model( "AppApiStatusRequest", {"enable_api": fields.Boolean(required=True, description="Enable or disable API")} ) ) @console_ns.response(200, "API status updated successfully", app_detail_model) @console_ns.response(403, "Insufficient permissions") @setup_required @login_required @is_admin_or_owner_required @account_initialization_required @get_app_model @marshal_with(app_detail_model) def post(self, app_model): parser = reqparse.RequestParser().add_argument("enable_api", type=bool, required=True, location="json") args = parser.parse_args() app_service = AppService() app_model = app_service.update_app_api_status(app_model, args["enable_api"]) return app_model @console_ns.route("/apps//trace") class AppTraceApi(Resource): @console_ns.doc("get_app_trace") @console_ns.doc(description="Get app tracing configuration") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.response(200, "Trace configuration retrieved successfully") @setup_required @login_required @account_initialization_required def get(self, app_id): """Get app trace""" app_trace_config = OpsTraceManager.get_app_tracing_config(app_id=app_id) return app_trace_config @console_ns.doc("update_app_trace") @console_ns.doc(description="Update app tracing configuration") @console_ns.doc(params={"app_id": "Application ID"}) @console_ns.expect( console_ns.model( "AppTraceRequest", { "enabled": fields.Boolean(required=True, description="Enable or disable tracing"), "tracing_provider": fields.String(required=True, description="Tracing provider"), }, ) ) @console_ns.response(200, "Trace configuration updated successfully") @console_ns.response(403, "Insufficient permissions") @setup_required @login_required @account_initialization_required @edit_permission_required def post(self, app_id): # add app trace parser = ( reqparse.RequestParser() .add_argument("enabled", type=bool, required=True, location="json") .add_argument("tracing_provider", type=str, required=True, location="json") ) args = parser.parse_args() OpsTraceManager.update_app_tracing_config( app_id=app_id, enabled=args["enabled"], tracing_provider=args["tracing_provider"], ) return {"result": "success"}