dify/api/controllers/openapi/app_dsl.py

from __future__ import annotations

from http import HTTPStatus
from typing import Final
from uuid import UUID, uuid4

from flask_restx import Resource
from werkzeug.exceptions import BadRequest, Forbidden, NotFound

from constants.oauth_bearer import Scope
from controllers.common.rbac import RBACPermission
from controllers.openapi import openapi_ns
from controllers.openapi._contract import Example, Kind, endpoint, op_of
from controllers.openapi._errors import DraftChanged, DraftNotFound, DslInvalid, ErrorDetail, VersionNotFound
from controllers.openapi._models import (
    AppDslExportQuery,
    AppDslExportResponse,
    AppDslImportPayload,
    AppDslImportResponse,
    CheckDependenciesResponse,
    DslCheckPayload,
    DslCheckResponse,
    DslIssueRow,
    Hint,
)
from controllers.openapi.auth.context import Context
from controllers.openapi.auth.requirements import EDITOR_ROLES, account_app_guards, workspace_write_guards
from controllers.openapi.plugins import PluginInstallApi
from core.plugin.entities.plugin import PluginDependencyType
from extensions.ext_application_services import application_services
from machinery.context import RequestContext
from services.app.console_service import ConsoleAppNotFoundError
from services.app_dsl_service import AppDslService
from services.entities.dsl_entities import AppImportParams, Import, ImportMode, ImportStatus
from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
from services.errors.base import NoPermissionError
from services.workflow.dsl_import import (
    DraftChangedError,
    DslNotCheckableError,
    DslRefusedError,
    YamlUrlFetchError,
    check_dsl,
    prepare_import,
)
from services.workflow.graph_check import IssueSeverity
from services.workflow.graph_diff import draft_token
from services.workflow_service import WorkflowService

_DSL_READ_GUARDS: Final = account_app_guards(
    RBACPermission.APP_IMPORT_EXPORT_DSL, scope=Scope.APPS_READ, roles=EDITOR_ROLES
)
_DSL_WRITE_GUARDS: Final = workspace_write_guards(RBACPermission.APP_IMPORT_EXPORT_DSL, roles=EDITOR_ROLES)


def _check_hint(context: RequestContext, app_id: str | None) -> Hint:
    return Hint(
        summary="See every problem, including warnings",
        op=op_of(AppDslCheckApi.post),
        input={"workspace_id": context.active_workspace_id, "yaml_content": None, "app_id": app_id},
    )


def _prepared_yaml(context: RequestContext, body: AppDslImportPayload) -> str:
    try:
        return prepare_import(
            context,
            mode=ImportMode(body.mode),
            source=body.source,
            app_id=body.app_id,
            draft_hash=body.draft_hash,
        )
    except ConsoleAppNotFoundError as error:
        raise NotFound(str(error)) from error
    except DraftChangedError as error:
        raise DraftChanged() from error
    except DslRefusedError as error:
        raise DslInvalid(
            details=[ErrorDetail(type=i.code, loc=list(i.loc), msg=i.message) for i in error.issues],
            hints=[_check_hint(context, body.app_id)],
        ) from error


def _import_response(result: Import, *, workspace_id: str) -> AppDslImportResponse:
    hints: list[Hint] = []
    if result.status == ImportStatus.PENDING:
        hints.append(
            Hint(
                summary="Confirm the pending import",
                op=op_of(AppDslImportConfirmApi.post),
                input={"workspace_id": workspace_id, "import_id": result.id},
            )
        )
    return AppDslImportResponse(**result.model_dump(), hints=hints)


@openapi_ns.route("/workspaces/<string:workspace_id>/apps/imports")
class AppDslImportApi(Resource):
    """Import a DSL YAML string into the specified workspace.

    Use ``mode=yaml-content`` with ``yaml_content`` for inline YAML, or
    ``mode=yaml-url`` with ``yaml_url`` for a remote URL.  Provide ``app_id``
    to overwrite an existing workflow or advanced-chat app; omit it to create
    a new app.

    Returns 202 when the DSL version requires an explicit confirmation step
    (major version mismatch).  Callers must then POST to the imports :confirm method.
    Returns 400 with the ``Import`` body when the import failed (an unreachable URL,
    unparsable DSL or a business error), 409 ``draft_changed`` when ``draft_hash`` no
    longer matches the draft, and 422 ``dsl_invalid`` when the graph has errors.
    """

    @endpoint(
        account_context=True,
        op="import.console_app.dsl",
        kind=Kind.OBJECT,
        summary="Import an app from DSL text or URL",
        examples=(
            Example(
                title="Import an app from inline YAML",
                input={"mode": "yaml-content", "yaml_content": "<yaml text of the app DSL>"},
            ),
            Example(
                title="Import an app from a URL",
                input={"mode": "yaml-url", "yaml_url": "https://example.com/apps/release-summary.yml"},
            ),
            Example(
                title="Overwrite an existing workflow app from YAML",
                input={"mode": "yaml-content", "yaml_content": "<yaml text of the app DSL>", "app_id": "<app_id>"},
            ),
        ),
        requirements=_DSL_WRITE_GUARDS,
        body=AppDslImportPayload,
        returns=(
            (HTTPStatus.OK, AppDslImportResponse, "Import completed"),
            (HTTPStatus.ACCEPTED, AppDslImportResponse, "Import pending confirmation"),
            (HTTPStatus.BAD_REQUEST, AppDslImportResponse, "Import failed"),
        ),
    )
    def post(self, ctx: RequestContext, workspace_id: str, *, body: AppDslImportPayload):
        try:
            yaml_content = _prepared_yaml(ctx, body)
        except YamlUrlFetchError as error:
            failed = Import(id=str(uuid4()), status=ImportStatus.FAILED, error=str(error))
            return _import_response(failed, workspace_id=workspace_id), HTTPStatus.BAD_REQUEST
        params = AppImportParams.model_validate(
            body.model_dump(exclude={"draft_hash"})
            | {"mode": ImportMode.YAML_CONTENT, "yaml_content": yaml_content, "yaml_url": None}
        )
        try:
            result = application_services().apps.imports.import_app(ctx, params)
        except NoPermissionError as exc:
            raise Forbidden(str(exc)) from exc

        response = _import_response(result, workspace_id=workspace_id)
        match result.status:
            case ImportStatus.FAILED:
                return response, HTTPStatus.BAD_REQUEST
            case ImportStatus.PENDING:
                return response, HTTPStatus.ACCEPTED
            case _:
                return response, HTTPStatus.OK


@openapi_ns.route("/workspaces/<string:workspace_id>/apps/imports:check")
class AppDslCheckApi(Resource):
    """Check a DSL without importing it.

    Errors are what the import refuses; warnings (missing credentials or models) are what the release check refuses.
    Pass ``app_id`` to check against an existing app's mode, as an overwrite import would.
    """

    @endpoint(
        account_context=True,
        op="check.console_app.dsl",
        kind=Kind.OBJECT,
        summary="Check a DSL without importing it: every problem with its node and field",
        examples=(Example(title="Check a DSL (pipe the YAML)", input={"yaml_content": "<yaml text of the app DSL>"}),),
        requirements=_DSL_WRITE_GUARDS,
        body=DslCheckPayload,
        returns=(HTTPStatus.OK, DslCheckResponse, "Check result"),
    )
    def post(self, ctx: RequestContext, workspace_id: str, *, body: DslCheckPayload):
        try:
            issues = check_dsl(ctx, body.yaml_content, body.app_id)
        except DslNotCheckableError as error:
            raise BadRequest(str(error)) from error
        except ConsoleAppNotFoundError as error:
            raise NotFound(str(error)) from error
        response = DslCheckResponse(
            valid=not any(i.code.severity is IssueSeverity.ERROR for i in issues),
            issues=[DslIssueRow.of(i) for i in issues],
        )
        return response, HTTPStatus.OK


@openapi_ns.route("/workspaces/<string:workspace_id>/apps/imports/<string:import_id>:confirm")
class AppDslImportConfirmApi(Resource):
    """Confirm a pending DSL import identified by ``import_id``.

    Required only when the initial import returned 202 (major DSL version
    mismatch that requires explicit acknowledgement).  The pending state is
    stored in Redis for 10 minutes; this call retrieves it and completes the
    import under the given workspace.

    Returns 400 when the pending data has expired or the import fails.
    """

    @endpoint(
        account_context=True,
        op="confirm.console_app.dsl_import",
        kind=Kind.OBJECT,
        summary="Confirm a pending DSL import",
        examples=(Example(title="Confirm an import that is pending confirmation", input={"import_id": "<import_id>"}),),
        requirements=_DSL_WRITE_GUARDS,
        returns=((HTTPStatus.OK, Import, "Import confirmed"), (HTTPStatus.BAD_REQUEST, Import, "Import failed")),
    )
    def post(self, ctx: RequestContext, workspace_id: str, import_id: str):
        try:
            result = application_services().apps.imports.confirm_import(ctx, import_id)
        except NoPermissionError as exc:
            raise Forbidden(str(exc)) from exc

        if result.status == ImportStatus.FAILED:
            return result, HTTPStatus.BAD_REQUEST
        return result, HTTPStatus.OK


@openapi_ns.route("/apps/<string:app_id>/dsl")
class AppDslExportApi(Resource):
    """Export an app's current draft configuration as a DSL YAML string.

    The auth pipeline resolves the app and its tenant from ``app_id``.  Pass
    ``include_secret=true`` to embed secret environment variable values in
    plaintext and keep tool credential ids; omit it to produce a DSL safe to share.

    Note: the pipeline enforces ``app.enable_api`` for all ``/apps/<app_id>``
    routes in the openapi group.  Apps with the service API disabled will
    receive a 403; enable the API in the console first if needed.
    """

    @endpoint(
        op="export.console_app.dsl",
        kind=Kind.OBJECT,
        summary="Export app DSL as YAML text inside a JSON object",
        examples=(
            Example(title="Export the current draft as YAML", input={"app_id": "<app_id>"}),
            Example(
                title="Export a published version including secrets",
                input={"app_id": "<app_id>", "workflow_id": "<workflow_id>", "include_secret": True},
            ),
        ),
        requirements=_DSL_READ_GUARDS,
        query=AppDslExportQuery,
        returns=(200, AppDslExportResponse, "Export successful"),
    )
    def get(self, ctx: Context, app_id: str, *, query: AppDslExportQuery):
        draft = WorkflowService().get_draft_workflow(app_model=ctx.app, session=ctx.session)
        try:
            data = AppDslService.export_dsl(
                app_model=ctx.app,
                session=ctx.session,
                include_secret=query.include_secret,
                workflow_id=query.workflow_id,
            )
        except WorkflowNotFoundError as exc:
            raise (VersionNotFound() if query.workflow_id else DraftNotFound()) from exc
        except IsDraftWorkflowError as exc:
            raise VersionNotFound() from exc
        return AppDslExportResponse(data=data, draft_hash=draft_token(draft) if draft else None), 200


@openapi_ns.route("/apps/<string:app_id>/dependencies:check")
class AppDslCheckDependenciesApi(Resource):
    """Check for leaked plugin dependencies after a DSL import.

    Call this after an import that reported ``COMPLETED_WITH_WARNINGS`` to
    find which plugin dependencies referenced in the DSL are not yet installed
    in the workspace.  Returns an empty ``leaked_dependencies`` list when all
    dependencies are satisfied.
    """

    @endpoint(
        account_context=True,
        op="check.console_app.dependency",
        kind=Kind.OBJECT,
        summary="Check plugin dependencies of an app",
        examples=(Example(title="Check which plugins an app needs", input={"app_id": "<app_id>"}),),
        requirements=_DSL_READ_GUARDS,
        returns=(HTTPStatus.OK, CheckDependenciesResponse, "Dependencies checked"),
    )
    def get(self, ctx: RequestContext, app_id: str):
        result = application_services().apps.imports.check_dependencies(ctx, str(UUID(app_id)))
        missing = [
            dependency.value.plugin_unique_identifier
            for dependency in result.leaked_dependencies
            if dependency.type == PluginDependencyType.Marketplace
        ]
        hints = (
            [
                Hint(
                    summary="Install the missing marketplace plugins",
                    op=op_of(PluginInstallApi.post),
                    input={"workspace_id": ctx.active_workspace_id, "identifiers": missing},
                )
            ]
            if missing
            else []
        )
        return CheckDependenciesResponse(**result.model_dump(), hints=hints), HTTPStatus.OK