Merge branch 'feat/queue-based-graph-engine' into feat/rag-2

Signed-off-by: -LAN- <laipz8200@outlook.com>

.github/workflows/autofix.yml

@@ -20,7 +20,7 @@ jobs:
           cd api
           uv sync --dev
           # Fix lint errors
-          uv run ruff check --fix .
+          uv run ruff check --fix-only .
           # Format code
           uv run ruff format .
       - name: ast-grep

.github/workflows/build-push.yml

@@ -8,6 +8,8 @@ on:
       - "deploy/enterprise"
       - "build/**"
       - "release/e-*"
+      - "deploy/rag-dev"
+      - "feat/rag-2"
     tags:
       - "*"
 

.github/workflows/deploy-dev.yml

@@ -4,7 +4,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "deploy/dev"
+      - "deploy/rag-dev"
     types:
       - completed
 
@@ -12,12 +12,13 @@ jobs:
   deploy:
     runs-on: ubuntu-latest
     if: |
-      github.event.workflow_run.conclusion == 'success'
+      github.event.workflow_run.conclusion == 'success' &&
+      github.event.workflow_run.head_branch == 'deploy/rag-dev'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v0.1.8
         with:
-          host: ${{ secrets.SSH_HOST }}
+          host: ${{ secrets.RAG_SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.github/workflows/deploy-enterprise.yml

@@ -19,23 +19,11 @@ jobs:
       github.event.workflow_run.head_branch == 'deploy/enterprise'
 
     steps:
-      - name: trigger deployments
-        env:
-          DEV_ENV_ADDRS: ${{ vars.DEV_ENV_ADDRS }}
-          DEPLOY_SECRET: ${{ secrets.DEPLOY_SECRET }}
-        run: |
-          IFS=',' read -ra ENDPOINTS <<< "${DEV_ENV_ADDRS:-}"
-          BODY='{"project":"dify-api","tag":"deploy-enterprise"}'
-
-          for ENDPOINT in "${ENDPOINTS[@]}"; do
-            ENDPOINT="$(echo "$ENDPOINT" | xargs)"
-            [ -z "$ENDPOINT" ] && continue
-
-            API_SIGNATURE=$(printf '%s' "$BODY" | openssl dgst -sha256 -hmac "$DEPLOY_SECRET" | awk '{print "sha256="$2}')
-
-            curl -sSf -X POST \
-              -H "Content-Type: application/json" \
-              -H "X-Hub-Signature-256: $API_SIGNATURE" \
-              -d "$BODY" \
-              "$ENDPOINT"
-          done
+      - name: Deploy to server
+        uses: appleboy/ssh-action@v0.1.8
+        with:
+          host: ${{ secrets.ENTERPRISE_SSH_HOST }}
+          username: ${{ secrets.ENTERPRISE_SSH_USER }}
+          password: ${{ secrets.ENTERPRISE_SSH_PASSWORD }}
+          script: |
+            ${{ vars.ENTERPRISE_SSH_SCRIPT || secrets.ENTERPRISE_SSH_SCRIPT }}

.github/workflows/style.yml

@@ -12,7 +12,6 @@ permissions:
   statuses: write
   contents: read
 
-
 jobs:
   python-style:
     name: Python Style

api/.env.example

@@ -461,6 +461,16 @@ WORKFLOW_CALL_MAX_DEPTH=5
 WORKFLOW_PARALLEL_DEPTH_LIMIT=3
 MAX_VARIABLE_SIZE=204800
 
+# GraphEngine Worker Pool Configuration
+# Minimum number of workers per GraphEngine instance (default: 1)
+GRAPH_ENGINE_MIN_WORKERS=1
+# Maximum number of workers per GraphEngine instance (default: 10)
+GRAPH_ENGINE_MAX_WORKERS=10
+# Queue depth threshold that triggers worker scale up (default: 3)
+GRAPH_ENGINE_SCALE_UP_THRESHOLD=3
+# Seconds of idle time before scaling down workers (default: 5.0)
+GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME=5.0
+
 # Workflow storage configuration
 # Options: rdbms, hybrid
 # rdbms: Use only the relational database (default)
@@ -569,7 +579,3 @@ QUEUE_MONITOR_INTERVAL=30
 # Swagger UI configuration
 SWAGGER_UI_ENABLED=true
 SWAGGER_UI_PATH=/swagger-ui.html
-
-# Whether to encrypt dataset IDs when exporting DSL files (default: true)
-# Set to false to export dataset IDs as plain text for easier cross-environment import
-DSL_EXPORT_ENCRYPT_DATASET_ID=true

api/.importlinter

@@ -0,0 +1,112 @@
+[importlinter]
+root_packages =
+    core
+    configs
+    controllers
+    models
+    tasks
+    services
+
+[importlinter:contract:workflow]
+name = Workflow
+type=layers
+layers =
+    graph_engine
+    graph_events
+    graph
+    nodes
+    node_events
+    entities
+containers =
+    core.workflow
+ignore_imports =
+    core.workflow.nodes.base.node -> core.workflow.graph_events
+    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_events
+    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine
+    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph
+    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine.command_channels
+    core.workflow.nodes.loop.loop_node -> core.workflow.graph_events
+    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine
+    core.workflow.nodes.loop.loop_node -> core.workflow.graph
+    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine.command_channels
+    core.workflow.nodes.node_factory -> core.workflow.graph
+
+[importlinter:contract:rsc]
+name = RSC
+type = layers
+layers =
+    graph_engine
+    response_coordinator
+containers =
+    core.workflow.graph_engine
+
+[importlinter:contract:worker]
+name = Worker
+type = layers
+layers =
+    graph_engine
+    worker
+containers =
+    core.workflow.graph_engine
+
+[importlinter:contract:graph-engine-architecture]
+name = Graph Engine Architecture
+type = layers
+layers =
+    graph_engine
+    orchestration
+    command_processing
+    event_management
+    error_handling
+    graph_traversal
+    state_management
+    worker_management
+    domain
+containers =
+    core.workflow.graph_engine
+
+[importlinter:contract:domain-isolation]
+name = Domain Model Isolation
+type = forbidden
+source_modules =
+    core.workflow.graph_engine.domain
+forbidden_modules =
+    core.workflow.graph_engine.worker_management
+    core.workflow.graph_engine.command_channels
+    core.workflow.graph_engine.layers
+    core.workflow.graph_engine.protocols
+
+[importlinter:contract:worker-management]
+name = Worker Management
+type = forbidden
+source_modules =
+    core.workflow.graph_engine.worker_management
+forbidden_modules =
+    core.workflow.graph_engine.orchestration
+    core.workflow.graph_engine.command_processing
+    core.workflow.graph_engine.event_management
+
+[importlinter:contract:error-handling-strategies]
+name = Error Handling Strategies
+type = independence
+modules =
+    core.workflow.graph_engine.error_handling.abort_strategy
+    core.workflow.graph_engine.error_handling.retry_strategy
+    core.workflow.graph_engine.error_handling.fail_branch_strategy
+    core.workflow.graph_engine.error_handling.default_value_strategy
+
+[importlinter:contract:graph-traversal-components]
+name = Graph Traversal Components
+type = layers
+layers =
+    edge_processor
+    skip_propagator
+containers =
+    core.workflow.graph_engine.graph_traversal
+
+[importlinter:contract:command-channels]
+name = Command Channels Independence
+type = independence
+modules =
+    core.workflow.graph_engine.command_channels.in_memory_channel
+    core.workflow.graph_engine.command_channels.redis_channel

api/.ruff.toml

@@ -45,7 +45,6 @@ select = [
     "G001", # don't use str format to logging messages
     "G003", # don't use + in logging messages
     "G004", # don't use f-strings to format logging messages
-    "UP042", # use StrEnum
 ]
 
 ignore = [

api/app.py

@@ -1,4 +1,3 @@
-import os
 import sys
 
 
@@ -17,20 +16,20 @@ else:
     # It seems that JetBrains Python debugger does not work well with gevent,
     # so we need to disable gevent in debug mode.
     # If you are using debugpy and set GEVENT_SUPPORT=True, you can debug with gevent.
-    if (flask_debug := os.environ.get("FLASK_DEBUG", "0")) and flask_debug.lower() in {"false", "0", "no"}:
-        from gevent import monkey
+    # if (flask_debug := os.environ.get("FLASK_DEBUG", "0")) and flask_debug.lower() in {"false", "0", "no"}:
+    # from gevent import monkey
+    #
+    # # gevent
+    # monkey.patch_all()
+    #
+    # from grpc.experimental import gevent as grpc_gevent  # type: ignore
+    #
+    # # grpc gevent
+    # grpc_gevent.init_gevent()
 
-        # gevent
-        monkey.patch_all()
-
-        from grpc.experimental import gevent as grpc_gevent  # type: ignore
-
-        # grpc gevent
-        grpc_gevent.init_gevent()
-
-        import psycogreen.gevent  # type: ignore
-
-        psycogreen.gevent.patch_psycopg()
+    # import psycogreen.gevent  # type: ignore
+    #
+    # psycogreen.gevent.patch_psycopg()
 
     from app_factory import create_app
 

api/celery_entrypoint.py

@@ -0,0 +1,22 @@
+import logging
+
+import psycogreen.gevent as pscycogreen_gevent  # type: ignore
+from grpc.experimental import gevent as grpc_gevent  # type: ignore
+
+_logger = logging.getLogger(__name__)
+
+
+def _log(message: str):
+    print(message, flush=True)
+
+
+# grpc gevent
+grpc_gevent.init_gevent()
+_log("gRPC  patched with gevent.")
+pscycogreen_gevent.patch_psycopg()
+_log("psycopg2 patched with gevent.")
+
+
+from app import app, celery
+
+__all__ = ["app", "celery"]

api/commands.py

@@ -13,11 +13,13 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from constants.languages import languages
-from core.plugin.entities.plugin import ToolProviderID
+from core.helper import encrypter
+from core.plugin.impl.plugin import PluginInstaller
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
 from core.rag.models.document import Document
+from core.tools.entities.tool_entities import CredentialType
 from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
@@ -30,12 +32,16 @@ from models import Tenant
 from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, DatasetMetadataBinding, DocumentSegment
 from models.dataset import Document as DatasetDocument
 from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation
+from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
 from models.provider import Provider, ProviderModel
+from models.provider_ids import DatasourceProviderID, ToolProviderID
+from models.source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
 from models.tools import ToolOAuthSystemClient
 from services.account_service import AccountService, RegisterService, TenantService
 from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpiredLogs
 from services.plugin.data_migration import PluginDataMigration
 from services.plugin.plugin_migration import PluginMigration
+from services.plugin.plugin_service import PluginService
 from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
 logger = logging.getLogger(__name__)
@@ -212,9 +218,7 @@ def migrate_annotation_vector_database():
                 if not dataset_collection_binding:
                     click.echo(f"App annotation collection binding not found: {app.id}")
                     continue
-                annotations = db.session.scalars(
-                    select(MessageAnnotation).where(MessageAnnotation.app_id == app.id)
-                ).all()
+                annotations = db.session.query(MessageAnnotation).where(MessageAnnotation.app_id == app.id).all()
                 dataset = Dataset(
                     id=app.id,
                     tenant_id=app.tenant_id,
@@ -369,25 +373,29 @@ def migrate_knowledge_vector_database():
                     )
                     raise e
 
-                dataset_documents = db.session.scalars(
-                    select(DatasetDocument).where(
+                dataset_documents = (
+                    db.session.query(DatasetDocument)
+                    .where(
                         DatasetDocument.dataset_id == dataset.id,
                         DatasetDocument.indexing_status == "completed",
                         DatasetDocument.enabled == True,
                         DatasetDocument.archived == False,
                     )
-                ).all()
+                    .all()
+                )
 
                 documents = []
                 segments_count = 0
                 for dataset_document in dataset_documents:
-                    segments = db.session.scalars(
-                        select(DocumentSegment).where(
+                    segments = (
+                        db.session.query(DocumentSegment)
+                        .where(
                             DocumentSegment.document_id == dataset_document.id,
                             DocumentSegment.status == "completed",
                             DocumentSegment.enabled == True,
                         )
-                    ).all()
+                        .all()
+                    )
 
                     for segment in segments:
                         document = Document(
@@ -509,7 +517,7 @@ def add_qdrant_index(field: str):
         from qdrant_client.http.exceptions import UnexpectedResponse
         from qdrant_client.http.models import PayloadSchemaType
 
-        from core.rag.datasource.vdb.qdrant.qdrant_vector import PathQdrantParams, QdrantConfig
+        from core.rag.datasource.vdb.qdrant.qdrant_vector import QdrantConfig
 
         for binding in bindings:
             if dify_config.QDRANT_URL is None:
@@ -523,21 +531,7 @@ def add_qdrant_index(field: str):
                 prefer_grpc=dify_config.QDRANT_GRPC_ENABLED,
             )
             try:
-                params = qdrant_config.to_qdrant_params()
-                # Check the type before using
-                if isinstance(params, PathQdrantParams):
-                    # PathQdrantParams case
-                    client = qdrant_client.QdrantClient(path=params.path)
-                else:
-                    # UrlQdrantParams case - params is UrlQdrantParams
-                    client = qdrant_client.QdrantClient(
-                        url=params.url,
-                        api_key=params.api_key,
-                        timeout=int(params.timeout),
-                        verify=params.verify,
-                        grpc_port=params.grpc_port,
-                        prefer_grpc=params.prefer_grpc,
-                    )
+                client = qdrant_client.QdrantClient(**qdrant_config.to_qdrant_params())
                 # create payload index
                 client.create_payload_index(binding.collection_name, field, field_schema=PayloadSchemaType.KEYWORD)
                 create_count += 1
@@ -1245,15 +1239,17 @@ def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
 
 def _count_orphaned_draft_variables() -> dict[str, Any]:
     """
-    Count orphaned draft variables by app.
+    Count orphaned draft variables by app, including associated file counts.
 
     Returns:
-        Dictionary with statistics about orphaned variables
+        Dictionary with statistics about orphaned variables and files
     """
-    query = """
+    # Count orphaned variables by app
+    variables_query = """
         SELECT
             wdv.app_id,
-            COUNT(*) as variable_count
+            COUNT(*) as variable_count,
+            COUNT(wdv.file_id) as file_count
         FROM workflow_draft_variables AS wdv
         WHERE NOT EXISTS(
             SELECT 1 FROM apps WHERE apps.id = wdv.app_id
@@ -1263,14 +1259,21 @@ def _count_orphaned_draft_variables() -> dict[str, Any]:
     """
 
     with db.engine.connect() as conn:
-        result = conn.execute(sa.text(query))
-        orphaned_by_app = {row[0]: row[1] for row in result}
+        result = conn.execute(sa.text(variables_query))
+        orphaned_by_app = {}
+        total_files = 0
 
-        total_orphaned = sum(orphaned_by_app.values())
+        for row in result:
+            app_id, variable_count, file_count = row
+            orphaned_by_app[app_id] = {"variables": variable_count, "files": file_count}
+            total_files += file_count
+
+        total_orphaned = sum(app_data["variables"] for app_data in orphaned_by_app.values())
         app_count = len(orphaned_by_app)
 
         return {
             "total_orphaned_variables": total_orphaned,
+            "total_orphaned_files": total_files,
             "orphaned_app_count": app_count,
             "orphaned_by_app": orphaned_by_app,
         }
@@ -1299,6 +1302,7 @@ def cleanup_orphaned_draft_variables(
     stats = _count_orphaned_draft_variables()
 
     logger.info("Found %s orphaned draft variables", stats["total_orphaned_variables"])
+    logger.info("Found %s associated offload files", stats["total_orphaned_files"])
     logger.info("Across %s non-existent apps", stats["orphaned_app_count"])
 
     if stats["total_orphaned_variables"] == 0:
@@ -1307,10 +1311,10 @@ def cleanup_orphaned_draft_variables(
 
     if dry_run:
         logger.info("DRY RUN: Would delete the following:")
-        for app_id, count in sorted(stats["orphaned_by_app"].items(), key=lambda x: x[1], reverse=True)[
+        for app_id, data in sorted(stats["orphaned_by_app"].items(), key=lambda x: x[1]["variables"], reverse=True)[
             :10
         ]:  # Show top 10
-            logger.info("  App %s: %s variables", app_id, count)
+            logger.info("  App %s: %s variables, %s files", app_id, data["variables"], data["files"])
         if len(stats["orphaned_by_app"]) > 10:
             logger.info("  ... and %s more apps", len(stats["orphaned_by_app"]) - 10)
         return
@@ -1319,7 +1323,8 @@ def cleanup_orphaned_draft_variables(
     if not force:
         click.confirm(
             f"Are you sure you want to delete {stats['total_orphaned_variables']} "
-            f"orphaned draft variables from {stats['orphaned_app_count']} apps?",
+            f"orphaned draft variables and {stats['total_orphaned_files']} associated files "
+            f"from {stats['orphaned_app_count']} apps?",
             abort=True,
         )
 
@@ -1352,3 +1357,231 @@ def cleanup_orphaned_draft_variables(
                 continue
 
     logger.info("Cleanup completed. Total deleted: %s variables across %s apps", total_deleted, processed_apps)
+
+
+@click.command("setup-datasource-oauth-client", help="Setup datasource oauth client.")
+@click.option("--provider", prompt=True, help="Provider name")
+@click.option("--client-params", prompt=True, help="Client Params")
+def setup_datasource_oauth_client(provider, client_params):
+    """
+    Setup datasource oauth client
+    """
+    provider_id = DatasourceProviderID(provider)
+    provider_name = provider_id.provider_name
+    plugin_id = provider_id.plugin_id
+
+    try:
+        # json validate
+        click.echo(click.style(f"Validating client params: {client_params}", fg="yellow"))
+        client_params_dict = TypeAdapter(dict[str, Any]).validate_json(client_params)
+        click.echo(click.style("Client params validated successfully.", fg="green"))
+    except Exception as e:
+        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
+        return
+
+    click.echo(click.style(f"Ready to delete existing oauth client params: {provider_name}", fg="yellow"))
+    deleted_count = (
+        db.session.query(DatasourceOauthParamConfig)
+        .filter_by(
+            provider=provider_name,
+            plugin_id=plugin_id,
+        )
+        .delete()
+    )
+    if deleted_count > 0:
+        click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
+
+    click.echo(click.style(f"Ready to setup datasource oauth client: {provider_name}", fg="yellow"))
+    oauth_client = DatasourceOauthParamConfig(
+        provider=provider_name,
+        plugin_id=plugin_id,
+        system_credentials=client_params_dict,
+    )
+    db.session.add(oauth_client)
+    db.session.commit()
+    click.echo(click.style(f"provider: {provider_name}", fg="green"))
+    click.echo(click.style(f"plugin_id: {plugin_id}", fg="green"))
+    click.echo(click.style(f"params: {json.dumps(client_params_dict, indent=2, ensure_ascii=False)}", fg="green"))
+    click.echo(click.style(f"Datasource oauth client setup successfully. id: {oauth_client.id}", fg="green"))
+
+
+@click.command("transform-datasource-credentials", help="Transform datasource credentials.")
+def transform_datasource_credentials():
+    """
+    Transform datasource credentials
+    """
+    try:
+        installer_manager = PluginInstaller()
+        plugin_migration = PluginMigration()
+
+        notion_plugin_id = "langgenius/notion_datasource"
+        firecrawl_plugin_id = "langgenius/firecrawl_datasource"
+        jina_plugin_id = "langgenius/jina_datasource"
+        notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)
+        firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)
+        jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)
+        oauth_credential_type = CredentialType.OAUTH2
+        api_key_credential_type = CredentialType.API_KEY
+
+        # deal notion credentials
+        deal_notion_count = 0
+        notion_credentials = db.session.query(DataSourceOauthBinding).filter_by(provider="notion").all()
+        if notion_credentials:
+            notion_credentials_tenant_mapping: dict[str, list[DataSourceOauthBinding]] = {}
+            for credential in notion_credentials:
+                tenant_id = credential.tenant_id
+                if tenant_id not in notion_credentials_tenant_mapping:
+                    notion_credentials_tenant_mapping[tenant_id] = []
+                notion_credentials_tenant_mapping[tenant_id].append(credential)
+            for tenant_id, credentials in notion_credentials_tenant_mapping.items():
+                # check notion plugin is installed
+                installed_plugins = installer_manager.list_plugins(tenant_id)
+                installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+                if notion_plugin_id not in installed_plugins_ids:
+                    if notion_plugin_unique_identifier:
+                        # install notion plugin
+                        PluginService.install_from_marketplace_pkg(tenant_id, [notion_plugin_unique_identifier])
+                auth_count = 0
+                for credential in credentials:
+                    auth_count += 1
+                    # get credential oauth params
+                    access_token = credential.access_token
+                    # notion info
+                    notion_info = credential.source_info
+                    workspace_id = notion_info.get("workspace_id")
+                    workspace_name = notion_info.get("workspace_name")
+                    workspace_icon = notion_info.get("workspace_icon")
+                    new_credentials = {
+                        "integration_secret": encrypter.encrypt_token(tenant_id, access_token),
+                        "workspace_id": workspace_id,
+                        "workspace_name": workspace_name,
+                        "workspace_icon": workspace_icon,
+                    }
+                    datasource_provider = DatasourceProvider(
+                        provider="notion_datasource",
+                        tenant_id=tenant_id,
+                        plugin_id=notion_plugin_id,
+                        auth_type=oauth_credential_type.value,
+                        encrypted_credentials=new_credentials,
+                        name=f"Auth {auth_count}",
+                        avatar_url=workspace_icon or "default",
+                        is_default=False,
+                    )
+                    db.session.add(datasource_provider)
+                    deal_notion_count += 1
+                db.session.commit()
+        # deal firecrawl credentials
+        deal_firecrawl_count = 0
+        firecrawl_credentials = db.session.query(DataSourceApiKeyAuthBinding).filter_by(provider="firecrawl").all()
+        if firecrawl_credentials:
+            firecrawl_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
+            for credential in firecrawl_credentials:
+                tenant_id = credential.tenant_id
+                if tenant_id not in firecrawl_credentials_tenant_mapping:
+                    firecrawl_credentials_tenant_mapping[tenant_id] = []
+                firecrawl_credentials_tenant_mapping[tenant_id].append(credential)
+            for tenant_id, credentials in firecrawl_credentials_tenant_mapping.items():
+                # check firecrawl plugin is installed
+                installed_plugins = installer_manager.list_plugins(tenant_id)
+                installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+                if firecrawl_plugin_id not in installed_plugins_ids:
+                    if firecrawl_plugin_unique_identifier:
+                        # install firecrawl plugin
+                        PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_plugin_unique_identifier])
+
+                auth_count = 0
+                for credential in credentials:
+                    auth_count += 1
+                    # get credential api key
+                    credentials_json = json.loads(credential.credentials)
+                    api_key = credentials_json.get("config", {}).get("api_key")
+                    base_url = credentials_json.get("config", {}).get("base_url")
+                    new_credentials = {
+                        "firecrawl_api_key": api_key,
+                        "base_url": base_url,
+                    }
+                    datasource_provider = DatasourceProvider(
+                        provider="firecrawl",
+                        tenant_id=tenant_id,
+                        plugin_id=firecrawl_plugin_id,
+                        auth_type=api_key_credential_type.value,
+                        encrypted_credentials=new_credentials,
+                        name=f"Auth {auth_count}",
+                        avatar_url="default",
+                        is_default=False,
+                    )
+                    db.session.add(datasource_provider)
+                    deal_firecrawl_count += 1
+                db.session.commit()
+        # deal jina credentials
+        deal_jina_count = 0
+        jina_credentials = db.session.query(DataSourceApiKeyAuthBinding).filter_by(provider="jinareader").all()
+        if jina_credentials:
+            jina_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
+            for credential in jina_credentials:
+                tenant_id = credential.tenant_id
+                if tenant_id not in jina_credentials_tenant_mapping:
+                    jina_credentials_tenant_mapping[tenant_id] = []
+                jina_credentials_tenant_mapping[tenant_id].append(credential)
+            for tenant_id, credentials in jina_credentials_tenant_mapping.items():
+                # check jina plugin is installed
+                installed_plugins = installer_manager.list_plugins(tenant_id)
+                installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+                if jina_plugin_id not in installed_plugins_ids:
+                    if jina_plugin_unique_identifier:
+                        # install jina plugin
+                        print(jina_plugin_unique_identifier)
+                        PluginService.install_from_marketplace_pkg(tenant_id, [jina_plugin_unique_identifier])
+
+                auth_count = 0
+                for credential in credentials:
+                    auth_count += 1
+                    # get credential api key
+                    credentials_json = json.loads(credential.credentials)
+                    api_key = credentials_json.get("config", {}).get("api_key")
+                    new_credentials = {
+                        "integration_secret": api_key,
+                    }
+                    datasource_provider = DatasourceProvider(
+                        provider="jina",
+                        tenant_id=tenant_id,
+                        plugin_id=jina_plugin_id,
+                        auth_type=api_key_credential_type.value,
+                        encrypted_credentials=new_credentials,
+                        name=f"Auth {auth_count}",
+                        avatar_url="default",
+                        is_default=False,
+                    )
+                    db.session.add(datasource_provider)
+                    deal_jina_count += 1
+                db.session.commit()
+    except Exception as e:
+        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
+        return
+    click.echo(click.style(f"Transforming notion successfully. deal_notion_count: {deal_notion_count}", fg="green"))
+    click.echo(
+        click.style(f"Transforming firecrawl successfully. deal_firecrawl_count: {deal_firecrawl_count}", fg="green")
+    )
+    click.echo(click.style(f"Transforming jina successfully. deal_jina_count: {deal_jina_count}", fg="green"))
+
+
+@click.command("install-rag-pipeline-plugins", help="Install rag pipeline plugins.")
+@click.option(
+    "--input_file", prompt=True, help="The file to store the extracted unique identifiers.", default="plugins.jsonl"
+)
+@click.option(
+    "--output_file", prompt=True, help="The file to store the installed plugins.", default="installed_plugins.jsonl"
+)
+@click.option("--workers", prompt=True, help="The number of workers to install plugins.", default=100)
+def install_rag_pipeline_plugins(input_file, output_file, workers):
+    """
+    Install rag pipeline plugins
+    """
+    click.echo(click.style("Installing rag pipeline plugins", fg="yellow"))
+    plugin_migration = PluginMigration()
+    plugin_migration.install_rag_pipeline_plugins(
+        input_file,
+        output_file,
+        workers,
+    )
+    click.echo(click.style("Installing rag pipeline plugins successfully", fg="green"))

api/configs/feature/__init__.py

@@ -499,6 +499,22 @@ class UpdateConfig(BaseSettings):
     )
 
 
+class WorkflowVariableTruncationConfig(BaseSettings):
+    WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE: PositiveInt = Field(
+        # 100KB
+        1024_000,
+        description="Maximum size for variable to trigger final truncation.",
+    )
+    WORKFLOW_VARIABLE_TRUNCATION_STRING_LENGTH: PositiveInt = Field(
+        50000,
+        description="maximum length for string to trigger tuncation, measure in number of characters",
+    )
+    WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH: PositiveInt = Field(
+        100,
+        description="maximum length for array to trigger truncation.",
+    )
+
+
 class WorkflowConfig(BaseSettings):
     """
     Configuration for workflow execution
@@ -529,6 +545,28 @@ class WorkflowConfig(BaseSettings):
         default=200 * 1024,
     )
 
+    # GraphEngine Worker Pool Configuration
+    GRAPH_ENGINE_MIN_WORKERS: PositiveInt = Field(
+        description="Minimum number of workers per GraphEngine instance",
+        default=1,
+    )
+
+    GRAPH_ENGINE_MAX_WORKERS: PositiveInt = Field(
+        description="Maximum number of workers per GraphEngine instance",
+        default=10,
+    )
+
+    GRAPH_ENGINE_SCALE_UP_THRESHOLD: PositiveInt = Field(
+        description="Queue depth threshold that triggers worker scale up",
+        default=3,
+    )
+
+    GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME: float = Field(
+        description="Seconds of idle time before scaling down workers",
+        default=5.0,
+        ge=0.1,
+    )
+
 
 class WorkflowNodeExecutionConfig(BaseSettings):
     """
@@ -796,11 +834,6 @@ class DataSetConfig(BaseSettings):
         default=30,
     )
 
-    DSL_EXPORT_ENCRYPT_DATASET_ID: bool = Field(
-        description="Enable or disable dataset ID encryption when exporting DSL files",
-        default=True,
-    )
-
 
 class WorkspaceConfig(BaseSettings):
     """
@@ -1030,5 +1063,6 @@ class FeatureConfig(
     CeleryBeatConfig,
     CeleryScheduleTasksConfig,
     WorkflowLogConfig,
+    WorkflowVariableTruncationConfig,
 ):
     pass

api/configs/feature/hosted_service/__init__.py

@@ -222,11 +222,28 @@ class HostedFetchAppTemplateConfig(BaseSettings):
     )
 
 
+class HostedFetchPipelineTemplateConfig(BaseSettings):
+    """
+    Configuration for fetching pipeline templates
+    """
+
+    HOSTED_FETCH_PIPELINE_TEMPLATES_MODE: str = Field(
+        description="Mode for fetching pipeline templates: remote, db, or builtin default to remote,",
+        default="remote",
+    )
+
+    HOSTED_FETCH_PIPELINE_TEMPLATES_REMOTE_DOMAIN: str = Field(
+        description="Domain for fetching remote pipeline templates",
+        default="https://tmpl.dify.ai",
+    )
+
+
 class HostedServiceConfig(
     # place the configs in alphabet order
     HostedAnthropicConfig,
     HostedAzureOpenAiConfig,
     HostedFetchAppTemplateConfig,
+    HostedFetchPipelineTemplateConfig,
     HostedMinmaxConfig,
     HostedOpenAiConfig,
     HostedSparkConfig,

api/constants/__init__.py

@@ -16,14 +16,14 @@ AUDIO_EXTENSIONS = ["mp3", "m4a", "wav", "amr", "mpga"]
 AUDIO_EXTENSIONS.extend([ext.upper() for ext in AUDIO_EXTENSIONS])
 
 
-_doc_extensions: list[str]
 if dify_config.ETL_TYPE == "Unstructured":
-    _doc_extensions = ["txt", "markdown", "md", "mdx", "pdf", "html", "htm", "xlsx", "xls", "vtt", "properties"]
-    _doc_extensions.extend(("doc", "docx", "csv", "eml", "msg", "pptx", "xml", "epub"))
+    DOCUMENT_EXTENSIONS = ["txt", "markdown", "md", "mdx", "pdf", "html", "htm", "xlsx", "xls", "vtt", "properties"]
+    DOCUMENT_EXTENSIONS.extend(("doc", "docx", "csv", "eml", "msg", "pptx", "xml", "epub"))
     if dify_config.UNSTRUCTURED_API_URL:
-        _doc_extensions.append("ppt")
+        DOCUMENT_EXTENSIONS.append("ppt")
+    DOCUMENT_EXTENSIONS.extend([ext.upper() for ext in DOCUMENT_EXTENSIONS])
 else:
-    _doc_extensions = [
+    DOCUMENT_EXTENSIONS = [
         "txt",
         "markdown",
         "md",
@@ -38,4 +38,4 @@ else:
         "vtt",
         "properties",
     ]
-DOCUMENT_EXTENSIONS = _doc_extensions + [ext.upper() for ext in _doc_extensions]
+    DOCUMENT_EXTENSIONS.extend([ext.upper() for ext in DOCUMENT_EXTENSIONS])

api/contexts/__init__.py

@@ -3,11 +3,13 @@ from threading import Lock
 from typing import TYPE_CHECKING
 
 from contexts.wrapper import RecyclableContextVar
+from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
 
 if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
+    from core.workflow.entities.variable_pool import VariablePool
 
 
 """
@@ -32,3 +34,11 @@ plugin_model_schema_lock: RecyclableContextVar[Lock] = RecyclableContextVar(Cont
 plugin_model_schemas: RecyclableContextVar[dict[str, "AIModelEntity"]] = RecyclableContextVar(
     ContextVar("plugin_model_schemas")
 )
+
+datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginProviderController"]] = (
+    RecyclableContextVar(ContextVar("datasource_plugin_providers"))
+)
+
+datasource_plugin_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
+    ContextVar("datasource_plugin_providers_lock")
+)

api/controllers/console/__init__.py

@@ -1,5 +1,4 @@
 from flask import Blueprint
-from flask_restx import Namespace
 
 from libs.external_api import ExternalApi
 
@@ -27,16 +26,7 @@ from .files import FileApi, FilePreviewApi, FileSupportTypeApi
 from .remote_files import RemoteFileInfoApi, RemoteFileUploadApi
 
 bp = Blueprint("console", __name__, url_prefix="/console/api")
-
-api = ExternalApi(
-    bp,
-    version="1.0",
-    title="Console API",
-    description="Console management APIs for app configuration, monitoring, and administration",
-)
-
-# Create namespace
-console_ns = Namespace("console", description="Console management API operations", path="/")
+api = ExternalApi(bp)
 
 # File
 api.add_resource(FileApi, "/files/upload")
@@ -53,90 +43,65 @@ api.add_resource(AppImportConfirmApi, "/apps/imports/<string:import_id>/confirm"
 api.add_resource(AppImportCheckDependenciesApi, "/apps/imports/<string:app_id>/check-dependencies")
 
 # Import other controllers
-from . import (
-    admin,  # pyright: ignore[reportUnusedImport]
-    apikey,  # pyright: ignore[reportUnusedImport]
-    extension,  # pyright: ignore[reportUnusedImport]
-    feature,  # pyright: ignore[reportUnusedImport]
-    init_validate,  # pyright: ignore[reportUnusedImport]
-    ping,  # pyright: ignore[reportUnusedImport]
-    setup,  # pyright: ignore[reportUnusedImport]
-    version,  # pyright: ignore[reportUnusedImport]
-)
+from . import admin, apikey, extension, feature, ping, setup, spec, version
 
 # Import app controllers
 from .app import (
-    advanced_prompt_template,  # pyright: ignore[reportUnusedImport]
-    agent,  # pyright: ignore[reportUnusedImport]
-    annotation,  # pyright: ignore[reportUnusedImport]
-    app,  # pyright: ignore[reportUnusedImport]
-    audio,  # pyright: ignore[reportUnusedImport]
-    completion,  # pyright: ignore[reportUnusedImport]
-    conversation,  # pyright: ignore[reportUnusedImport]
-    conversation_variables,  # pyright: ignore[reportUnusedImport]
-    generator,  # pyright: ignore[reportUnusedImport]
-    mcp_server,  # pyright: ignore[reportUnusedImport]
-    message,  # pyright: ignore[reportUnusedImport]
-    model_config,  # pyright: ignore[reportUnusedImport]
-    ops_trace,  # pyright: ignore[reportUnusedImport]
-    site,  # pyright: ignore[reportUnusedImport]
-    statistic,  # pyright: ignore[reportUnusedImport]
-    workflow,  # pyright: ignore[reportUnusedImport]
-    workflow_app_log,  # pyright: ignore[reportUnusedImport]
-    workflow_draft_variable,  # pyright: ignore[reportUnusedImport]
-    workflow_run,  # pyright: ignore[reportUnusedImport]
-    workflow_statistic,  # pyright: ignore[reportUnusedImport]
+    advanced_prompt_template,
+    agent,
+    annotation,
+    app,
+    audio,
+    completion,
+    conversation,
+    conversation_variables,
+    generator,
+    mcp_server,
+    message,
+    model_config,
+    ops_trace,
+    site,
+    statistic,
+    workflow,
+    workflow_app_log,
+    workflow_draft_variable,
+    workflow_run,
+    workflow_statistic,
 )
 
 # Import auth controllers
-from .auth import (
-    activate,  # pyright: ignore[reportUnusedImport]
-    data_source_bearer_auth,  # pyright: ignore[reportUnusedImport]
-    data_source_oauth,  # pyright: ignore[reportUnusedImport]
-    forgot_password,  # pyright: ignore[reportUnusedImport]
-    login,  # pyright: ignore[reportUnusedImport]
-    oauth,  # pyright: ignore[reportUnusedImport]
-    oauth_server,  # pyright: ignore[reportUnusedImport]
-)
+from .auth import activate, data_source_bearer_auth, data_source_oauth, forgot_password, login, oauth, oauth_server
 
 # Import billing controllers
-from .billing import billing, compliance  # pyright: ignore[reportUnusedImport]
+from .billing import billing, compliance
 
 # Import datasets controllers
 from .datasets import (
-    data_source,  # pyright: ignore[reportUnusedImport]
-    datasets,  # pyright: ignore[reportUnusedImport]
-    datasets_document,  # pyright: ignore[reportUnusedImport]
-    datasets_segments,  # pyright: ignore[reportUnusedImport]
-    external,  # pyright: ignore[reportUnusedImport]
-    hit_testing,  # pyright: ignore[reportUnusedImport]
-    metadata,  # pyright: ignore[reportUnusedImport]
-    website,  # pyright: ignore[reportUnusedImport]
+    data_source,
+    datasets,
+    datasets_document,
+    datasets_segments,
+    external,
+    hit_testing,
+    metadata,
+    website,
+)
+from .datasets.rag_pipeline import (
+    datasource_auth,
+    datasource_content_preview,
+    rag_pipeline,
+    rag_pipeline_datasets,
+    rag_pipeline_draft_variable,
+    rag_pipeline_import,
+    rag_pipeline_workflow,
 )
 
 # Import explore controllers
 from .explore import (
-    installed_app,  # pyright: ignore[reportUnusedImport]
-    parameter,  # pyright: ignore[reportUnusedImport]
-    recommended_app,  # pyright: ignore[reportUnusedImport]
-    saved_message,  # pyright: ignore[reportUnusedImport]
-)
-
-# Import tag controllers
-from .tag import tags  # pyright: ignore[reportUnusedImport]
-
-# Import workspace controllers
-from .workspace import (
-    account,  # pyright: ignore[reportUnusedImport]
-    agent_providers,  # pyright: ignore[reportUnusedImport]
-    endpoint,  # pyright: ignore[reportUnusedImport]
-    load_balancing_config,  # pyright: ignore[reportUnusedImport]
-    members,  # pyright: ignore[reportUnusedImport]
-    model_providers,  # pyright: ignore[reportUnusedImport]
-    models,  # pyright: ignore[reportUnusedImport]
-    plugin,  # pyright: ignore[reportUnusedImport]
-    tool_providers,  # pyright: ignore[reportUnusedImport]
-    workspace,  # pyright: ignore[reportUnusedImport]
+    installed_app,
+    parameter,
+    recommended_app,
+    saved_message,
 )
 
 # Explore Audio
@@ -210,4 +175,19 @@ api.add_resource(
     InstalledAppWorkflowTaskStopApi, "/installed-apps/<uuid:installed_app_id>/workflows/tasks/<string:task_id>/stop"
 )
 
-api.add_namespace(console_ns)
+# Import tag controllers
+from .tag import tags
+
+# Import workspace controllers
+from .workspace import (
+    account,
+    agent_providers,
+    endpoint,
+    load_balancing_config,
+    members,
+    model_providers,
+    models,
+    plugin,
+    tool_providers,
+    workspace,
+)

api/controllers/console/admin.py

@@ -3,7 +3,7 @@ from functools import wraps
 from typing import ParamSpec, TypeVar
 
 from flask import request
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, reqparse
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import NotFound, Unauthorized
@@ -12,7 +12,7 @@ P = ParamSpec("P")
 R = TypeVar("R")
 from configs import dify_config
 from constants.languages import supported_language
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.wraps import only_edition_cloud
 from extensions.ext_database import db
 from models.model import App, InstalledApp, RecommendedApp
@@ -45,28 +45,7 @@ def admin_required(view: Callable[P, R]):
     return decorated
 
 
-@console_ns.route("/admin/insert-explore-apps")
 class InsertExploreAppListApi(Resource):
-    @api.doc("insert_explore_app")
-    @api.doc(description="Insert or update an app in the explore list")
-    @api.expect(
-        api.model(
-            "InsertExploreAppRequest",
-            {
-                "app_id": fields.String(required=True, description="Application ID"),
-                "desc": fields.String(description="App description"),
-                "copyright": fields.String(description="Copyright information"),
-                "privacy_policy": fields.String(description="Privacy policy"),
-                "custom_disclaimer": fields.String(description="Custom disclaimer"),
-                "language": fields.String(required=True, description="Language code"),
-                "category": fields.String(required=True, description="App category"),
-                "position": fields.Integer(required=True, description="Display position"),
-            },
-        )
-    )
-    @api.response(200, "App updated successfully")
-    @api.response(201, "App inserted successfully")
-    @api.response(404, "App not found")
     @only_edition_cloud
     @admin_required
     def post(self):
@@ -136,12 +115,7 @@ class InsertExploreAppListApi(Resource):
                 return {"result": "success"}, 200
 
 
-@console_ns.route("/admin/insert-explore-apps/<uuid:app_id>")
 class InsertExploreAppApi(Resource):
-    @api.doc("delete_explore_app")
-    @api.doc(description="Remove an app from the explore list")
-    @api.doc(params={"app_id": "Application ID to remove"})
-    @api.response(204, "App removed successfully")
     @only_edition_cloud
     @admin_required
     def delete(self, app_id):
@@ -178,3 +152,7 @@ class InsertExploreAppApi(Resource):
         db.session.commit()
 
         return {"result": "success"}, 204
+
+
+api.add_resource(InsertExploreAppListApi, "/admin/insert-explore-apps")
+api.add_resource(InsertExploreAppApi, "/admin/insert-explore-apps/<uuid:app_id>")

api/controllers/console/apikey.py

@@ -1,9 +1,8 @@
-from typing import Optional
+from typing import Any, Optional
 
 import flask_restx
 from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with
-from flask_restx._http import HTTPStatus
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden
@@ -14,7 +13,7 @@ from libs.login import login_required
 from models.dataset import Dataset
 from models.model import ApiToken, App
 
-from . import api, console_ns
+from . import api
 from .wraps import account_initialization_required, setup_required
 
 api_key_fields = {
@@ -41,7 +40,7 @@ def _get_resource(resource_id, tenant_id, resource_model):
             ).scalar_one_or_none()
 
     if resource is None:
-        flask_restx.abort(HTTPStatus.NOT_FOUND, message=f"{resource_model.__name__} not found.")
+        flask_restx.abort(404, message=f"{resource_model.__name__} not found.")
 
     return resource
 
@@ -50,7 +49,7 @@ class BaseApiKeyListResource(Resource):
     method_decorators = [account_initialization_required, login_required, setup_required]
 
     resource_type: str | None = None
-    resource_model: Optional[type] = None
+    resource_model: Optional[Any] = None
     resource_id_field: str | None = None
     token_prefix: str | None = None
     max_keys = 10
@@ -60,11 +59,11 @@ class BaseApiKeyListResource(Resource):
         assert self.resource_id_field is not None, "resource_id_field must be set"
         resource_id = str(resource_id)
         _get_resource(resource_id, current_user.current_tenant_id, self.resource_model)
-        keys = db.session.scalars(
-            select(ApiToken).where(
-                ApiToken.type == self.resource_type, getattr(ApiToken, self.resource_id_field) == resource_id
-            )
-        ).all()
+        keys = (
+            db.session.query(ApiToken)
+            .where(ApiToken.type == self.resource_type, getattr(ApiToken, self.resource_id_field) == resource_id)
+            .all()
+        )
         return {"items": keys}
 
     @marshal_with(api_key_fields)
@@ -83,7 +82,7 @@ class BaseApiKeyListResource(Resource):
 
         if current_key_count >= self.max_keys:
             flask_restx.abort(
-                HTTPStatus.BAD_REQUEST,
+                400,
                 message=f"Cannot create more than {self.max_keys} API keys for this resource type.",
                 custom="max_keys_exceeded",
             )
@@ -103,7 +102,7 @@ class BaseApiKeyResource(Resource):
     method_decorators = [account_initialization_required, login_required, setup_required]
 
     resource_type: str | None = None
-    resource_model: Optional[type] = None
+    resource_model: Optional[Any] = None
     resource_id_field: str | None = None
 
     def delete(self, resource_id, api_key_id):
@@ -127,7 +126,7 @@ class BaseApiKeyResource(Resource):
         )
 
         if key is None:
-            flask_restx.abort(HTTPStatus.NOT_FOUND, message="API key not found")
+            flask_restx.abort(404, message="API key not found")
 
         db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
@@ -135,25 +134,7 @@ class BaseApiKeyResource(Resource):
         return {"result": "success"}, 204
 
 
-@console_ns.route("/apps/<uuid:resource_id>/api-keys")
 class AppApiKeyListResource(BaseApiKeyListResource):
-    @api.doc("get_app_api_keys")
-    @api.doc(description="Get all API keys for an app")
-    @api.doc(params={"resource_id": "App ID"})
-    @api.response(200, "Success", api_key_list)
-    def get(self, resource_id):
-        """Get all API keys for an app"""
-        return super().get(resource_id)
-
-    @api.doc("create_app_api_key")
-    @api.doc(description="Create a new API key for an app")
-    @api.doc(params={"resource_id": "App ID"})
-    @api.response(201, "API key created successfully", api_key_fields)
-    @api.response(400, "Maximum keys exceeded")
-    def post(self, resource_id):
-        """Create a new API key for an app"""
-        return super().post(resource_id)
-
     def after_request(self, resp):
         resp.headers["Access-Control-Allow-Origin"] = "*"
         resp.headers["Access-Control-Allow-Credentials"] = "true"
@@ -165,16 +146,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
     token_prefix = "app-"
 
 
-@console_ns.route("/apps/<uuid:resource_id>/api-keys/<uuid:api_key_id>")
 class AppApiKeyResource(BaseApiKeyResource):
-    @api.doc("delete_app_api_key")
-    @api.doc(description="Delete an API key for an app")
-    @api.doc(params={"resource_id": "App ID", "api_key_id": "API key ID"})
-    @api.response(204, "API key deleted successfully")
-    def delete(self, resource_id, api_key_id):
-        """Delete an API key for an app"""
-        return super().delete(resource_id, api_key_id)
-
     def after_request(self, resp):
         resp.headers["Access-Control-Allow-Origin"] = "*"
         resp.headers["Access-Control-Allow-Credentials"] = "true"
@@ -185,25 +157,7 @@ class AppApiKeyResource(BaseApiKeyResource):
     resource_id_field = "app_id"
 
 
-@console_ns.route("/datasets/<uuid:resource_id>/api-keys")
 class DatasetApiKeyListResource(BaseApiKeyListResource):
-    @api.doc("get_dataset_api_keys")
-    @api.doc(description="Get all API keys for a dataset")
-    @api.doc(params={"resource_id": "Dataset ID"})
-    @api.response(200, "Success", api_key_list)
-    def get(self, resource_id):
-        """Get all API keys for a dataset"""
-        return super().get(resource_id)
-
-    @api.doc("create_dataset_api_key")
-    @api.doc(description="Create a new API key for a dataset")
-    @api.doc(params={"resource_id": "Dataset ID"})
-    @api.response(201, "API key created successfully", api_key_fields)
-    @api.response(400, "Maximum keys exceeded")
-    def post(self, resource_id):
-        """Create a new API key for a dataset"""
-        return super().post(resource_id)
-
     def after_request(self, resp):
         resp.headers["Access-Control-Allow-Origin"] = "*"
         resp.headers["Access-Control-Allow-Credentials"] = "true"
@@ -215,16 +169,7 @@ class DatasetApiKeyListResource(BaseApiKeyListResource):
     token_prefix = "ds-"
 
 
-@console_ns.route("/datasets/<uuid:resource_id>/api-keys/<uuid:api_key_id>")
 class DatasetApiKeyResource(BaseApiKeyResource):
-    @api.doc("delete_dataset_api_key")
-    @api.doc(description="Delete an API key for a dataset")
-    @api.doc(params={"resource_id": "Dataset ID", "api_key_id": "API key ID"})
-    @api.response(204, "API key deleted successfully")
-    def delete(self, resource_id, api_key_id):
-        """Delete an API key for a dataset"""
-        return super().delete(resource_id, api_key_id)
-
     def after_request(self, resp):
         resp.headers["Access-Control-Allow-Origin"] = "*"
         resp.headers["Access-Control-Allow-Credentials"] = "true"
@@ -233,3 +178,9 @@ class DatasetApiKeyResource(BaseApiKeyResource):
     resource_type = "dataset"
     resource_model = Dataset
     resource_id_field = "dataset_id"
+
+
+api.add_resource(AppApiKeyListResource, "/apps/<uuid:resource_id>/api-keys")
+api.add_resource(AppApiKeyResource, "/apps/<uuid:resource_id>/api-keys/<uuid:api_key_id>")
+api.add_resource(DatasetApiKeyListResource, "/datasets/<uuid:resource_id>/api-keys")
+api.add_resource(DatasetApiKeyResource, "/datasets/<uuid:resource_id>/api-keys/<uuid:api_key_id>")

api/controllers/console/app/app.py

@@ -115,10 +115,6 @@ class AppListApi(Resource):
             raise BadRequest("mode is required")
 
         app_service = AppService()
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
-        if current_user.current_tenant_id is None:
-            raise ValueError("current_user.current_tenant_id cannot be None")
         app = app_service.create_app(current_user.current_tenant_id, args, current_user)
 
         return app, 201
@@ -165,26 +161,14 @@ class AppApi(Resource):
         args = parser.parse_args()
 
         app_service = AppService()
-        # Construct ArgsDict from parsed arguments
-        from services.app_service import AppService as AppServiceType
-
-        args_dict: AppServiceType.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)
+        app_model = app_service.update_app(app_model, args)
 
         return app_model
 
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def delete(self, app_model):
         """Delete app"""
         # The role of the current user in the ta table must be admin, owner, or editor
@@ -240,10 +224,10 @@ class AppCopyApi(Resource):
 
 
 class AppExportApi(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         """Export app"""
         # The role of the current user in the ta table must be admin, owner, or editor
@@ -279,7 +263,7 @@ class AppNameApi(Resource):
         args = parser.parse_args()
 
         app_service = AppService()
-        app_model = app_service.update_app_name(app_model, args["name"])
+        app_model = app_service.update_app_name(app_model, args.get("name"))
 
         return app_model
 
@@ -301,7 +285,7 @@ class AppIconApi(Resource):
         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 "")
+        app_model = app_service.update_app_icon(app_model, args.get("icon"), args.get("icon_background"))
 
         return app_model
 
@@ -322,7 +306,7 @@ class AppSiteStatus(Resource):
         args = parser.parse_args()
 
         app_service = AppService()
-        app_model = app_service.update_app_site_status(app_model, args["enable_site"])
+        app_model = app_service.update_app_site_status(app_model, args.get("enable_site"))
 
         return app_model
 
@@ -343,7 +327,7 @@ class AppApiStatus(Resource):
         args = parser.parse_args()
 
         app_service = AppService()
-        app_model = app_service.update_app_api_status(app_model, args["enable_api"])
+        app_model = app_service.update_app_api_status(app_model, args.get("enable_api"))
 
         return app_model
 

api/controllers/console/app/audio.py

@@ -77,10 +77,10 @@ class ChatMessageAudioApi(Resource):
 
 
 class ChatMessageTextApi(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def post(self, app_model: App):
         try:
             parser = reqparse.RequestParser()
@@ -125,10 +125,10 @@ class ChatMessageTextApi(Resource):
 
 
 class TextModesApi(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         try:
             parser = reqparse.RequestParser()

api/controllers/console/app/completion.py

@@ -1,8 +1,9 @@
 import logging
 
+import flask_login
 from flask import request
 from flask_restx import Resource, reqparse
-from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
+from werkzeug.exceptions import InternalServerError, NotFound
 
 import services
 from controllers.console import api
@@ -28,8 +29,7 @@ from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
-from libs.login import current_user, login_required
-from models import Account
+from libs.login import login_required
 from models.model import AppMode
 from services.app_generate_service import AppGenerateService
 from services.errors.llm import InvokeRateLimitError
@@ -56,11 +56,11 @@ class CompletionMessageApi(Resource):
         streaming = args["response_mode"] != "blocking"
         args["auto_generate_name"] = False
 
+        account = flask_login.current_user
+
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account or EndUser instance")
             response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=streaming
+                app_model=app_model, user=account, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=streaming
             )
 
             return helper.compact_generate_response(response)
@@ -92,9 +92,9 @@ class CompletionMessageStopApi(Resource):
     @account_initialization_required
     @get_app_model(mode=AppMode.COMPLETION)
     def post(self, app_model, task_id):
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
+        account = flask_login.current_user
+
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, account.id)
 
         return {"result": "success"}, 200
 
@@ -105,12 +105,6 @@ class ChatMessageApi(Resource):
     @account_initialization_required
     @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT])
     def post(self, app_model):
-        if not isinstance(current_user, Account):
-            raise Forbidden()
-
-        if not current_user.has_edit_permission:
-            raise Forbidden()
-
         parser = reqparse.RequestParser()
         parser.add_argument("inputs", type=dict, required=True, location="json")
         parser.add_argument("query", type=str, required=True, location="json")
@@ -129,11 +123,11 @@ class ChatMessageApi(Resource):
         if external_trace_id:
             args["external_trace_id"] = external_trace_id
 
+        account = flask_login.current_user
+
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account or EndUser instance")
             response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=streaming
+                app_model=app_model, user=account, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=streaming
             )
 
             return helper.compact_generate_response(response)
@@ -167,9 +161,9 @@ class ChatMessageStopApi(Resource):
     @account_initialization_required
     @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
     def post(self, app_model, task_id):
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
+        account = flask_login.current_user
+
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, account.id)
 
         return {"result": "success"}, 200
 

api/controllers/console/app/conversation.py

@@ -22,7 +22,7 @@ from fields.conversation_fields import (
 from libs.datetime_utils import naive_utc_now
 from libs.helper import DatetimeString
 from libs.login import login_required
-from models import Account, Conversation, EndUser, Message, MessageAnnotation
+from models import Conversation, EndUser, Message, MessageAnnotation
 from models.model import AppMode
 from services.conversation_service import ConversationService
 from services.errors.conversation import ConversationNotExistsError
@@ -124,8 +124,6 @@ class CompletionConversationDetailApi(Resource):
         conversation_id = str(conversation_id)
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             ConversationService.delete(app_model, conversation_id, current_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
@@ -284,8 +282,6 @@ class ChatConversationDetailApi(Resource):
         conversation_id = str(conversation_id)
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             ConversationService.delete(app_model, conversation_id, current_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")

api/controllers/console/app/generator.py

@@ -16,7 +16,10 @@ from core.helper.code_executor.javascript.javascript_code_provider import Javasc
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.llm_generator.llm_generator import LLMGenerator
 from core.model_runtime.errors.invoke import InvokeError
+from extensions.ext_database import db
 from libs.login import login_required
+from models import App
+from services.workflow_service import WorkflowService
 
 
 class RuleGenerateApi(Resource):
@@ -135,9 +138,6 @@ class InstructionGenerateApi(Resource):
         try:
             # Generate from nothing for a workflow node
             if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":
-                from models import App, db
-                from services.workflow_service import WorkflowService
-
                 app = db.session.query(App).where(App.id == args["flow_id"]).first()
                 if not app:
                     return {"error": f"app {args['flow_id']} not found"}, 400

api/controllers/console/app/message.py

@@ -1,5 +1,6 @@
 import logging
 
+from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from sqlalchemy import exists, select
@@ -26,8 +27,7 @@ from extensions.ext_database import db
 from fields.conversation_fields import annotation_fields, message_detail_fields
 from libs.helper import uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
 from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
@@ -118,14 +118,11 @@ class ChatMessageListApi(Resource):
 
 
 class MessageFeedbackApi(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def post(self, app_model):
-        if current_user is None:
-            raise Forbidden()
-
         parser = reqparse.RequestParser()
         parser.add_argument("message_id", required=True, type=uuid_value, location="json")
         parser.add_argument("rating", type=str, choices=["like", "dislike", None], location="json")
@@ -170,9 +167,7 @@ class MessageAnnotationApi(Resource):
     @get_app_model
     @marshal_with(annotation_fields)
     def post(self, app_model):
-        if not isinstance(current_user, Account):
-            raise Forbidden()
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -187,10 +182,10 @@ class MessageAnnotationApi(Resource):
 
 
 class MessageAnnotationCountApi(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         count = db.session.query(MessageAnnotation).where(MessageAnnotation.app_id == app_model.id).count()
 

api/controllers/console/app/model_config.py

@@ -2,8 +2,8 @@ import json
 from typing import cast
 
 from flask import request
+from flask_login import current_user
 from flask_restx import Resource
-from werkzeug.exceptions import Forbidden
 
 from controllers.console import api
 from controllers.console.app.wraps import get_app_model
@@ -13,8 +13,7 @@ from core.tools.tool_manager import ToolManager
 from core.tools.utils.configuration import ToolParameterConfigurationManager
 from events.app_event import app_model_config_was_updated
 from extensions.ext_database import db
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from models.model import AppMode, AppModelConfig
 from services.app_model_config_service import AppModelConfigService
 
@@ -26,13 +25,6 @@ class ModelConfigResource(Resource):
     @get_app_model(mode=[AppMode.AGENT_CHAT, AppMode.CHAT, AppMode.COMPLETION])
     def post(self, app_model):
         """Modify app model config"""
-        if not isinstance(current_user, Account):
-            raise Forbidden()
-
-        if not current_user.has_edit_permission:
-            raise Forbidden()
-
-        assert current_user.current_tenant_id is not None, "The tenant information should be loaded."
         # validate config
         model_configuration = AppModelConfigService.validate_configuration(
             tenant_id=current_user.current_tenant_id,

api/controllers/console/app/site.py

@@ -10,7 +10,7 @@ from extensions.ext_database import db
 from fields.app_fields import app_site_fields
 from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
-from models import Account, Site
+from models import Site
 
 
 def parse_app_site_args():
@@ -75,8 +75,6 @@ class AppSite(Resource):
             if value is not None:
                 setattr(site, attr_name, value)
 
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         site.updated_by = current_user.id
         site.updated_at = naive_utc_now()
         db.session.commit()
@@ -101,8 +99,6 @@ class AppSiteAccessTokenReset(Resource):
             raise NotFound
 
         site.code = Site.generate_code(16)
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         site.updated_by = current_user.id
         site.updated_at = naive_utc_now()
         db.session.commit()

api/controllers/console/app/statistic.py

@@ -18,10 +18,10 @@ from models import AppMode, Message
 
 
 class DailyMessageStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 
@@ -75,10 +75,10 @@ WHERE
 
 
 class DailyConversationStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 
@@ -127,10 +127,10 @@ class DailyConversationStatistic(Resource):
 
 
 class DailyTerminalsStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 
@@ -184,10 +184,10 @@ WHERE
 
 
 class DailyTokenCostStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 
@@ -320,10 +320,10 @@ ORDER BY
 
 
 class UserSatisfactionRateStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 
@@ -443,10 +443,10 @@ WHERE
 
 
 class TokensPerSecondStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 

api/controllers/console/app/workflow.py

@@ -11,7 +11,11 @@ from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 import services
 from configs import dify_config
 from controllers.console import api
-from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
+from controllers.console.app.error import (
+    ConversationCompletedError,
+    DraftWorkflowNotExist,
+    DraftWorkflowNotSync,
+)
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
@@ -20,6 +24,7 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
+from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -69,7 +74,7 @@ class DraftWorkflowApi(Resource):
         """
         # The role of the current user in the ta table must be admin, owner, or editor
         assert isinstance(current_user, Account)
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         # fetch draft workflow by app_model
@@ -92,7 +97,7 @@ class DraftWorkflowApi(Resource):
         """
         # The role of the current user in the ta table must be admin, owner, or editor
         assert isinstance(current_user, Account)
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         content_type = request.headers.get("Content-Type", "")
@@ -170,7 +175,7 @@ class AdvancedChatDraftWorkflowRunApi(Resource):
         """
         # The role of the current user in the ta table must be admin, owner, or editor
         assert isinstance(current_user, Account)
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         if not isinstance(current_user, Account):
@@ -220,7 +225,7 @@ class AdvancedChatDraftRunIterationNodeApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -256,7 +261,7 @@ class WorkflowDraftRunIterationNodeApi(Resource):
         # The role of the current user in the ta table must be admin, owner, or editor
         if not isinstance(current_user, Account):
             raise Forbidden()
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -293,7 +298,7 @@ class AdvancedChatDraftRunLoopNodeApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -330,7 +335,7 @@ class WorkflowDraftRunLoopNodeApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -367,7 +372,7 @@ class DraftWorkflowRunApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -406,10 +411,15 @@ class WorkflowTaskStopApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}
 
@@ -428,7 +438,7 @@ class DraftWorkflowNodeRunApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -476,7 +486,7 @@ class PublishedWorkflowApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         # fetch published workflow by app_model
@@ -497,7 +507,7 @@ class PublishedWorkflowApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -547,7 +557,7 @@ class DefaultBlockConfigsApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         # Get default block configs
@@ -567,7 +577,7 @@ class DefaultBlockConfigApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -602,7 +612,7 @@ class ConvertToWorkflowApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         if request.data:
@@ -651,7 +661,7 @@ class PublishedAllWorkflowApi(Resource):
 
         if not isinstance(current_user, Account):
             raise Forbidden()
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -702,7 +712,7 @@ class WorkflowByIdApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # Check permission
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -758,7 +768,7 @@ class WorkflowByIdApi(Resource):
         if not isinstance(current_user, Account):
             raise Forbidden()
         # Check permission
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         workflow_service = WorkflowService()

api/controllers/console/app/workflow_app_log.py

@@ -6,7 +6,7 @@ from sqlalchemy.orm import Session
 from controllers.console import api
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
+from core.workflow.enums import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import workflow_app_log_pagination_fields
 from libs.login import login_required

api/controllers/console/app/workflow_draft_variable.py

@@ -13,14 +13,16 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.file import helpers as file_helpers
 from core.variables.segment_group import SegmentGroup
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type
 from libs.login import current_user, login_required
-from models import App, AppMode, db
+from models import App, AppMode
 from models.account import Account
 from models.workflow import WorkflowDraftVariable
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
@@ -74,6 +76,22 @@ def _serialize_variable_type(workflow_draft_var: WorkflowDraftVariable) -> str:
     return value_type.exposed_type().value
 
 
+def _serialize_full_content(variable: WorkflowDraftVariable) -> dict | None:
+    """Serialize full_content information for large variables."""
+    if not variable.is_truncated():
+        return None
+
+    variable_file = variable.variable_file
+    assert variable_file is not None
+
+    return {
+        "size_bytes": variable_file.size,
+        "value_type": variable_file.value_type.exposed_type().value,
+        "length": variable_file.length,
+        "download_url": file_helpers.get_signed_file_url(variable_file.upload_file_id, as_attachment=True),
+    }
+
+
 _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
     "id": fields.String,
     "type": fields.String(attribute=lambda model: model.get_variable_type()),
@@ -83,11 +101,13 @@ _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
     "value_type": fields.String(attribute=_serialize_variable_type),
     "edited": fields.Boolean(attribute=lambda model: model.edited),
     "visible": fields.Boolean,
+    "is_truncated": fields.Boolean(attribute=lambda model: model.file_id is not None),
 }
 
 _WORKFLOW_DRAFT_VARIABLE_FIELDS = dict(
     _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,
     value=fields.Raw(attribute=_serialize_var_value),
+    full_content=fields.Raw(attribute=_serialize_full_content),
 )
 
 _WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS = {
@@ -137,7 +157,7 @@ def _api_prerequisite(f):
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
     def wrapper(*args, **kwargs):
         assert isinstance(current_user, Account)
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
         return f(*args, **kwargs)
 

api/controllers/console/app/workflow_statistic.py

@@ -18,10 +18,10 @@ from models.model import AppMode
 
 
 class WorkflowDailyRunsStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 
@@ -80,10 +80,10 @@ WHERE
 
 
 class WorkflowDailyTerminalsStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 
@@ -142,10 +142,10 @@ WHERE
 
 
 class WorkflowDailyTokenCostStatistic(Resource):
-    @get_app_model
     @setup_required
     @login_required
     @account_initialization_required
+    @get_app_model
     def get(self, app_model):
         account = current_user
 

api/controllers/console/app/wraps.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import Optional, ParamSpec, TypeVar, Union
+from typing import Optional, Union
 
 from controllers.console.app.error import AppNotFoundError
 from extensions.ext_database import db
@@ -8,9 +8,6 @@ from libs.login import current_user
 from models import App, AppMode
 from models.account import Account
 
-P = ParamSpec("P")
-R = TypeVar("R")
-
 
 def _load_app_model(app_id: str) -> Optional[App]:
     assert isinstance(current_user, Account)
@@ -22,10 +19,10 @@ def _load_app_model(app_id: str) -> Optional[App]:
     return app_model
 
 
-def get_app_model(view: Optional[Callable[P, R]] = None, *, mode: Union[AppMode, list[AppMode], None] = None):
-    def decorator(view_func: Callable[P, R]):
+def get_app_model(view: Optional[Callable] = None, *, mode: Union[AppMode, list[AppMode], None] = None):
+    def decorator(view_func):
         @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        def decorated_view(*args, **kwargs):
             if not kwargs.get("app_id"):
                 raise ValueError("missing app_id in path parameters")
 

api/controllers/console/auth/activate.py

@@ -1,8 +1,8 @@
 from flask import request
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, reqparse
 
 from constants.languages import supported_language
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.error import AlreadyActivateError
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
@@ -10,36 +10,14 @@ from libs.helper import StrLen, email, extract_remote_ip, timezone
 from models.account import AccountStatus
 from services.account_service import AccountService, RegisterService
 
-active_check_parser = reqparse.RequestParser()
-active_check_parser.add_argument(
-    "workspace_id", type=str, required=False, nullable=True, location="args", help="Workspace ID"
-)
-active_check_parser.add_argument(
-    "email", type=email, required=False, nullable=True, location="args", help="Email address"
-)
-active_check_parser.add_argument(
-    "token", type=str, required=True, nullable=False, location="args", help="Activation token"
-)
 
-
-@console_ns.route("/activate/check")
 class ActivateCheckApi(Resource):
-    @api.doc("check_activation_token")
-    @api.doc(description="Check if activation token is valid")
-    @api.expect(active_check_parser)
-    @api.response(
-        200,
-        "Success",
-        api.model(
-            "ActivationCheckResponse",
-            {
-                "is_valid": fields.Boolean(description="Whether token is valid"),
-                "data": fields.Raw(description="Activation data if valid"),
-            },
-        ),
-    )
     def get(self):
-        args = active_check_parser.parse_args()
+        parser = reqparse.RequestParser()
+        parser.add_argument("workspace_id", type=str, required=False, nullable=True, location="args")
+        parser.add_argument("email", type=email, required=False, nullable=True, location="args")
+        parser.add_argument("token", type=str, required=True, nullable=False, location="args")
+        args = parser.parse_args()
 
         workspaceId = args["workspace_id"]
         reg_email = args["email"]
@@ -60,36 +38,18 @@ class ActivateCheckApi(Resource):
             return {"is_valid": False}
 
 
-active_parser = reqparse.RequestParser()
-active_parser.add_argument("workspace_id", type=str, required=False, nullable=True, location="json")
-active_parser.add_argument("email", type=email, required=False, nullable=True, location="json")
-active_parser.add_argument("token", type=str, required=True, nullable=False, location="json")
-active_parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
-active_parser.add_argument(
-    "interface_language", type=supported_language, required=True, nullable=False, location="json"
-)
-active_parser.add_argument("timezone", type=timezone, required=True, nullable=False, location="json")
-
-
-@console_ns.route("/activate")
 class ActivateApi(Resource):
-    @api.doc("activate_account")
-    @api.doc(description="Activate account with invitation token")
-    @api.expect(active_parser)
-    @api.response(
-        200,
-        "Account activated successfully",
-        api.model(
-            "ActivationResponse",
-            {
-                "result": fields.String(description="Operation result"),
-                "data": fields.Raw(description="Login token data"),
-            },
-        ),
-    )
-    @api.response(400, "Already activated or invalid token")
     def post(self):
-        args = active_parser.parse_args()
+        parser = reqparse.RequestParser()
+        parser.add_argument("workspace_id", type=str, required=False, nullable=True, location="json")
+        parser.add_argument("email", type=email, required=False, nullable=True, location="json")
+        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument(
+            "interface_language", type=supported_language, required=True, nullable=False, location="json"
+        )
+        parser.add_argument("timezone", type=timezone, required=True, nullable=False, location="json")
+        args = parser.parse_args()
 
         invitation = RegisterService.get_invitation_if_token_valid(args["workspace_id"], args["email"], args["token"])
         if invitation is None:
@@ -110,3 +70,7 @@ class ActivateApi(Resource):
         token_pair = AccountService.login(account, ip_address=extract_remote_ip(request))
 
         return {"result": "success", "data": token_pair.model_dump()}
+
+
+api.add_resource(ActivateCheckApi, "/activate/check")
+api.add_resource(ActivateApi, "/activate")

api/controllers/console/auth/data_source_oauth.py

@@ -3,11 +3,11 @@ import logging
 import requests
 from flask import current_app, redirect, request
 from flask_login import current_user
-from flask_restx import Resource, fields
+from flask_restx import Resource
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.console import api, console_ns
+from controllers.console import api
 from libs.login import login_required
 from libs.oauth_data_source import NotionOAuth
 
@@ -28,21 +28,7 @@ def get_oauth_providers():
         return OAUTH_PROVIDERS
 
 
-@console_ns.route("/oauth/data-source/<string:provider>")
 class OAuthDataSource(Resource):
-    @api.doc("oauth_data_source")
-    @api.doc(description="Get OAuth authorization URL for data source provider")
-    @api.doc(params={"provider": "Data source provider name (notion)"})
-    @api.response(
-        200,
-        "Authorization URL or internal setup success",
-        api.model(
-            "OAuthDataSourceResponse",
-            {"data": fields.Raw(description="Authorization URL or 'internal' for internal setup")},
-        ),
-    )
-    @api.response(400, "Invalid provider")
-    @api.response(403, "Admin privileges required")
     def get(self, provider: str):
         # The role of the current user in the table must be admin or owner
         if not current_user.is_admin_or_owner:
@@ -63,19 +49,7 @@ class OAuthDataSource(Resource):
             return {"data": auth_url}, 200
 
 
-@console_ns.route("/oauth/data-source/callback/<string:provider>")
 class OAuthDataSourceCallback(Resource):
-    @api.doc("oauth_data_source_callback")
-    @api.doc(description="Handle OAuth callback from data source provider")
-    @api.doc(
-        params={
-            "provider": "Data source provider name (notion)",
-            "code": "Authorization code from OAuth provider",
-            "error": "Error message from OAuth provider",
-        }
-    )
-    @api.response(302, "Redirect to console with result")
-    @api.response(400, "Invalid provider")
     def get(self, provider: str):
         OAUTH_DATASOURCE_PROVIDERS = get_oauth_providers()
         with current_app.app_context():
@@ -94,19 +68,7 @@ class OAuthDataSourceCallback(Resource):
             return redirect(f"{dify_config.CONSOLE_WEB_URL}?type=notion&error=Access denied")
 
 
-@console_ns.route("/oauth/data-source/binding/<string:provider>")
 class OAuthDataSourceBinding(Resource):
-    @api.doc("oauth_data_source_binding")
-    @api.doc(description="Bind OAuth data source with authorization code")
-    @api.doc(
-        params={"provider": "Data source provider name (notion)", "code": "Authorization code from OAuth provider"}
-    )
-    @api.response(
-        200,
-        "Data source binding success",
-        api.model("OAuthDataSourceBindingResponse", {"result": fields.String(description="Operation result")}),
-    )
-    @api.response(400, "Invalid provider or code")
     def get(self, provider: str):
         OAUTH_DATASOURCE_PROVIDERS = get_oauth_providers()
         with current_app.app_context():
@@ -128,17 +90,7 @@ class OAuthDataSourceBinding(Resource):
             return {"result": "success"}, 200
 
 
-@console_ns.route("/oauth/data-source/<string:provider>/<uuid:binding_id>/sync")
 class OAuthDataSourceSync(Resource):
-    @api.doc("oauth_data_source_sync")
-    @api.doc(description="Sync data from OAuth data source")
-    @api.doc(params={"provider": "Data source provider name (notion)", "binding_id": "Data source binding ID"})
-    @api.response(
-        200,
-        "Data source sync success",
-        api.model("OAuthDataSourceSyncResponse", {"result": fields.String(description="Operation result")}),
-    )
-    @api.response(400, "Invalid provider or sync failed")
     @setup_required
     @login_required
     @account_initialization_required
@@ -159,3 +111,9 @@ class OAuthDataSourceSync(Resource):
             return {"error": "OAuth data source process failed"}, 400
 
         return {"result": "success"}, 200
+
+
+api.add_resource(OAuthDataSource, "/oauth/data-source/<string:provider>")
+api.add_resource(OAuthDataSourceCallback, "/oauth/data-source/callback/<string:provider>")
+api.add_resource(OAuthDataSourceBinding, "/oauth/data-source/binding/<string:provider>")
+api.add_resource(OAuthDataSourceSync, "/oauth/data-source/<string:provider>/<uuid:binding_id>/sync")

api/controllers/console/auth/forgot_password.py

@@ -2,12 +2,12 @@ import base64
 import secrets
 
 from flask import request
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, reqparse
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from constants.languages import languages
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.auth.error import (
     EmailCodeError,
     EmailPasswordResetLimitError,
@@ -28,32 +28,7 @@ from services.errors.workspace import WorkSpaceNotAllowedCreateError, Workspaces
 from services.feature_service import FeatureService
 
 
-@console_ns.route("/forgot-password")
 class ForgotPasswordSendEmailApi(Resource):
-    @api.doc("send_forgot_password_email")
-    @api.doc(description="Send password reset email")
-    @api.expect(
-        api.model(
-            "ForgotPasswordEmailRequest",
-            {
-                "email": fields.String(required=True, description="Email address"),
-                "language": fields.String(description="Language for email (zh-Hans/en-US)"),
-            },
-        )
-    )
-    @api.response(
-        200,
-        "Email sent successfully",
-        api.model(
-            "ForgotPasswordEmailResponse",
-            {
-                "result": fields.String(description="Operation result"),
-                "data": fields.String(description="Reset token"),
-                "code": fields.String(description="Error code if account not found"),
-            },
-        ),
-    )
-    @api.response(400, "Invalid email or rate limit exceeded")
     @setup_required
     @email_password_login_enabled
     def post(self):
@@ -86,33 +61,7 @@ class ForgotPasswordSendEmailApi(Resource):
         return {"result": "success", "data": token}
 
 
-@console_ns.route("/forgot-password/validity")
 class ForgotPasswordCheckApi(Resource):
-    @api.doc("check_forgot_password_code")
-    @api.doc(description="Verify password reset code")
-    @api.expect(
-        api.model(
-            "ForgotPasswordCheckRequest",
-            {
-                "email": fields.String(required=True, description="Email address"),
-                "code": fields.String(required=True, description="Verification code"),
-                "token": fields.String(required=True, description="Reset token"),
-            },
-        )
-    )
-    @api.response(
-        200,
-        "Code verified successfully",
-        api.model(
-            "ForgotPasswordCheckResponse",
-            {
-                "is_valid": fields.Boolean(description="Whether code is valid"),
-                "email": fields.String(description="Email address"),
-                "token": fields.String(description="New reset token"),
-            },
-        ),
-    )
-    @api.response(400, "Invalid code or token")
     @setup_required
     @email_password_login_enabled
     def post(self):
@@ -151,26 +100,7 @@ class ForgotPasswordCheckApi(Resource):
         return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
 
 
-@console_ns.route("/forgot-password/resets")
 class ForgotPasswordResetApi(Resource):
-    @api.doc("reset_password")
-    @api.doc(description="Reset password with verification token")
-    @api.expect(
-        api.model(
-            "ForgotPasswordResetRequest",
-            {
-                "token": fields.String(required=True, description="Verification token"),
-                "new_password": fields.String(required=True, description="New password"),
-                "password_confirm": fields.String(required=True, description="Password confirmation"),
-            },
-        )
-    )
-    @api.response(
-        200,
-        "Password reset successfully",
-        api.model("ForgotPasswordResetResponse", {"result": fields.String(description="Operation result")}),
-    )
-    @api.response(400, "Invalid token or password mismatch")
     @setup_required
     @email_password_login_enabled
     def post(self):
@@ -242,3 +172,8 @@ class ForgotPasswordResetApi(Resource):
             pass
         except AccountRegisterError:
             raise AccountInFreezeError()
+
+
+api.add_resource(ForgotPasswordSendEmailApi, "/forgot-password")
+api.add_resource(ForgotPasswordCheckApi, "/forgot-password/validity")
+api.add_resource(ForgotPasswordResetApi, "/forgot-password/resets")

api/controllers/console/auth/oauth.py

@@ -22,7 +22,7 @@ from services.errors.account import AccountNotFoundError, AccountRegisterError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkSpaceNotFoundError
 from services.feature_service import FeatureService
 
-from .. import api, console_ns
+from .. import api
 
 logger = logging.getLogger(__name__)
 
@@ -50,13 +50,7 @@ def get_oauth_providers():
         return OAUTH_PROVIDERS
 
 
-@console_ns.route("/oauth/login/<provider>")
 class OAuthLogin(Resource):
-    @api.doc("oauth_login")
-    @api.doc(description="Initiate OAuth login process")
-    @api.doc(params={"provider": "OAuth provider name (github/google)", "invite_token": "Optional invitation token"})
-    @api.response(302, "Redirect to OAuth authorization URL")
-    @api.response(400, "Invalid provider")
     def get(self, provider: str):
         invite_token = request.args.get("invite_token") or None
         OAUTH_PROVIDERS = get_oauth_providers()
@@ -69,19 +63,7 @@ class OAuthLogin(Resource):
         return redirect(auth_url)
 
 
-@console_ns.route("/oauth/authorize/<provider>")
 class OAuthCallback(Resource):
-    @api.doc("oauth_callback")
-    @api.doc(description="Handle OAuth callback and complete login process")
-    @api.doc(
-        params={
-            "provider": "OAuth provider name (github/google)",
-            "code": "Authorization code from OAuth provider",
-            "state": "Optional state parameter (used for invite token)",
-        }
-    )
-    @api.response(302, "Redirect to console with access token")
-    @api.response(400, "OAuth process failed")
     def get(self, provider: str):
         OAUTH_PROVIDERS = get_oauth_providers()
         with current_app.app_context():
@@ -95,9 +77,6 @@ class OAuthCallback(Resource):
         if state:
             invite_token = state
 
-        if not code:
-            return {"error": "Authorization code is required"}, 400
-
         try:
             token = oauth_provider.get_access_token(code)
             user_info = oauth_provider.get_user_info(token)
@@ -107,7 +86,7 @@ class OAuthCallback(Resource):
             return {"error": "OAuth process failed"}, 400
 
         if invite_token and RegisterService.is_valid_invite_token(invite_token):
-            invitation = RegisterService.get_invitation_by_token(token=invite_token)
+            invitation = RegisterService._get_invitation_by_token(token=invite_token)
             if invitation:
                 invitation_email = invitation.get("email", None)
                 if invitation_email != user_info.email:
@@ -202,3 +181,7 @@ def _generate_account(provider: str, user_info: OAuthUserInfo):
     AccountService.link_account_integrate(provider, user_info.id, account)
 
     return account
+
+
+api.add_resource(OAuthLogin, "/oauth/login/<provider>")
+api.add_resource(OAuthCallback, "/oauth/authorize/<provider>")

api/controllers/console/datasets/data_source.py

@@ -1,4 +1,6 @@
 import json
+from collections.abc import Generator
+from typing import cast
 
 from flask import request
 from flask_login import current_user
@@ -9,6 +11,8 @@ from werkzeug.exceptions import NotFound
 
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required, setup_required
+from core.datasource.entities.datasource_entities import DatasourceProviderType, OnlineDocumentPagesMessage
+from core.datasource.online_document.online_document_plugin import OnlineDocumentDatasourcePlugin
 from core.indexing_runner import IndexingRunner
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
@@ -19,6 +23,7 @@ from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import DataSourceOauthBinding, Document
 from services.dataset_service import DatasetService, DocumentService
+from services.datasource_provider_service import DatasourceProviderService
 from tasks.document_indexing_sync_task import document_indexing_sync_task
 
 
@@ -29,12 +34,14 @@ class DataSourceApi(Resource):
     @marshal_with(integrate_list_fields)
     def get(self):
         # get workspace data source integrates
-        data_source_integrates = db.session.scalars(
-            select(DataSourceOauthBinding).where(
+        data_source_integrates = (
+            db.session.query(DataSourceOauthBinding)
+            .where(
                 DataSourceOauthBinding.tenant_id == current_user.current_tenant_id,
                 DataSourceOauthBinding.disabled == False,
             )
-        ).all()
+            .all()
+        )
 
         base_url = request.url_root.rstrip("/")
         data_source_oauth_base_path = "/console/api/oauth/data-source"
@@ -111,6 +118,18 @@ class DataSourceNotionListApi(Resource):
     @marshal_with(integrate_notion_info_list_fields)
     def get(self):
         dataset_id = request.args.get("dataset_id", default=None, type=str)
+        credential_id = request.args.get("credential_id", default=None, type=str)
+        if not credential_id:
+            raise ValueError("Credential id is required.")
+        datasource_provider_service = DatasourceProviderService()
+        credential = datasource_provider_service.get_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            credential_id=credential_id,
+            provider="notion_datasource",
+            plugin_id="langgenius/notion_datasource",
+        )
+        if not credential:
+            raise NotFound("Credential not found.")
         exist_page_ids = []
         with Session(db.engine) as session:
             # import notion in the exist dataset
@@ -134,31 +153,49 @@ class DataSourceNotionListApi(Resource):
                         data_source_info = json.loads(document.data_source_info)
                         exist_page_ids.append(data_source_info["notion_page_id"])
             # get all authorized pages
-            data_source_bindings = session.scalars(
-                select(DataSourceOauthBinding).filter_by(
-                    tenant_id=current_user.current_tenant_id, provider="notion", disabled=False
+            from core.datasource.datasource_manager import DatasourceManager
+
+            datasource_runtime = DatasourceManager.get_datasource_runtime(
+                provider_id="langgenius/notion_datasource/notion_datasource",
+                datasource_name="notion_datasource",
+                tenant_id=current_user.current_tenant_id,
+                datasource_type=DatasourceProviderType.ONLINE_DOCUMENT,
+            )
+            datasource_provider_service = DatasourceProviderService()
+            if credential:
+                datasource_runtime.runtime.credentials = credential
+            datasource_runtime = cast(OnlineDocumentDatasourcePlugin, datasource_runtime)
+            online_document_result: Generator[OnlineDocumentPagesMessage, None, None] = (
+                datasource_runtime.get_online_document_pages(
+                    user_id=current_user.id,
+                    datasource_parameters={},
+                    provider_type=datasource_runtime.datasource_provider_type(),
                 )
-            ).all()
-            if not data_source_bindings:
-                return {"notion_info": []}, 200
-            pre_import_info_list = []
-            for data_source_binding in data_source_bindings:
-                source_info = data_source_binding.source_info
-                pages = source_info["pages"]
-                # Filter out already bound pages
-                for page in pages:
-                    if page["page_id"] in exist_page_ids:
-                        page["is_bound"] = True
-                    else:
-                        page["is_bound"] = False
-                pre_import_info = {
-                    "workspace_name": source_info["workspace_name"],
-                    "workspace_icon": source_info["workspace_icon"],
-                    "workspace_id": source_info["workspace_id"],
-                    "pages": pages,
-                }
-                pre_import_info_list.append(pre_import_info)
-            return {"notion_info": pre_import_info_list}, 200
+            )
+            try:
+                pages = []
+                workspace_info = {}
+                for message in online_document_result:
+                    result = message.result
+                    for info in result:
+                        workspace_info = {
+                            "workspace_id": info.workspace_id,
+                            "workspace_name": info.workspace_name,
+                            "workspace_icon": info.workspace_icon,
+                        }
+                        for page in info.pages:
+                            page_info = {
+                                "page_id": page.page_id,
+                                "page_name": page.page_name,
+                                "type": page.type,
+                                "parent_id": page.parent_id,
+                                "is_bound": page.page_id in exist_page_ids,
+                                "page_icon": page.page_icon,
+                            }
+                            pages.append(page_info)
+            except Exception as e:
+                raise e
+            return {"notion_info": {**workspace_info, "pages": pages}}, 200
 
 
 class DataSourceNotionApi(Resource):
@@ -166,27 +203,25 @@ class DataSourceNotionApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, workspace_id, page_id, page_type):
+        credential_id = request.args.get("credential_id", default=None, type=str)
+        if not credential_id:
+            raise ValueError("Credential id is required.")
+        datasource_provider_service = DatasourceProviderService()
+        credential = datasource_provider_service.get_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            credential_id=credential_id,
+            provider="notion_datasource",
+            plugin_id="langgenius/notion_datasource",
+        )
+
         workspace_id = str(workspace_id)
         page_id = str(page_id)
-        with Session(db.engine) as session:
-            data_source_binding = session.execute(
-                select(DataSourceOauthBinding).where(
-                    db.and_(
-                        DataSourceOauthBinding.tenant_id == current_user.current_tenant_id,
-                        DataSourceOauthBinding.provider == "notion",
-                        DataSourceOauthBinding.disabled == False,
-                        DataSourceOauthBinding.source_info["workspace_id"] == f'"{workspace_id}"',
-                    )
-                )
-            ).scalar_one_or_none()
-        if not data_source_binding:
-            raise NotFound("Data source binding not found.")
 
         extractor = NotionExtractor(
             notion_workspace_id=workspace_id,
             notion_obj_id=page_id,
             notion_page_type=page_type,
-            notion_access_token=data_source_binding.access_token,
+            notion_access_token=credential.get("integration_secret"),
             tenant_id=current_user.current_tenant_id,
         )
 
@@ -211,10 +246,12 @@ class DataSourceNotionApi(Resource):
         extract_settings = []
         for notion_info in notion_info_list:
             workspace_id = notion_info["workspace_id"]
+            credential_id = notion_info.get("credential_id")
             for page in notion_info["pages"]:
                 extract_setting = ExtractSetting(
                     datasource_type=DatasourceType.NOTION.value,
                     notion_info={
+                        "credential_id": credential_id,
                         "notion_workspace_id": workspace_id,
                         "notion_obj_id": page["page_id"],
                         "notion_page_type": page["type"],
@@ -247,7 +284,7 @@ class DataSourceNotionDatasetSyncApi(Resource):
         documents = DocumentService.get_document_by_dataset_id(dataset_id_str)
         for document in documents:
             document_indexing_sync_task.delay(dataset_id_str, document.id)
-        return {"result": "success"}, 200
+        return 200
 
 
 class DataSourceNotionDocumentSyncApi(Resource):
@@ -265,7 +302,7 @@ class DataSourceNotionDocumentSyncApi(Resource):
         if document is None:
             raise NotFound("Document not found.")
         document_indexing_sync_task.delay(dataset_id_str, document_id_str)
-        return {"result": "success"}, 200
+        return 200
 
 
 api.add_resource(DataSourceApi, "/data-source/integrates", "/data-source/integrates/<uuid:binding_id>/<string:action>")

api/controllers/console/datasets/datasets.py

@@ -2,7 +2,6 @@ import flask_restx
 from flask import request
 from flask_login import current_user
 from flask_restx import Resource, marshal, marshal_with, reqparse
-from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
@@ -20,7 +19,6 @@ from controllers.console.wraps import (
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.indexing_runner import IndexingRunner
 from core.model_runtime.entities.model_entities import ModelType
-from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.extractor.entity.datasource_type import DatasourceType
@@ -33,6 +31,7 @@ from fields.document_fields import document_status_fields
 from libs.login import login_required
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermissionEnum
+from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 
 
@@ -281,6 +280,15 @@ class DatasetApi(Resource):
             location="json",
             help="Invalid external knowledge api id.",
         )
+
+        parser.add_argument(
+            "icon_info",
+            type=dict,
+            required=False,
+            nullable=True,
+            location="json",
+            help="Invalid icon info.",
+        )
         args = parser.parse_args()
         data = request.get_json()
 
@@ -412,11 +420,11 @@ class DatasetIndexingEstimateApi(Resource):
         extract_settings = []
         if args["info_list"]["data_source_type"] == "upload_file":
             file_ids = args["info_list"]["file_info_list"]["file_ids"]
-            file_details = db.session.scalars(
-                select(UploadFile).where(
-                    UploadFile.tenant_id == current_user.current_tenant_id, UploadFile.id.in_(file_ids)
-                )
-            ).all()
+            file_details = (
+                db.session.query(UploadFile)
+                .where(UploadFile.tenant_id == current_user.current_tenant_id, UploadFile.id.in_(file_ids))
+                .all()
+            )
 
             if file_details is None:
                 raise NotFound("File not found.")
@@ -433,10 +441,12 @@ class DatasetIndexingEstimateApi(Resource):
             notion_info_list = args["info_list"]["notion_info_list"]
             for notion_info in notion_info_list:
                 workspace_id = notion_info["workspace_id"]
+                credential_id = notion_info.get("credential_id")
                 for page in notion_info["pages"]:
                     extract_setting = ExtractSetting(
                         datasource_type=DatasourceType.NOTION.value,
                         notion_info={
+                            "credential_id": credential_id,
                             "notion_workspace_id": workspace_id,
                             "notion_obj_id": page["page_id"],
                             "notion_page_type": page["type"],
@@ -519,11 +529,11 @@ class DatasetIndexingStatusApi(Resource):
     @account_initialization_required
     def get(self, dataset_id):
         dataset_id = str(dataset_id)
-        documents = db.session.scalars(
-            select(Document).where(
-                Document.dataset_id == dataset_id, Document.tenant_id == current_user.current_tenant_id
-            )
-        ).all()
+        documents = (
+            db.session.query(Document)
+            .where(Document.dataset_id == dataset_id, Document.tenant_id == current_user.current_tenant_id)
+            .all()
+        )
         documents_status = []
         for document in documents:
             completed_segments = (
@@ -570,11 +580,11 @@ class DatasetApiKeyApi(Resource):
     @account_initialization_required
     @marshal_with(api_key_list)
     def get(self):
-        keys = db.session.scalars(
-            select(ApiToken).where(
-                ApiToken.type == self.resource_type, ApiToken.tenant_id == current_user.current_tenant_id
-            )
-        ).all()
+        keys = (
+            db.session.query(ApiToken)
+            .where(ApiToken.type == self.resource_type, ApiToken.tenant_id == current_user.current_tenant_id)
+            .all()
+        )
         return {"items": keys}
 
     @setup_required

api/controllers/console/datasets/datasets_document.py

@@ -1,6 +1,6 @@
+import json
 import logging
 from argparse import ArgumentTypeError
-from collections.abc import Sequence
 from typing import Literal, cast
 
 from flask import request
@@ -53,6 +53,7 @@ from fields.document_fields import (
 from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import Dataset, DatasetProcessRule, Document, DocumentSegment, UploadFile
+from models.dataset import DocumentPipelineExecutionLog
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig
 
@@ -80,7 +81,7 @@ class DocumentResource(Resource):
 
         return document
 
-    def get_batch_documents(self, dataset_id: str, batch: str) -> Sequence[Document]:
+    def get_batch_documents(self, dataset_id: str, batch: str) -> list[Document]:
         dataset = DatasetService.get_dataset(dataset_id)
         if not dataset:
             raise NotFound("Dataset not found.")
@@ -499,6 +500,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 extract_setting = ExtractSetting(
                     datasource_type=DatasourceType.NOTION.value,
                     notion_info={
+                        "credential_id": data_source_info["credential_id"],
                         "notion_workspace_id": data_source_info["notion_workspace_id"],
                         "notion_obj_id": data_source_info["notion_page_id"],
                         "notion_page_type": data_source_info["type"],
@@ -654,7 +656,7 @@ class DocumentApi(DocumentResource):
             response = {"id": document.id, "doc_type": document.doc_type, "doc_metadata": document.doc_metadata_details}
         elif metadata == "without":
             dataset_process_rules = DatasetService.get_process_rules(dataset_id)
-            document_process_rules = document.dataset_process_rule.to_dict()
+            document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
             data_source_info = document.data_source_detail_dict
             response = {
                 "id": document.id,
@@ -1017,6 +1019,41 @@ class WebsiteDocumentSyncApi(DocumentResource):
         return {"result": "success"}, 200
 
 
+class DocumentPipelineExecutionLogApi(DocumentResource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, document_id):
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        document = DocumentService.get_document(dataset.id, document_id)
+        if not document:
+            raise NotFound("Document not found.")
+        log = (
+            db.session.query(DocumentPipelineExecutionLog)
+            .filter_by(document_id=document_id)
+            .order_by(DocumentPipelineExecutionLog.created_at.desc())
+            .first()
+        )
+        if not log:
+            return {
+                "datasource_info": None,
+                "datasource_type": None,
+                "input_data": None,
+                "datasource_node_id": None,
+            }, 200
+        return {
+            "datasource_info": json.loads(log.datasource_info),
+            "datasource_type": log.datasource_type,
+            "input_data": log.input_data,
+            "datasource_node_id": log.datasource_node_id,
+        }, 200
+
+
 api.add_resource(GetProcessRuleApi, "/datasets/process-rule")
 api.add_resource(DatasetDocumentListApi, "/datasets/<uuid:dataset_id>/documents")
 api.add_resource(DatasetInitApi, "/datasets/init")
@@ -1038,3 +1075,6 @@ api.add_resource(DocumentRetryApi, "/datasets/<uuid:dataset_id>/retry")
 api.add_resource(DocumentRenameApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/rename")
 
 api.add_resource(WebsiteDocumentSyncApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/website-sync")
+api.add_resource(
+    DocumentPipelineExecutionLogApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/pipeline-execution-log"
+)

api/controllers/console/datasets/error.py

@@ -71,3 +71,9 @@ class ChildChunkDeleteIndexError(BaseHTTPException):
     error_code = "child_chunk_delete_index_error"
     description = "Delete child chunk index failed: {message}"
     code = 500
+
+
+class PipelineNotFoundError(BaseHTTPException):
+    error_code = "pipeline_not_found"
+    description = "Pipeline not found."
+    code = 404

api/controllers/console/datasets/metadata.py

@@ -113,7 +113,7 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
             MetadataService.enable_built_in_field(dataset)
         elif action == "disable":
             MetadataService.disable_built_in_field(dataset)
-        return {"result": "success"}, 200
+        return 200
 
 
 class DocumentMetadataEditApi(Resource):
@@ -135,7 +135,7 @@ class DocumentMetadataEditApi(Resource):
 
         MetadataService.update_documents_metadata(dataset, metadata_args)
 
-        return {"result": "success"}, 200
+        return 200
 
 
 api.add_resource(DatasetMetadataCreateApi, "/datasets/<uuid:dataset_id>/metadata")

api/controllers/console/datasets/rag_pipeline/datasource_auth.py

@@ -0,0 +1,362 @@
+from fastapi.encoders import jsonable_encoder
+from flask import make_response, redirect, request
+from flask_login import current_user
+from flask_restx import Resource, reqparse
+from werkzeug.exceptions import Forbidden, NotFound
+
+from configs import dify_config
+from controllers.console import api
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from core.model_runtime.errors.validate import CredentialsValidateFailedError
+from core.plugin.impl.oauth import OAuthHandler
+from libs.helper import StrLen
+from libs.login import login_required
+from models.provider_ids import DatasourceProviderID
+from services.datasource_provider_service import DatasourceProviderService
+from services.plugin.oauth_service import OAuthProxyService
+
+
+class DatasourcePluginOAuthAuthorizationUrl(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider_id: str):
+        user = current_user
+        tenant_id = user.current_tenant_id
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        credential_id = request.args.get("credential_id")
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        provider_name = datasource_provider_id.provider_name
+        plugin_id = datasource_provider_id.plugin_id
+        oauth_config = DatasourceProviderService().get_oauth_client(
+            tenant_id=tenant_id,
+            datasource_provider_id=datasource_provider_id,
+        )
+        if not oauth_config:
+            raise ValueError(f"No OAuth Client Config for {provider_id}")
+
+        context_id = OAuthProxyService.create_proxy_context(
+            user_id=current_user.id,
+            tenant_id=tenant_id,
+            plugin_id=plugin_id,
+            provider=provider_name,
+            credential_id=credential_id,
+        )
+        oauth_handler = OAuthHandler()
+        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/datasource/callback"
+        authorization_url_response = oauth_handler.get_authorization_url(
+            tenant_id=tenant_id,
+            user_id=user.id,
+            plugin_id=plugin_id,
+            provider=provider_name,
+            redirect_uri=redirect_uri,
+            system_credentials=oauth_config,
+        )
+        response = make_response(jsonable_encoder(authorization_url_response))
+        response.set_cookie(
+            "context_id",
+            context_id,
+            httponly=True,
+            samesite="Lax",
+            max_age=OAuthProxyService.__MAX_AGE__,
+        )
+        return response
+
+
+class DatasourceOAuthCallback(Resource):
+    @setup_required
+    def get(self, provider_id: str):
+        context_id = request.cookies.get("context_id") or request.args.get("context_id")
+        if not context_id:
+            raise Forbidden("context_id not found")
+
+        context = OAuthProxyService.use_proxy_context(context_id)
+        if context is None:
+            raise Forbidden("Invalid context_id")
+
+        user_id, tenant_id = context.get("user_id"), context.get("tenant_id")
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        plugin_id = datasource_provider_id.plugin_id
+        datasource_provider_service = DatasourceProviderService()
+        oauth_client_params = datasource_provider_service.get_oauth_client(
+            tenant_id=tenant_id,
+            datasource_provider_id=datasource_provider_id,
+        )
+        if not oauth_client_params:
+            raise NotFound()
+        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/datasource/callback"
+        oauth_handler = OAuthHandler()
+        oauth_response = oauth_handler.get_credentials(
+            tenant_id=tenant_id,
+            user_id=user_id,
+            plugin_id=plugin_id,
+            provider=datasource_provider_id.provider_name,
+            redirect_uri=redirect_uri,
+            system_credentials=oauth_client_params,
+            request=request,
+        )
+        credential_id = context.get("credential_id")
+        if credential_id:
+            datasource_provider_service.reauthorize_datasource_oauth_provider(
+                tenant_id=tenant_id,
+                provider_id=datasource_provider_id,
+                avatar_url=oauth_response.metadata.get("avatar_url") or None,
+                name=oauth_response.metadata.get("name") or None,
+                expire_at=oauth_response.expires_at,
+                credentials=dict(oauth_response.credentials),
+                credential_id=context.get("credential_id"),
+            )
+        else:
+            datasource_provider_service.add_datasource_oauth_provider(
+                tenant_id=tenant_id,
+                provider_id=datasource_provider_id,
+                avatar_url=oauth_response.metadata.get("avatar_url") or None,
+                name=oauth_response.metadata.get("name") or None,
+                expire_at=oauth_response.expires_at,
+                credentials=dict(oauth_response.credentials),
+            )
+        return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
+
+
+class DatasourceAuth(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name", type=StrLen(max_length=100), required=False, nullable=True, location="json", default=None
+        )
+        parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+
+        try:
+            datasource_provider_service.add_datasource_api_key_provider(
+                tenant_id=current_user.current_tenant_id,
+                provider_id=datasource_provider_id,
+                credentials=args["credentials"],
+                name=args["name"],
+            )
+        except CredentialsValidateFailedError as ex:
+            raise ValueError(str(ex))
+        return {"result": "success"}, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider_id: str):
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasources = datasource_provider_service.list_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            provider=datasource_provider_id.provider_name,
+            plugin_id=datasource_provider_id.plugin_id,
+        )
+        return {"result": datasources}, 200
+
+
+class DatasourceAuthDeleteApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        plugin_id = datasource_provider_id.plugin_id
+        provider_name = datasource_provider_id.provider_name
+        if not current_user.is_editor:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.remove_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            auth_id=args["credential_id"],
+            provider=provider_name,
+            plugin_id=plugin_id,
+        )
+        return {"result": "success"}, 200
+
+
+class DatasourceAuthUpdateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        parser = reqparse.RequestParser()
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("name", type=StrLen(max_length=100), required=False, nullable=True, location="json")
+        parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        if not current_user.is_editor:
+            raise Forbidden()
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.update_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            auth_id=args["credential_id"],
+            provider=datasource_provider_id.provider_name,
+            plugin_id=datasource_provider_id.plugin_id,
+            credentials=args.get("credentials", {}),
+            name=args.get("name", None),
+        )
+        return {"result": "success"}, 201
+
+
+class DatasourceAuthListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        datasource_provider_service = DatasourceProviderService()
+        datasources = datasource_provider_service.get_all_datasource_credentials(
+            tenant_id=current_user.current_tenant_id
+        )
+        return {"result": jsonable_encoder(datasources)}, 200
+
+
+class DatasourceHardCodeAuthListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        datasource_provider_service = DatasourceProviderService()
+        datasources = datasource_provider_service.get_hard_code_datasource_credentials(
+            tenant_id=current_user.current_tenant_id
+        )
+        return {"result": jsonable_encoder(datasources)}, 200
+
+
+class DatasourceAuthOauthCustomClient(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        if not current_user.is_editor:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("client_params", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("enable_oauth_custom_client", type=bool, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.setup_oauth_custom_client_params(
+            tenant_id=current_user.current_tenant_id,
+            datasource_provider_id=datasource_provider_id,
+            client_params=args.get("client_params", {}),
+            enabled=args.get("enable_oauth_custom_client", False),
+        )
+        return {"result": "success"}, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, provider_id: str):
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.remove_oauth_custom_client_params(
+            tenant_id=current_user.current_tenant_id,
+            datasource_provider_id=datasource_provider_id,
+        )
+        return {"result": "success"}, 200
+
+
+class DatasourceAuthDefaultApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        if not current_user.is_editor:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.set_default_datasource_provider(
+            tenant_id=current_user.current_tenant_id,
+            datasource_provider_id=datasource_provider_id,
+            credential_id=args["id"],
+        )
+        return {"result": "success"}, 200
+
+
+class DatasourceUpdateProviderNameApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        if not current_user.is_editor:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("name", type=StrLen(max_length=100), required=True, nullable=False, location="json")
+        parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.update_datasource_provider_name(
+            tenant_id=current_user.current_tenant_id,
+            datasource_provider_id=datasource_provider_id,
+            name=args["name"],
+            credential_id=args["credential_id"],
+        )
+        return {"result": "success"}, 200
+
+
+api.add_resource(
+    DatasourcePluginOAuthAuthorizationUrl,
+    "/oauth/plugin/<path:provider_id>/datasource/get-authorization-url",
+)
+api.add_resource(
+    DatasourceOAuthCallback,
+    "/oauth/plugin/<path:provider_id>/datasource/callback",
+)
+api.add_resource(
+    DatasourceAuth,
+    "/auth/plugin/datasource/<path:provider_id>",
+)
+
+api.add_resource(
+    DatasourceAuthUpdateApi,
+    "/auth/plugin/datasource/<path:provider_id>/update",
+)
+
+api.add_resource(
+    DatasourceAuthDeleteApi,
+    "/auth/plugin/datasource/<path:provider_id>/delete",
+)
+
+api.add_resource(
+    DatasourceAuthListApi,
+    "/auth/plugin/datasource/list",
+)
+
+api.add_resource(
+    DatasourceHardCodeAuthListApi,
+    "/auth/plugin/datasource/default-list",
+)
+
+api.add_resource(
+    DatasourceAuthOauthCustomClient,
+    "/auth/plugin/datasource/<path:provider_id>/custom-client",
+)
+
+api.add_resource(
+    DatasourceAuthDefaultApi,
+    "/auth/plugin/datasource/<path:provider_id>/default",
+)
+
+api.add_resource(
+    DatasourceUpdateProviderNameApi,
+    "/auth/plugin/datasource/<path:provider_id>/update-name",
+)

api/controllers/console/datasets/rag_pipeline/datasource_content_preview.py

@@ -0,0 +1,57 @@
+from flask_restx import (  # type: ignore
+    Resource,  # type: ignore
+    reqparse,
+)
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import api
+from controllers.console.datasets.wraps import get_rag_pipeline
+from controllers.console.wraps import account_initialization_required, setup_required
+from libs.login import current_user, login_required
+from models import Account
+from models.dataset import Pipeline
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+
+
+class DataSourceContentPreviewApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline, node_id: str):
+        """
+        Run datasource content preview
+        """
+        if not isinstance(current_user, Account):
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("datasource_type", type=str, required=True, location="json")
+        parser.add_argument("credential_id", type=str, required=False, location="json")
+        args = parser.parse_args()
+
+        inputs = args.get("inputs")
+        if inputs is None:
+            raise ValueError("missing inputs")
+        datasource_type = args.get("datasource_type")
+        if datasource_type is None:
+            raise ValueError("missing datasource_type")
+
+        rag_pipeline_service = RagPipelineService()
+        preview_content = rag_pipeline_service.run_datasource_node_preview(
+            pipeline=pipeline,
+            node_id=node_id,
+            user_inputs=inputs,
+            account=current_user,
+            datasource_type=datasource_type,
+            is_published=True,
+            credential_id=args.get("credential_id"),
+        )
+        return preview_content, 200
+
+
+api.add_resource(
+    DataSourceContentPreviewApi,
+    "/rag/pipelines/<uuid:pipeline_id>/workflows/published/datasource/nodes/<string:node_id>/preview",
+)

api/controllers/console/datasets/rag_pipeline/rag_pipeline.py

@@ -0,0 +1,164 @@
+import logging
+
+from flask import request
+from flask_restx import Resource, reqparse
+from sqlalchemy.orm import Session
+
+from controllers.console import api
+from controllers.console.wraps import (
+    account_initialization_required,
+    enterprise_license_required,
+    knowledge_pipeline_publish_enabled,
+    setup_required,
+)
+from extensions.ext_database import db
+from libs.login import login_required
+from models.dataset import PipelineCustomizedTemplate
+from services.entities.knowledge_entities.rag_pipeline_entities import PipelineTemplateInfoEntity
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+
+logger = logging.getLogger(__name__)
+
+
+def _validate_name(name):
+    if not name or len(name) < 1 or len(name) > 40:
+        raise ValueError("Name must be between 1 to 40 characters.")
+    return name
+
+
+def _validate_description_length(description):
+    if len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
+class PipelineTemplateListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def get(self):
+        type = request.args.get("type", default="built-in", type=str)
+        language = request.args.get("language", default="en-US", type=str)
+        # get pipeline templates
+        pipeline_templates = RagPipelineService.get_pipeline_templates(type, language)
+        return pipeline_templates, 200
+
+
+class PipelineTemplateDetailApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def get(self, template_id: str):
+        type = request.args.get("type", default="built-in", type=str)
+        rag_pipeline_service = RagPipelineService()
+        pipeline_template = rag_pipeline_service.get_pipeline_template_detail(template_id, type)
+        return pipeline_template, 200
+
+
+class CustomizedPipelineTemplateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def patch(self, template_id: str):
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="Name must be between 1 to 40 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument(
+            "description",
+            type=str,
+            nullable=True,
+            required=False,
+            default="",
+        )
+        parser.add_argument(
+            "icon_info",
+            type=dict,
+            location="json",
+            nullable=True,
+        )
+        args = parser.parse_args()
+        pipeline_template_info = PipelineTemplateInfoEntity(**args)
+        RagPipelineService.update_customized_pipeline_template(template_id, pipeline_template_info)
+        return 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def delete(self, template_id: str):
+        RagPipelineService.delete_customized_pipeline_template(template_id)
+        return 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def post(self, template_id: str):
+        with Session(db.engine) as session:
+            template = (
+                session.query(PipelineCustomizedTemplate).filter(PipelineCustomizedTemplate.id == template_id).first()
+            )
+            if not template:
+                raise ValueError("Customized pipeline template not found.")
+
+        return {"data": template.yaml_content}, 200
+
+
+class PublishCustomizedPipelineTemplateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    @knowledge_pipeline_publish_enabled
+    def post(self, pipeline_id: str):
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="Name must be between 1 to 40 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument(
+            "description",
+            type=str,
+            nullable=True,
+            required=False,
+            default="",
+        )
+        parser.add_argument(
+            "icon_info",
+            type=dict,
+            location="json",
+            nullable=True,
+        )
+        args = parser.parse_args()
+        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service.publish_customized_pipeline_template(pipeline_id, args)
+        return {"result": "success"}
+
+
+api.add_resource(
+    PipelineTemplateListApi,
+    "/rag/pipeline/templates",
+)
+api.add_resource(
+    PipelineTemplateDetailApi,
+    "/rag/pipeline/templates/<string:template_id>",
+)
+api.add_resource(
+    CustomizedPipelineTemplateApi,
+    "/rag/pipeline/customized/templates/<string:template_id>",
+)
+api.add_resource(
+    PublishCustomizedPipelineTemplateApi,
+    "/rag/pipelines/<string:pipeline_id>/customized/publish",
+)

api/controllers/console/datasets/rag_pipeline/rag_pipeline_datasets.py

@@ -0,0 +1,114 @@
+from flask_login import current_user  # type: ignore  # type: ignore
+from flask_restx import Resource, marshal, reqparse  # type: ignore
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+import services
+from controllers.console import api
+from controllers.console.datasets.error import DatasetNameDuplicateError
+from controllers.console.wraps import (
+    account_initialization_required,
+    cloud_edition_billing_rate_limit_check,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.dataset_fields import dataset_detail_fields
+from libs.login import login_required
+from models.dataset import DatasetPermissionEnum
+from services.dataset_service import DatasetPermissionService, DatasetService
+from services.entities.knowledge_entities.rag_pipeline_entities import IconInfo, RagPipelineDatasetCreateEntity
+from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
+
+
+def _validate_name(name):
+    if not name or len(name) < 1 or len(name) > 40:
+        raise ValueError("Name must be between 1 to 40 characters.")
+    return name
+
+
+def _validate_description_length(description):
+    if len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
+class CreateRagPipelineDatasetApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def post(self):
+        parser = reqparse.RequestParser()
+
+        parser.add_argument(
+            "yaml_content",
+            type=str,
+            nullable=False,
+            required=True,
+            help="yaml_content is required.",
+        )
+
+        args = parser.parse_args()
+
+        # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+        rag_pipeline_dataset_create_entity = RagPipelineDatasetCreateEntity(
+            name="",
+            description="",
+            icon_info=IconInfo(
+                icon="📙",
+                icon_background="#FFF4ED",
+                icon_type="emoji",
+            ),
+            permission=DatasetPermissionEnum.ONLY_ME,
+            partial_member_list=None,
+            yaml_content=args["yaml_content"],
+        )
+        try:
+            with Session(db.engine) as session:
+                rag_pipeline_dsl_service = RagPipelineDslService(session)
+                import_info = rag_pipeline_dsl_service.create_rag_pipeline_dataset(
+                    tenant_id=current_user.current_tenant_id,
+                    rag_pipeline_dataset_create_entity=rag_pipeline_dataset_create_entity,
+                )
+            if rag_pipeline_dataset_create_entity.permission == "partial_members":
+                DatasetPermissionService.update_partial_member_list(
+                    current_user.current_tenant_id,
+                    import_info["dataset_id"],
+                    rag_pipeline_dataset_create_entity.partial_member_list,
+                )
+        except services.errors.dataset.DatasetNameDuplicateError:
+            raise DatasetNameDuplicateError()
+
+        return import_info, 201
+
+
+class CreateEmptyRagPipelineDatasetApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def post(self):
+        # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+        dataset = DatasetService.create_empty_rag_pipeline_dataset(
+            tenant_id=current_user.current_tenant_id,
+            rag_pipeline_dataset_create_entity=RagPipelineDatasetCreateEntity(
+                name="",
+                description="",
+                icon_info=IconInfo(
+                    icon="📙",
+                    icon_background="#FFF4ED",
+                    icon_type="emoji",
+                ),
+                permission=DatasetPermissionEnum.ONLY_ME,
+                partial_member_list=None,
+            ),
+        )
+        return marshal(dataset, dataset_detail_fields), 201
+
+
+api.add_resource(CreateRagPipelineDatasetApi, "/rag/pipeline/dataset")
+api.add_resource(CreateEmptyRagPipelineDatasetApi, "/rag/pipeline/empty-dataset")

api/controllers/console/datasets/rag_pipeline/rag_pipeline_draft_variable.py

@@ -0,0 +1,389 @@
+import logging
+from typing import Any, NoReturn
+
+from flask import Response
+from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqparse
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import api
+from controllers.console.app.error import (
+    DraftWorkflowNotExist,
+)
+from controllers.console.app.workflow_draft_variable import (
+    _WORKFLOW_DRAFT_VARIABLE_FIELDS,
+    _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,
+)
+from controllers.console.datasets.wraps import get_rag_pipeline
+from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.variables.segment_group import SegmentGroup
+from core.variables.segments import ArrayFileSegment, FileSegment, Segment
+from core.variables.types import SegmentType
+from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from extensions.ext_database import db
+from factories.file_factory import build_from_mapping, build_from_mappings
+from factories.variable_factory import build_segment_with_type
+from libs.login import current_user, login_required
+from models.account import Account
+from models.dataset import Pipeline
+from models.workflow import WorkflowDraftVariable
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
+
+logger = logging.getLogger(__name__)
+
+
+def _convert_values_to_json_serializable_object(value: Segment) -> Any:
+    if isinstance(value, FileSegment):
+        return value.value.model_dump()
+    elif isinstance(value, ArrayFileSegment):
+        return [i.model_dump() for i in value.value]
+    elif isinstance(value, SegmentGroup):
+        return [_convert_values_to_json_serializable_object(i) for i in value.value]
+    else:
+        return value.value
+
+
+def _serialize_var_value(variable: WorkflowDraftVariable) -> Any:
+    value = variable.get_value()
+    # create a copy of the value to avoid affecting the model cache.
+    value = value.model_copy(deep=True)
+    # Refresh the url signature before returning it to client.
+    if isinstance(value, FileSegment):
+        file = value.value
+        file.remote_url = file.generate_url()
+    elif isinstance(value, ArrayFileSegment):
+        files = value.value
+        for file in files:
+            file.remote_url = file.generate_url()
+    return _convert_values_to_json_serializable_object(value)
+
+
+def _create_pagination_parser():
+    parser = reqparse.RequestParser()
+    parser.add_argument(
+        "page",
+        type=inputs.int_range(1, 100_000),
+        required=False,
+        default=1,
+        location="args",
+        help="the page of data requested",
+    )
+    parser.add_argument("limit", type=inputs.int_range(1, 100), required=False, default=20, location="args")
+    return parser
+
+
+def _get_items(var_list: WorkflowDraftVariableList) -> list[WorkflowDraftVariable]:
+    return var_list.variables
+
+
+_WORKFLOW_DRAFT_VARIABLE_LIST_WITHOUT_VALUE_FIELDS = {
+    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS), attribute=_get_items),
+    "total": fields.Raw(),
+}
+
+_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS = {
+    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_VARIABLE_FIELDS), attribute=_get_items),
+}
+
+
+def _api_prerequisite(f):
+    """Common prerequisites for all draft workflow variable APIs.
+
+    It ensures the following conditions are satisfied:
+
+    - Dify has been property setup.
+    - The request user has logged in and initialized.
+    - The requested app is a workflow or a chat flow.
+    - The request user has the edit permission for the app.
+    """
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def wrapper(*args, **kwargs):
+        if not isinstance(current_user, Account) or not current_user.is_editor:
+            raise Forbidden()
+        return f(*args, **kwargs)
+
+    return wrapper
+
+
+class RagPipelineVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_WITHOUT_VALUE_FIELDS)
+    def get(self, pipeline: Pipeline):
+        """
+        Get draft workflow
+        """
+        parser = _create_pagination_parser()
+        args = parser.parse_args()
+
+        # fetch draft workflow by app_model
+        rag_pipeline_service = RagPipelineService()
+        workflow_exist = rag_pipeline_service.is_workflow_exist(pipeline=pipeline)
+        if not workflow_exist:
+            raise DraftWorkflowNotExist()
+
+        # fetch draft workflow by app_model
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(
+                session=session,
+            )
+        workflow_vars = draft_var_srv.list_variables_without_values(
+            app_id=pipeline.id,
+            page=args.page,
+            limit=args.limit,
+        )
+
+        return workflow_vars
+
+    @_api_prerequisite
+    def delete(self, pipeline: Pipeline):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        draft_var_srv.delete_workflow_variables(pipeline.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+def validate_node_id(node_id: str) -> NoReturn | None:
+    if node_id in [
+        CONVERSATION_VARIABLE_NODE_ID,
+        SYSTEM_VARIABLE_NODE_ID,
+    ]:
+        # NOTE(QuantumGhost): While we store the system and conversation variables as node variables
+        # with specific `node_id` in database, we still want to make the API separated. By disallowing
+        # accessing system and conversation variables in `WorkflowDraftNodeVariableListApi`,
+        # we mitigate the risk that user of the API depending on the implementation detail of the API.
+        #
+        # ref: [Hyrum's Law](https://www.hyrumslaw.com/)
+
+        raise InvalidArgumentError(
+            f"invalid node_id, please use correspond api for conversation and system variables, node_id={node_id}",
+        )
+    return None
+
+
+class RagPipelineNodeVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
+    def get(self, pipeline: Pipeline, node_id: str):
+        validate_node_id(node_id)
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(
+                session=session,
+            )
+            node_vars = draft_var_srv.list_node_variables(pipeline.id, node_id)
+
+        return node_vars
+
+    @_api_prerequisite
+    def delete(self, pipeline: Pipeline, node_id: str):
+        validate_node_id(node_id)
+        srv = WorkflowDraftVariableService(db.session())
+        srv.delete_node_variables(pipeline.id, node_id)
+        db.session.commit()
+        return Response("", 204)
+
+
+class RagPipelineVariableApi(Resource):
+    _PATCH_NAME_FIELD = "name"
+    _PATCH_VALUE_FIELD = "value"
+
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_FIELDS)
+    def get(self, pipeline: Pipeline, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != pipeline.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        return variable
+
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_FIELDS)
+    def patch(self, pipeline: Pipeline, variable_id: str):
+        # Request payload for file types:
+        #
+        # Local File:
+        #
+        #     {
+        #         "type": "image",
+        #         "transfer_method": "local_file",
+        #         "url": "",
+        #         "upload_file_id": "daded54f-72c7-4f8e-9d18-9b0abdd9f190"
+        #     }
+        #
+        # Remote File:
+        #
+        #
+        #     {
+        #         "type": "image",
+        #         "transfer_method": "remote_url",
+        #         "url": "http://127.0.0.1:5001/files/1602650a-4fe4-423c-85a2-af76c083e3c4/file-preview?timestamp=1750041099&nonce=...&sign=...=",
+        #         "upload_file_id": "1602650a-4fe4-423c-85a2-af76c083e3c4"
+        #     }
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(self._PATCH_NAME_FIELD, type=str, required=False, nullable=True, location="json")
+        # Parse 'value' field as-is to maintain its original data structure
+        parser.add_argument(self._PATCH_VALUE_FIELD, type=lambda x: x, required=False, nullable=True, location="json")
+
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        args = parser.parse_args(strict=True)
+
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != pipeline.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+        new_name = args.get(self._PATCH_NAME_FIELD, None)
+        raw_value = args.get(self._PATCH_VALUE_FIELD, None)
+        if new_name is None and raw_value is None:
+            return variable
+
+        new_value = None
+        if raw_value is not None:
+            if variable.value_type == SegmentType.FILE:
+                if not isinstance(raw_value, dict):
+                    raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
+                raw_value = build_from_mapping(mapping=raw_value, tenant_id=pipeline.tenant_id)
+            elif variable.value_type == SegmentType.ARRAY_FILE:
+                if not isinstance(raw_value, list):
+                    raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
+                if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
+                    raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
+                raw_value = build_from_mappings(mappings=raw_value, tenant_id=pipeline.tenant_id)
+            new_value = build_segment_with_type(variable.value_type, raw_value)
+        draft_var_srv.update_variable(variable, name=new_name, value=new_value)
+        db.session.commit()
+        return variable
+
+    @_api_prerequisite
+    def delete(self, pipeline: Pipeline, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != pipeline.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        draft_var_srv.delete_variable(variable)
+        db.session.commit()
+        return Response("", 204)
+
+
+class RagPipelineVariableResetApi(Resource):
+    @_api_prerequisite
+    def put(self, pipeline: Pipeline, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+
+        rag_pipeline_service = RagPipelineService()
+        draft_workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
+        if draft_workflow is None:
+            raise NotFoundError(
+                f"Draft workflow not found, pipeline_id={pipeline.id}",
+            )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != pipeline.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+        resetted = draft_var_srv.reset_variable(draft_workflow, variable)
+        db.session.commit()
+        if resetted is None:
+            return Response("", 204)
+        else:
+            return marshal(resetted, _WORKFLOW_DRAFT_VARIABLE_FIELDS)
+
+
+def _get_variable_list(pipeline: Pipeline, node_id) -> WorkflowDraftVariableList:
+    with Session(bind=db.engine, expire_on_commit=False) as session:
+        draft_var_srv = WorkflowDraftVariableService(
+            session=session,
+        )
+        if node_id == CONVERSATION_VARIABLE_NODE_ID:
+            draft_vars = draft_var_srv.list_conversation_variables(pipeline.id)
+        elif node_id == SYSTEM_VARIABLE_NODE_ID:
+            draft_vars = draft_var_srv.list_system_variables(pipeline.id)
+        else:
+            draft_vars = draft_var_srv.list_node_variables(app_id=pipeline.id, node_id=node_id)
+    return draft_vars
+
+
+class RagPipelineSystemVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
+    def get(self, pipeline: Pipeline):
+        return _get_variable_list(pipeline, SYSTEM_VARIABLE_NODE_ID)
+
+
+class RagPipelineEnvironmentVariableCollectionApi(Resource):
+    @_api_prerequisite
+    def get(self, pipeline: Pipeline):
+        """
+        Get draft workflow
+        """
+        # fetch draft workflow by app_model
+        rag_pipeline_service = RagPipelineService()
+        workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
+        if workflow is None:
+            raise DraftWorkflowNotExist()
+
+        env_vars = workflow.environment_variables
+        env_vars_list = []
+        for v in env_vars:
+            env_vars_list.append(
+                {
+                    "id": v.id,
+                    "type": "env",
+                    "name": v.name,
+                    "description": v.description,
+                    "selector": v.selector,
+                    "value_type": v.value_type.value,
+                    "value": v.value,
+                    # Do not track edited for env vars.
+                    "edited": False,
+                    "visible": True,
+                    "editable": True,
+                }
+            )
+
+        return {"items": env_vars_list}
+
+
+api.add_resource(
+    RagPipelineVariableCollectionApi,
+    "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables",
+)
+api.add_resource(
+    RagPipelineNodeVariableCollectionApi,
+    "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/variables",
+)
+api.add_resource(
+    RagPipelineVariableApi, "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables/<uuid:variable_id>"
+)
+api.add_resource(
+    RagPipelineVariableResetApi, "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables/<uuid:variable_id>/reset"
+)
+api.add_resource(
+    RagPipelineSystemVariableCollectionApi, "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/system-variables"
+)
+api.add_resource(
+    RagPipelineEnvironmentVariableCollectionApi,
+    "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/environment-variables",
+)

api/controllers/console/datasets/rag_pipeline/rag_pipeline_import.py

@@ -0,0 +1,147 @@
+from typing import cast
+
+from flask_login import current_user  # type: ignore
+from flask_restx import Resource, marshal_with, reqparse  # type: ignore
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import api
+from controllers.console.datasets.wraps import get_rag_pipeline
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.rag_pipeline_fields import pipeline_import_check_dependencies_fields, pipeline_import_fields
+from libs.login import login_required
+from models import Account
+from models.dataset import Pipeline
+from services.app_dsl_service import ImportStatus
+from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
+
+
+class RagPipelineImportApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(pipeline_import_fields)
+    def post(self):
+        # Check user role first
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("mode", type=str, required=True, location="json")
+        parser.add_argument("yaml_content", type=str, location="json")
+        parser.add_argument("yaml_url", type=str, location="json")
+        parser.add_argument("name", type=str, location="json")
+        parser.add_argument("description", type=str, location="json")
+        parser.add_argument("icon_type", type=str, location="json")
+        parser.add_argument("icon", type=str, location="json")
+        parser.add_argument("icon_background", type=str, location="json")
+        parser.add_argument("pipeline_id", type=str, location="json")
+        args = parser.parse_args()
+
+        # Create service with session
+        with Session(db.engine) as session:
+            import_service = RagPipelineDslService(session)
+            # Import app
+            account = cast(Account, current_user)
+            result = import_service.import_rag_pipeline(
+                account=account,
+                import_mode=args["mode"],
+                yaml_content=args.get("yaml_content"),
+                yaml_url=args.get("yaml_url"),
+                pipeline_id=args.get("pipeline_id"),
+                dataset_name=args.get("name"),
+            )
+            session.commit()
+
+        # Return appropriate status code based on result
+        status = result.status
+        if status == ImportStatus.FAILED.value:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING.value:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
+
+
+class RagPipelineImportConfirmApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(pipeline_import_fields)
+    def post(self, import_id):
+        # Check user role first
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        # Create service with session
+        with Session(db.engine) as session:
+            import_service = RagPipelineDslService(session)
+            # Confirm import
+            account = cast(Account, current_user)
+            result = import_service.confirm_import(import_id=import_id, account=account)
+            session.commit()
+
+        # Return appropriate status code based on result
+        if result.status == ImportStatus.FAILED.value:
+            return result.model_dump(mode="json"), 400
+        return result.model_dump(mode="json"), 200
+
+
+class RagPipelineImportCheckDependenciesApi(Resource):
+    @setup_required
+    @login_required
+    @get_rag_pipeline
+    @account_initialization_required
+    @marshal_with(pipeline_import_check_dependencies_fields)
+    def get(self, pipeline: Pipeline):
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        with Session(db.engine) as session:
+            import_service = RagPipelineDslService(session)
+            result = import_service.check_dependencies(pipeline=pipeline)
+
+        return result.model_dump(mode="json"), 200
+
+
+class RagPipelineExportApi(Resource):
+    @setup_required
+    @login_required
+    @get_rag_pipeline
+    @account_initialization_required
+    def get(self, pipeline: Pipeline):
+        if not current_user.is_editor:
+            raise Forbidden()
+
+            # Add include_secret params
+        parser = reqparse.RequestParser()
+        parser.add_argument("include_secret", type=bool, default=False, location="args")
+        args = parser.parse_args()
+
+        with Session(db.engine) as session:
+            export_service = RagPipelineDslService(session)
+            result = export_service.export_rag_pipeline_dsl(pipeline=pipeline, include_secret=args["include_secret"])
+
+        return {"data": result}, 200
+
+
+# Import Rag Pipeline
+api.add_resource(
+    RagPipelineImportApi,
+    "/rag/pipelines/imports",
+)
+api.add_resource(
+    RagPipelineImportConfirmApi,
+    "/rag/pipelines/imports/<string:import_id>/confirm",
+)
+api.add_resource(
+    RagPipelineImportCheckDependenciesApi,
+    "/rag/pipelines/imports/<string:pipeline_id>/check-dependencies",
+)
+api.add_resource(
+    RagPipelineExportApi,
+    "/rag/pipelines/<string:pipeline_id>/exports",
+)

api/controllers/console/datasets/wraps.py

@@ -0,0 +1,47 @@
+from collections.abc import Callable
+from functools import wraps
+from typing import Optional
+
+from controllers.console.datasets.error import PipelineNotFoundError
+from extensions.ext_database import db
+from libs.login import current_user
+from models.account import Account
+from models.dataset import Pipeline
+
+
+def get_rag_pipeline(
+    view: Optional[Callable] = None,
+):
+    def decorator(view_func):
+        @wraps(view_func)
+        def decorated_view(*args, **kwargs):
+            if not kwargs.get("pipeline_id"):
+                raise ValueError("missing pipeline_id in path parameters")
+
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user is not an account")
+
+            pipeline_id = kwargs.get("pipeline_id")
+            pipeline_id = str(pipeline_id)
+
+            del kwargs["pipeline_id"]
+
+            pipeline = (
+                db.session.query(Pipeline)
+                .filter(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_user.current_tenant_id)
+                .first()
+            )
+
+            if not pipeline:
+                raise PipelineNotFoundError()
+
+            kwargs["pipeline"] = pipeline
+
+            return view_func(*args, **kwargs)
+
+        return decorated_view
+
+    if view is None:
+        return decorator
+    else:
+        return decorator(view)

api/controllers/console/explore/completion.py

@@ -1,5 +1,6 @@
 import logging
 
+from flask_login import current_user
 from flask_restx import reqparse
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -27,8 +28,6 @@ from extensions.ext_database import db
 from libs import helper
 from libs.datetime_utils import naive_utc_now
 from libs.helper import uuid_value
-from libs.login import current_user
-from models import Account
 from models.model import AppMode
 from services.app_generate_service import AppGenerateService
 from services.errors.llm import InvokeRateLimitError
@@ -58,8 +57,6 @@ class CompletionApi(InstalledAppResource):
         db.session.commit()
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             response = AppGenerateService.generate(
                 app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=streaming
             )
@@ -93,8 +90,6 @@ class CompletionStopApi(InstalledAppResource):
         if app_model.mode != "completion":
             raise NotCompletionAppError()
 
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         AppQueueManager.set_stop_flag(task_id, InvokeFrom.EXPLORE, current_user.id)
 
         return {"result": "success"}, 200
@@ -122,8 +117,6 @@ class ChatApi(InstalledAppResource):
         db.session.commit()
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             response = AppGenerateService.generate(
                 app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
             )
@@ -160,8 +153,6 @@ class ChatStopApi(InstalledAppResource):
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         AppQueueManager.set_stop_flag(task_id, InvokeFrom.EXPLORE, current_user.id)
 
         return {"result": "success"}, 200

api/controllers/console/explore/conversation.py

@@ -1,3 +1,4 @@
+from flask_login import current_user
 from flask_restx import marshal_with, reqparse
 from flask_restx.inputs import int_range
 from sqlalchemy.orm import Session
@@ -9,8 +10,6 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
 from fields.conversation_fields import conversation_infinite_scroll_pagination_fields, simple_conversation_fields
 from libs.helper import uuid_value
-from libs.login import current_user
-from models import Account
 from models.model import AppMode
 from services.conversation_service import ConversationService
 from services.errors.conversation import ConversationNotExistsError, LastConversationNotExistsError
@@ -36,8 +35,6 @@ class ConversationListApi(InstalledAppResource):
             pinned = args["pinned"] == "true"
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             with Session(db.engine) as session:
                 return WebConversationService.pagination_by_last_id(
                     session=session,
@@ -61,8 +58,6 @@ class ConversationApi(InstalledAppResource):
 
         conversation_id = str(c_id)
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             ConversationService.delete(app_model, conversation_id, current_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
@@ -86,8 +81,6 @@ class ConversationRenameApi(InstalledAppResource):
         args = parser.parse_args()
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             return ConversationService.rename(
                 app_model, conversation_id, current_user, args["name"], args["auto_generate"]
             )
@@ -105,8 +98,6 @@ class ConversationPinApi(InstalledAppResource):
         conversation_id = str(c_id)
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             WebConversationService.pin(app_model, conversation_id, current_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
@@ -122,8 +113,6 @@ class ConversationUnPinApi(InstalledAppResource):
             raise NotChatAppError()
 
         conversation_id = str(c_id)
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         WebConversationService.unpin(app_model, conversation_id, current_user)
 
         return {"result": "success"}

api/controllers/console/explore/installed_app.py

@@ -2,8 +2,9 @@ import logging
 from typing import Any
 
 from flask import request
+from flask_login import current_user
 from flask_restx import Resource, inputs, marshal_with, reqparse
-from sqlalchemy import and_, select
+from sqlalchemy import and_
 from werkzeug.exceptions import BadRequest, Forbidden, NotFound
 
 from controllers.console import api
@@ -12,8 +13,8 @@ from controllers.console.wraps import account_initialization_required, cloud_edi
 from extensions.ext_database import db
 from fields.installed_app_fields import installed_app_list_fields
 from libs.datetime_utils import naive_utc_now
-from libs.login import current_user, login_required
-from models import Account, App, InstalledApp, RecommendedApp
+from libs.login import login_required
+from models import App, InstalledApp, RecommendedApp
 from services.account_service import TenantService
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
@@ -28,23 +29,17 @@ class InstalledAppsListApi(Resource):
     @marshal_with(installed_app_list_fields)
     def get(self):
         app_id = request.args.get("app_id", default=None, type=str)
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         current_tenant_id = current_user.current_tenant_id
 
         if app_id:
-            installed_apps = db.session.scalars(
-                select(InstalledApp).where(
-                    and_(InstalledApp.tenant_id == current_tenant_id, InstalledApp.app_id == app_id)
-                )
-            ).all()
+            installed_apps = (
+                db.session.query(InstalledApp)
+                .where(and_(InstalledApp.tenant_id == current_tenant_id, InstalledApp.app_id == app_id))
+                .all()
+            )
         else:
-            installed_apps = db.session.scalars(
-                select(InstalledApp).where(InstalledApp.tenant_id == current_tenant_id)
-            ).all()
+            installed_apps = db.session.query(InstalledApp).where(InstalledApp.tenant_id == current_tenant_id).all()
 
-        if current_user.current_tenant is None:
-            raise ValueError("current_user.current_tenant must not be None")
         current_user.role = TenantService.get_user_role(current_user, current_user.current_tenant)
         installed_app_list: list[dict[str, Any]] = [
             {
@@ -120,8 +115,6 @@ class InstalledAppsListApi(Resource):
         if recommended_app is None:
             raise NotFound("App not found")
 
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         current_tenant_id = current_user.current_tenant_id
         app = db.session.query(App).where(App.id == args["app_id"]).first()
 
@@ -161,8 +154,6 @@ class InstalledAppApi(InstalledAppResource):
     """
 
     def delete(self, installed_app):
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         if installed_app.app_owner_tenant_id == current_user.current_tenant_id:
             raise BadRequest("You can't uninstall an app owned by the current tenant")
 

api/controllers/console/explore/message.py

@@ -1,5 +1,6 @@
 import logging
 
+from flask_login import current_user
 from flask_restx import marshal_with, reqparse
 from flask_restx.inputs import int_range
 from werkzeug.exceptions import InternalServerError, NotFound
@@ -23,8 +24,6 @@ from core.model_runtime.errors.invoke import InvokeError
 from fields.message_fields import message_infinite_scroll_pagination_fields
 from libs import helper
 from libs.helper import uuid_value
-from libs.login import current_user
-from models import Account
 from models.model import AppMode
 from services.app_generate_service import AppGenerateService
 from services.errors.app import MoreLikeThisDisabledError
@@ -55,8 +54,6 @@ class MessageListApi(InstalledAppResource):
         args = parser.parse_args()
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             return MessageService.pagination_by_first_id(
                 app_model, current_user, args["conversation_id"], args["first_id"], args["limit"]
             )
@@ -78,8 +75,6 @@ class MessageFeedbackApi(InstalledAppResource):
         args = parser.parse_args()
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             MessageService.create_feedback(
                 app_model=app_model,
                 message_id=message_id,
@@ -110,8 +105,6 @@ class MessageMoreLikeThisApi(InstalledAppResource):
         streaming = args["response_mode"] == "streaming"
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             response = AppGenerateService.generate_more_like_this(
                 app_model=app_model,
                 user=current_user,
@@ -149,8 +142,6 @@ class MessageSuggestedQuestionApi(InstalledAppResource):
         message_id = str(message_id)
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             questions = MessageService.get_suggested_questions_after_answer(
                 app_model=app_model, user=current_user, message_id=message_id, invoke_from=InvokeFrom.EXPLORE
             )

api/controllers/console/explore/recommended_app.py

@@ -1,10 +1,11 @@
+from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with, reqparse
 
 from constants.languages import languages
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required
 from libs.helper import AppIconUrlField
-from libs.login import current_user, login_required
+from libs.login import login_required
 from services.recommended_app_service import RecommendedAppService
 
 app_fields = {
@@ -45,9 +46,8 @@ class RecommendedAppListApi(Resource):
         parser.add_argument("language", type=str, location="args")
         args = parser.parse_args()
 
-        language = args.get("language")
-        if language and language in languages:
-            language_prefix = language
+        if args.get("language") and args.get("language") in languages:
+            language_prefix = args.get("language")
         elif current_user and current_user.interface_language:
             language_prefix = current_user.interface_language
         else:

api/controllers/console/explore/saved_message.py

@@ -1,3 +1,4 @@
+from flask_login import current_user
 from flask_restx import fields, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from werkzeug.exceptions import NotFound
@@ -7,8 +8,6 @@ from controllers.console.explore.error import NotCompletionAppError
 from controllers.console.explore.wraps import InstalledAppResource
 from fields.conversation_fields import message_file_fields
 from libs.helper import TimestampField, uuid_value
-from libs.login import current_user
-from models import Account
 from services.errors.message import MessageNotExistsError
 from services.saved_message_service import SavedMessageService
 
@@ -43,8 +42,6 @@ class SavedMessageListApi(InstalledAppResource):
         parser.add_argument("limit", type=int_range(1, 100), required=False, default=20, location="args")
         args = parser.parse_args()
 
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         return SavedMessageService.pagination_by_last_id(app_model, current_user, args["last_id"], args["limit"])
 
     def post(self, installed_app):
@@ -57,8 +54,6 @@ class SavedMessageListApi(InstalledAppResource):
         args = parser.parse_args()
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
             SavedMessageService.save(app_model, current_user, args["message_id"])
         except MessageNotExistsError:
             raise NotFound("Message Not Exists.")
@@ -75,8 +70,6 @@ class SavedMessageApi(InstalledAppResource):
         if app_model.mode != "completion":
             raise NotCompletionAppError()
 
-        if not isinstance(current_user, Account):
-            raise ValueError("current_user must be an Account instance")
         SavedMessageService.delete(app_model, current_user, message_id)
 
         return {"result": "success"}, 204

api/controllers/console/explore/workflow.py

@@ -20,6 +20,7 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.model_runtime.errors.invoke import InvokeError
+from core.workflow.graph_engine.manager import GraphEngineManager
 from libs import helper
 from libs.login import current_user
 from models.model import AppMode, InstalledApp
@@ -82,6 +83,11 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
             raise NotWorkflowAppError()
         assert current_user is not None
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.EXPLORE, current_user.id)
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}

api/controllers/console/extension.py

@@ -1,8 +1,8 @@
 from flask_login import current_user
-from flask_restx import Resource, fields, marshal_with, reqparse
+from flask_restx import Resource, marshal_with, reqparse
 
 from constants import HIDDEN_VALUE
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.wraps import account_initialization_required, setup_required
 from fields.api_based_extension_fields import api_based_extension_fields
 from libs.login import login_required
@@ -11,21 +11,7 @@ from services.api_based_extension_service import APIBasedExtensionService
 from services.code_based_extension_service import CodeBasedExtensionService
 
 
-@console_ns.route("/code-based-extension")
 class CodeBasedExtensionAPI(Resource):
-    @api.doc("get_code_based_extension")
-    @api.doc(description="Get code-based extension data by module name")
-    @api.expect(
-        api.parser().add_argument("module", type=str, required=True, location="args", help="Extension module name")
-    )
-    @api.response(
-        200,
-        "Success",
-        api.model(
-            "CodeBasedExtensionResponse",
-            {"module": fields.String(description="Module name"), "data": fields.Raw(description="Extension data")},
-        ),
-    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -37,11 +23,7 @@ class CodeBasedExtensionAPI(Resource):
         return {"module": args["module"], "data": CodeBasedExtensionService.get_code_based_extension(args["module"])}
 
 
-@console_ns.route("/api-based-extension")
 class APIBasedExtensionAPI(Resource):
-    @api.doc("get_api_based_extensions")
-    @api.doc(description="Get all API-based extensions for current tenant")
-    @api.response(200, "Success", fields.List(fields.Nested(api_based_extension_fields)))
     @setup_required
     @login_required
     @account_initialization_required
@@ -50,19 +32,6 @@ class APIBasedExtensionAPI(Resource):
         tenant_id = current_user.current_tenant_id
         return APIBasedExtensionService.get_all_by_tenant_id(tenant_id)
 
-    @api.doc("create_api_based_extension")
-    @api.doc(description="Create a new API-based extension")
-    @api.expect(
-        api.model(
-            "CreateAPIBasedExtensionRequest",
-            {
-                "name": fields.String(required=True, description="Extension name"),
-                "api_endpoint": fields.String(required=True, description="API endpoint URL"),
-                "api_key": fields.String(required=True, description="API key for authentication"),
-            },
-        )
-    )
-    @api.response(201, "Extension created successfully", api_based_extension_fields)
     @setup_required
     @login_required
     @account_initialization_required
@@ -84,12 +53,7 @@ class APIBasedExtensionAPI(Resource):
         return APIBasedExtensionService.save(extension_data)
 
 
-@console_ns.route("/api-based-extension/<uuid:id>")
 class APIBasedExtensionDetailAPI(Resource):
-    @api.doc("get_api_based_extension")
-    @api.doc(description="Get API-based extension by ID")
-    @api.doc(params={"id": "Extension ID"})
-    @api.response(200, "Success", api_based_extension_fields)
     @setup_required
     @login_required
     @account_initialization_required
@@ -100,20 +64,6 @@ class APIBasedExtensionDetailAPI(Resource):
 
         return APIBasedExtensionService.get_with_tenant_id(tenant_id, api_based_extension_id)
 
-    @api.doc("update_api_based_extension")
-    @api.doc(description="Update API-based extension")
-    @api.doc(params={"id": "Extension ID"})
-    @api.expect(
-        api.model(
-            "UpdateAPIBasedExtensionRequest",
-            {
-                "name": fields.String(required=True, description="Extension name"),
-                "api_endpoint": fields.String(required=True, description="API endpoint URL"),
-                "api_key": fields.String(required=True, description="API key for authentication"),
-            },
-        )
-    )
-    @api.response(200, "Extension updated successfully", api_based_extension_fields)
     @setup_required
     @login_required
     @account_initialization_required
@@ -138,10 +88,6 @@ class APIBasedExtensionDetailAPI(Resource):
 
         return APIBasedExtensionService.save(extension_data_from_db)
 
-    @api.doc("delete_api_based_extension")
-    @api.doc(description="Delete API-based extension")
-    @api.doc(params={"id": "Extension ID"})
-    @api.response(204, "Extension deleted successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -154,3 +100,9 @@ class APIBasedExtensionDetailAPI(Resource):
         APIBasedExtensionService.delete(extension_data_from_db)
 
         return {"result": "success"}, 204
+
+
+api.add_resource(CodeBasedExtensionAPI, "/code-based-extension")
+
+api.add_resource(APIBasedExtensionAPI, "/api-based-extension")
+api.add_resource(APIBasedExtensionDetailAPI, "/api-based-extension/<uuid:id>")

api/controllers/console/feature.py

@@ -1,40 +1,26 @@
 from flask_login import current_user
-from flask_restx import Resource, fields
+from flask_restx import Resource
 
 from libs.login import login_required
 from services.feature_service import FeatureService
 
-from . import api, console_ns
+from . import api
 from .wraps import account_initialization_required, cloud_utm_record, setup_required
 
 
-@console_ns.route("/features")
 class FeatureApi(Resource):
-    @api.doc("get_tenant_features")
-    @api.doc(description="Get feature configuration for current tenant")
-    @api.response(
-        200,
-        "Success",
-        api.model("FeatureResponse", {"features": fields.Raw(description="Feature configuration object")}),
-    )
     @setup_required
     @login_required
     @account_initialization_required
     @cloud_utm_record
     def get(self):
-        """Get feature configuration for current tenant"""
         return FeatureService.get_features(current_user.current_tenant_id).model_dump()
 
 
-@console_ns.route("/system-features")
 class SystemFeatureApi(Resource):
-    @api.doc("get_system_features")
-    @api.doc(description="Get system-wide feature configuration")
-    @api.response(
-        200,
-        "Success",
-        api.model("SystemFeatureResponse", {"features": fields.Raw(description="System feature configuration object")}),
-    )
     def get(self):
-        """Get system-wide feature configuration"""
         return FeatureService.get_system_features().model_dump()
+
+
+api.add_resource(FeatureApi, "/features")
+api.add_resource(SystemFeatureApi, "/system-features")

api/controllers/console/files.py

@@ -20,9 +20,9 @@ from controllers.console.wraps import (
     cloud_edition_billing_resource_check,
     setup_required,
 )
+from extensions.ext_database import db
 from fields.file_fields import file_fields, upload_config_fields
 from libs.login import login_required
-from models import Account
 from services.file_service import FileService
 
 PREVIEW_WORDS_LIMIT = 3000
@@ -69,9 +69,7 @@ class FileApi(Resource):
             source = None
 
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("Invalid user account")
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,
@@ -92,7 +90,7 @@ class FilePreviewApi(Resource):
     @account_initialization_required
     def get(self, file_id):
         file_id = str(file_id)
-        text = FileService.get_file_preview(file_id)
+        text = FileService(db.engine).get_file_preview(file_id)
         return {"content": text}
 
 

api/controllers/console/init_validate.py

@@ -1,7 +1,7 @@
 import os
 
 from flask import session
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, reqparse
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -11,47 +11,20 @@ from libs.helper import StrLen
 from models.model import DifySetup
 from services.account_service import TenantService
 
-from . import api, console_ns
+from . import api
 from .error import AlreadySetupError, InitValidateFailedError
 from .wraps import only_edition_self_hosted
 
 
-@console_ns.route("/init")
 class InitValidateAPI(Resource):
-    @api.doc("get_init_status")
-    @api.doc(description="Get initialization validation status")
-    @api.response(
-        200,
-        "Success",
-        model=api.model(
-            "InitStatusResponse",
-            {"status": fields.String(description="Initialization status", enum=["finished", "not_started"])},
-        ),
-    )
     def get(self):
-        """Get initialization validation status"""
         init_status = get_init_validate_status()
         if init_status:
             return {"status": "finished"}
         return {"status": "not_started"}
 
-    @api.doc("validate_init_password")
-    @api.doc(description="Validate initialization password for self-hosted edition")
-    @api.expect(
-        api.model(
-            "InitValidateRequest",
-            {"password": fields.String(required=True, description="Initialization password", max_length=30)},
-        )
-    )
-    @api.response(
-        201,
-        "Success",
-        model=api.model("InitValidateResponse", {"result": fields.String(description="Operation result")}),
-    )
-    @api.response(400, "Already setup or validation failed")
     @only_edition_self_hosted
     def post(self):
-        """Validate initialization password"""
         # is tenant created
         tenant_count = TenantService.get_tenant_count()
         if tenant_count > 0:
@@ -79,3 +52,6 @@ def get_init_validate_status():
                 return db_session.execute(select(DifySetup)).scalar_one_or_none()
 
     return True
+
+
+api.add_resource(InitValidateAPI, "/init")

api/controllers/console/ping.py

@@ -1,17 +1,14 @@
-from flask_restx import Resource, fields
+from flask_restx import Resource
 
-from . import api, console_ns
+from controllers.console import api
 
 
-@console_ns.route("/ping")
 class PingApi(Resource):
-    @api.doc("health_check")
-    @api.doc(description="Health check endpoint for connection testing")
-    @api.response(
-        200,
-        "Success",
-        api.model("PingResponse", {"result": fields.String(description="Health check result", example="pong")}),
-    )
     def get(self):
-        """Health check endpoint for connection testing"""
+        """
+        For connection health check
+        """
         return {"result": "pong"}
+
+
+api.add_resource(PingApi, "/ping")

api/controllers/console/remote_files.py

@@ -14,6 +14,7 @@ from controllers.common.errors import (
 )
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
+from extensions.ext_database import db
 from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields
 from models.account import Account
 from services.file_service import FileService
@@ -61,7 +62,7 @@ class RemoteFileUploadApi(Resource):
 
         try:
             user = cast(Account, current_user)
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file_info.filename,
                 content=content,
                 mimetype=file_info.mimetype,

api/controllers/console/setup.py

@@ -1,5 +1,5 @@
 from flask import request
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, reqparse
 
 from configs import dify_config
 from libs.helper import StrLen, email, extract_remote_ip
@@ -7,56 +7,23 @@ from libs.password import valid_password
 from models.model import DifySetup, db
 from services.account_service import RegisterService, TenantService
 
-from . import api, console_ns
+from . import api
 from .error import AlreadySetupError, NotInitValidateError
 from .init_validate import get_init_validate_status
 from .wraps import only_edition_self_hosted
 
 
-@console_ns.route("/setup")
 class SetupApi(Resource):
-    @api.doc("get_setup_status")
-    @api.doc(description="Get system setup status")
-    @api.response(
-        200,
-        "Success",
-        api.model(
-            "SetupStatusResponse",
-            {
-                "step": fields.String(description="Setup step status", enum=["not_started", "finished"]),
-                "setup_at": fields.String(description="Setup completion time (ISO format)", required=False),
-            },
-        ),
-    )
     def get(self):
-        """Get system setup status"""
         if dify_config.EDITION == "SELF_HOSTED":
             setup_status = get_setup_status()
-            # Check if setup_status is a DifySetup object rather than a bool
-            if setup_status and not isinstance(setup_status, bool):
+            if setup_status:
                 return {"step": "finished", "setup_at": setup_status.setup_at.isoformat()}
-            elif setup_status:
-                return {"step": "finished"}
             return {"step": "not_started"}
         return {"step": "finished"}
 
-    @api.doc("setup_system")
-    @api.doc(description="Initialize system setup with admin account")
-    @api.expect(
-        api.model(
-            "SetupRequest",
-            {
-                "email": fields.String(required=True, description="Admin email address"),
-                "name": fields.String(required=True, description="Admin name (max 30 characters)"),
-                "password": fields.String(required=True, description="Admin password"),
-            },
-        )
-    )
-    @api.response(201, "Success", api.model("SetupResponse", {"result": fields.String(description="Setup result")}))
-    @api.response(400, "Already setup or validation failed")
     @only_edition_self_hosted
     def post(self):
-        """Initialize system setup with admin account"""
         # is set up
         if get_setup_status():
             raise AlreadySetupError()
@@ -88,3 +55,6 @@ def get_setup_status():
         return db.session.query(DifySetup).first()
     else:
         return True
+
+
+api.add_resource(SetupApi, "/setup")

api/controllers/console/spec.py

@@ -0,0 +1,35 @@
+import logging
+
+from flask_restx import Resource
+
+from controllers.console import api
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from core.schemas.schema_manager import SchemaManager
+from libs.login import login_required
+
+logger = logging.getLogger(__name__)
+
+
+class SpecSchemaDefinitionsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """
+        Get system JSON Schema definitions specification
+        Used for frontend component type mapping
+        """
+        try:
+            schema_manager = SchemaManager()
+            schema_definitions = schema_manager.get_all_schema_definitions()
+            return schema_definitions, 200
+        except Exception:
+            logger.exception("Failed to get schema definitions from local registry")
+            # Return empty array as fallback
+            return [], 200
+
+
+api.add_resource(SpecSchemaDefinitionsApi, "/spec/schema-definitions")

api/controllers/console/tag/tags.py

@@ -111,7 +111,7 @@ class TagBindingCreateApi(Resource):
         args = parser.parse_args()
         TagService.save_tag_binding(args)
 
-        return {"result": "success"}, 200
+        return 200
 
 
 class TagBindingDeleteApi(Resource):
@@ -132,7 +132,7 @@ class TagBindingDeleteApi(Resource):
         args = parser.parse_args()
         TagService.delete_tag_binding(args)
 
-        return {"result": "success"}, 200
+        return 200
 
 
 api.add_resource(TagListApi, "/tags")

api/controllers/console/version.py

@@ -2,41 +2,18 @@ import json
 import logging
 
 import requests
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, reqparse
 from packaging import version
 
 from configs import dify_config
 
-from . import api, console_ns
+from . import api
 
 logger = logging.getLogger(__name__)
 
 
-@console_ns.route("/version")
 class VersionApi(Resource):
-    @api.doc("check_version_update")
-    @api.doc(description="Check for application version updates")
-    @api.expect(
-        api.parser().add_argument(
-            "current_version", type=str, required=True, location="args", help="Current application version"
-        )
-    )
-    @api.response(
-        200,
-        "Success",
-        api.model(
-            "VersionResponse",
-            {
-                "version": fields.String(description="Latest version number"),
-                "release_date": fields.String(description="Release date of latest version"),
-                "release_notes": fields.String(description="Release notes for latest version"),
-                "can_auto_update": fields.Boolean(description="Whether auto-update is supported"),
-                "features": fields.Raw(description="Feature flags and capabilities"),
-            },
-        ),
-    )
     def get(self):
-        """Check for application version updates"""
         parser = reqparse.RequestParser()
         parser.add_argument("current_version", type=str, required=True, location="args")
         args = parser.parse_args()
@@ -57,14 +34,14 @@ class VersionApi(Resource):
             return result
 
         try:
-            response = requests.get(check_update_url, {"current_version": args["current_version"]}, timeout=(3, 10))
+            response = requests.get(check_update_url, {"current_version": args.get("current_version")}, timeout=(3, 10))
         except Exception as error:
             logger.warning("Check update version error: %s.", str(error))
-            result["version"] = args["current_version"]
+            result["version"] = args.get("current_version")
             return result
 
         content = json.loads(response.content)
-        if _has_new_version(latest_version=content["version"], current_version=f"{args['current_version']}"):
+        if _has_new_version(latest_version=content["version"], current_version=f"{args.get('current_version')}"):
             result["version"] = content["version"]
             result["release_date"] = content["releaseDate"]
             result["release_notes"] = content["releaseNotes"]
@@ -82,3 +59,6 @@ def _has_new_version(*, latest_version: str, current_version: str) -> bool:
     except version.InvalidVersion:
         logger.warning("Invalid version format: latest=%s, current=%s", latest_version, current_version)
         return False
+
+
+api.add_resource(VersionApi, "/version")

api/controllers/console/workspace/account.py

@@ -49,8 +49,6 @@ class AccountInitApi(Resource):
     @setup_required
     @login_required
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         account = current_user
 
         if account.status == "active":
@@ -104,8 +102,6 @@ class AccountProfileApi(Resource):
     @marshal_with(account_fields)
     @enterprise_license_required
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         return current_user
 
 
@@ -115,8 +111,6 @@ class AccountNameApi(Resource):
     @account_initialization_required
     @marshal_with(account_fields)
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, location="json")
         args = parser.parse_args()
@@ -136,8 +130,6 @@ class AccountAvatarApi(Resource):
     @account_initialization_required
     @marshal_with(account_fields)
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("avatar", type=str, required=True, location="json")
         args = parser.parse_args()
@@ -153,8 +145,6 @@ class AccountInterfaceLanguageApi(Resource):
     @account_initialization_required
     @marshal_with(account_fields)
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("interface_language", type=supported_language, required=True, location="json")
         args = parser.parse_args()
@@ -170,8 +160,6 @@ class AccountInterfaceThemeApi(Resource):
     @account_initialization_required
     @marshal_with(account_fields)
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("interface_theme", type=str, choices=["light", "dark"], required=True, location="json")
         args = parser.parse_args()
@@ -187,8 +175,6 @@ class AccountTimezoneApi(Resource):
     @account_initialization_required
     @marshal_with(account_fields)
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("timezone", type=str, required=True, location="json")
         args = parser.parse_args()
@@ -208,8 +194,6 @@ class AccountPasswordApi(Resource):
     @account_initialization_required
     @marshal_with(account_fields)
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("password", type=str, required=False, location="json")
         parser.add_argument("new_password", type=str, required=True, location="json")
@@ -244,13 +228,9 @@ class AccountIntegrateApi(Resource):
     @account_initialization_required
     @marshal_with(integrate_list_fields)
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         account = current_user
 
-        account_integrates = db.session.scalars(
-            select(AccountIntegrate).where(AccountIntegrate.account_id == account.id)
-        ).all()
+        account_integrates = db.session.query(AccountIntegrate).where(AccountIntegrate.account_id == account.id).all()
 
         base_url = request.url_root.rstrip("/")
         oauth_base_path = "/console/api/oauth/login"
@@ -288,8 +268,6 @@ class AccountDeleteVerifyApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         account = current_user
 
         token, code = AccountService.generate_account_deletion_verification_code(account)
@@ -303,8 +281,6 @@ class AccountDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         account = current_user
 
         parser = reqparse.RequestParser()
@@ -345,8 +321,6 @@ class EducationVerifyApi(Resource):
     @cloud_edition_billing_enabled
     @marshal_with(verify_fields)
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         account = current_user
 
         return BillingService.EducationIdentity.verify(account.id, account.email)
@@ -366,8 +340,6 @@ class EducationApi(Resource):
     @only_edition_cloud
     @cloud_edition_billing_enabled
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         account = current_user
 
         parser = reqparse.RequestParser()
@@ -385,8 +357,6 @@ class EducationApi(Resource):
     @cloud_edition_billing_enabled
     @marshal_with(status_fields)
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         account = current_user
 
         res = BillingService.EducationIdentity.status(account.id)
@@ -451,8 +421,6 @@ class ChangeEmailSendEmailApi(Resource):
                 raise InvalidTokenError()
             user_email = reset_data.get("email", "")
 
-            if not isinstance(current_user, Account):
-                raise ValueError("Invalid user account")
             if user_email != current_user.email:
                 raise InvalidEmailError()
         else:
@@ -533,8 +501,6 @@ class ChangeEmailResetApi(Resource):
         AccountService.revoke_change_email_token(args["token"])
 
         old_email = reset_data.get("old_email", "")
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         if current_user.email != old_email:
             raise AccountNotFound()
 

api/controllers/console/workspace/agent_providers.py

@@ -1,22 +1,14 @@
 from flask_login import current_user
-from flask_restx import Resource, fields
+from flask_restx import Resource
 
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import login_required
 from services.agent_service import AgentService
 
 
-@console_ns.route("/workspaces/current/agent-providers")
 class AgentProviderListApi(Resource):
-    @api.doc("list_agent_providers")
-    @api.doc(description="Get list of available agent providers")
-    @api.response(
-        200,
-        "Success",
-        fields.List(fields.Raw(description="Agent provider information")),
-    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -29,16 +21,7 @@ class AgentProviderListApi(Resource):
         return jsonable_encoder(AgentService.list_agent_providers(user_id, tenant_id))
 
 
-@console_ns.route("/workspaces/current/agent-provider/<path:provider_name>")
 class AgentProviderApi(Resource):
-    @api.doc("get_agent_provider")
-    @api.doc(description="Get specific agent provider details")
-    @api.doc(params={"provider_name": "Agent provider name"})
-    @api.response(
-        200,
-        "Success",
-        fields.Raw(description="Agent provider details"),
-    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -47,3 +30,7 @@ class AgentProviderApi(Resource):
         user_id = user.id
         tenant_id = user.current_tenant_id
         return jsonable_encoder(AgentService.get_agent_provider(user_id, tenant_id, provider_name))
+
+
+api.add_resource(AgentProviderListApi, "/workspaces/current/agent-providers")
+api.add_resource(AgentProviderApi, "/workspaces/current/agent-provider/<path:provider_name>")

api/controllers/console/workspace/endpoint.py

@@ -1,8 +1,8 @@
 from flask_login import current_user
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, reqparse
 from werkzeug.exceptions import Forbidden
 
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.impl.exc import PluginPermissionDeniedError
@@ -10,26 +10,7 @@ from libs.login import login_required
 from services.plugin.endpoint_service import EndpointService
 
 
-@console_ns.route("/workspaces/current/endpoints/create")
 class EndpointCreateApi(Resource):
-    @api.doc("create_endpoint")
-    @api.doc(description="Create a new plugin endpoint")
-    @api.expect(
-        api.model(
-            "EndpointCreateRequest",
-            {
-                "plugin_unique_identifier": fields.String(required=True, description="Plugin unique identifier"),
-                "settings": fields.Raw(required=True, description="Endpoint settings"),
-                "name": fields.String(required=True, description="Endpoint name"),
-            },
-        )
-    )
-    @api.response(
-        200,
-        "Endpoint created successfully",
-        api.model("EndpointCreateResponse", {"success": fields.Boolean(description="Operation success")}),
-    )
-    @api.response(403, "Admin privileges required")
     @setup_required
     @login_required
     @account_initialization_required
@@ -62,20 +43,7 @@ class EndpointCreateApi(Resource):
             raise ValueError(e.description) from e
 
 
-@console_ns.route("/workspaces/current/endpoints/list")
 class EndpointListApi(Resource):
-    @api.doc("list_endpoints")
-    @api.doc(description="List plugin endpoints with pagination")
-    @api.expect(
-        api.parser()
-        .add_argument("page", type=int, required=True, location="args", help="Page number")
-        .add_argument("page_size", type=int, required=True, location="args", help="Page size")
-    )
-    @api.response(
-        200,
-        "Success",
-        api.model("EndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}),
-    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -102,23 +70,7 @@ class EndpointListApi(Resource):
         )
 
 
-@console_ns.route("/workspaces/current/endpoints/list/plugin")
 class EndpointListForSinglePluginApi(Resource):
-    @api.doc("list_plugin_endpoints")
-    @api.doc(description="List endpoints for a specific plugin")
-    @api.expect(
-        api.parser()
-        .add_argument("page", type=int, required=True, location="args", help="Page number")
-        .add_argument("page_size", type=int, required=True, location="args", help="Page size")
-        .add_argument("plugin_id", type=str, required=True, location="args", help="Plugin ID")
-    )
-    @api.response(
-        200,
-        "Success",
-        api.model(
-            "PluginEndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
-    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -148,19 +100,7 @@ class EndpointListForSinglePluginApi(Resource):
         )
 
 
-@console_ns.route("/workspaces/current/endpoints/delete")
 class EndpointDeleteApi(Resource):
-    @api.doc("delete_endpoint")
-    @api.doc(description="Delete a plugin endpoint")
-    @api.expect(
-        api.model("EndpointDeleteRequest", {"endpoint_id": fields.String(required=True, description="Endpoint ID")})
-    )
-    @api.response(
-        200,
-        "Endpoint deleted successfully",
-        api.model("EndpointDeleteResponse", {"success": fields.Boolean(description="Operation success")}),
-    )
-    @api.response(403, "Admin privileges required")
     @setup_required
     @login_required
     @account_initialization_required
@@ -183,26 +123,7 @@ class EndpointDeleteApi(Resource):
         }
 
 
-@console_ns.route("/workspaces/current/endpoints/update")
 class EndpointUpdateApi(Resource):
-    @api.doc("update_endpoint")
-    @api.doc(description="Update a plugin endpoint")
-    @api.expect(
-        api.model(
-            "EndpointUpdateRequest",
-            {
-                "endpoint_id": fields.String(required=True, description="Endpoint ID"),
-                "settings": fields.Raw(required=True, description="Updated settings"),
-                "name": fields.String(required=True, description="Updated name"),
-            },
-        )
-    )
-    @api.response(
-        200,
-        "Endpoint updated successfully",
-        api.model("EndpointUpdateResponse", {"success": fields.Boolean(description="Operation success")}),
-    )
-    @api.response(403, "Admin privileges required")
     @setup_required
     @login_required
     @account_initialization_required
@@ -233,19 +154,7 @@ class EndpointUpdateApi(Resource):
         }
 
 
-@console_ns.route("/workspaces/current/endpoints/enable")
 class EndpointEnableApi(Resource):
-    @api.doc("enable_endpoint")
-    @api.doc(description="Enable a plugin endpoint")
-    @api.expect(
-        api.model("EndpointEnableRequest", {"endpoint_id": fields.String(required=True, description="Endpoint ID")})
-    )
-    @api.response(
-        200,
-        "Endpoint enabled successfully",
-        api.model("EndpointEnableResponse", {"success": fields.Boolean(description="Operation success")}),
-    )
-    @api.response(403, "Admin privileges required")
     @setup_required
     @login_required
     @account_initialization_required
@@ -268,19 +177,7 @@ class EndpointEnableApi(Resource):
         }
 
 
-@console_ns.route("/workspaces/current/endpoints/disable")
 class EndpointDisableApi(Resource):
-    @api.doc("disable_endpoint")
-    @api.doc(description="Disable a plugin endpoint")
-    @api.expect(
-        api.model("EndpointDisableRequest", {"endpoint_id": fields.String(required=True, description="Endpoint ID")})
-    )
-    @api.response(
-        200,
-        "Endpoint disabled successfully",
-        api.model("EndpointDisableResponse", {"success": fields.Boolean(description="Operation success")}),
-    )
-    @api.response(403, "Admin privileges required")
     @setup_required
     @login_required
     @account_initialization_required
@@ -301,3 +198,12 @@ class EndpointDisableApi(Resource):
                 tenant_id=user.current_tenant_id, user_id=user.id, endpoint_id=endpoint_id
             )
         }
+
+
+api.add_resource(EndpointCreateApi, "/workspaces/current/endpoints/create")
+api.add_resource(EndpointListApi, "/workspaces/current/endpoints/list")
+api.add_resource(EndpointListForSinglePluginApi, "/workspaces/current/endpoints/list/plugin")
+api.add_resource(EndpointDeleteApi, "/workspaces/current/endpoints/delete")
+api.add_resource(EndpointUpdateApi, "/workspaces/current/endpoints/update")
+api.add_resource(EndpointEnableApi, "/workspaces/current/endpoints/enable")
+api.add_resource(EndpointDisableApi, "/workspaces/current/endpoints/disable")

api/controllers/console/workspace/members.py

@@ -1,8 +1,8 @@
 from urllib import parse
 
-from flask import abort, request
+from flask import request
 from flask_login import current_user
-from flask_restx import Resource, marshal_with, reqparse
+from flask_restx import Resource, abort, marshal_with, reqparse
 
 import services
 from configs import dify_config
@@ -41,10 +41,6 @@ class MemberListApi(Resource):
     @account_initialization_required
     @marshal_with(account_with_role_list_fields)
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant:
-            raise ValueError("No current tenant")
         members = TenantService.get_tenant_members(current_user.current_tenant)
         return {"result": "success", "accounts": members}, 200
 
@@ -69,11 +65,7 @@ class MemberInviteEmailApi(Resource):
         if not TenantAccountRole.is_non_owner_role(invitee_role):
             return {"code": "invalid-role", "message": "Invalid role"}, 400
 
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         inviter = current_user
-        if not inviter.current_tenant:
-            raise ValueError("No current tenant")
         invitation_results = []
         console_web_url = dify_config.CONSOLE_WEB_URL
 
@@ -84,8 +76,6 @@ class MemberInviteEmailApi(Resource):
 
         for invitee_email in invitee_emails:
             try:
-                if not inviter.current_tenant:
-                    raise ValueError("No current tenant")
                 token = RegisterService.invite_new_member(
                     inviter.current_tenant, invitee_email, interface_language, role=invitee_role, inviter=inviter
                 )
@@ -107,7 +97,7 @@ class MemberInviteEmailApi(Resource):
         return {
             "result": "success",
             "invitation_results": invitation_results,
-            "tenant_id": str(inviter.current_tenant.id) if inviter.current_tenant else "",
+            "tenant_id": str(current_user.current_tenant.id),
         }, 201
 
 
@@ -118,10 +108,6 @@ class MemberCancelInviteApi(Resource):
     @login_required
     @account_initialization_required
     def delete(self, member_id):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant:
-            raise ValueError("No current tenant")
         member = db.session.query(Account).where(Account.id == str(member_id)).first()
         if member is None:
             abort(404)
@@ -137,10 +123,7 @@ class MemberCancelInviteApi(Resource):
             except Exception as e:
                 raise ValueError(str(e))
 
-        return {
-            "result": "success",
-            "tenant_id": str(current_user.current_tenant.id) if current_user.current_tenant else "",
-        }, 200
+        return {"result": "success", "tenant_id": str(current_user.current_tenant.id)}, 200
 
 
 class MemberUpdateRoleApi(Resource):
@@ -158,10 +141,6 @@ class MemberUpdateRoleApi(Resource):
         if not TenantAccountRole.is_valid_role(new_role):
             return {"code": "invalid-role", "message": "Invalid role"}, 400
 
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant:
-            raise ValueError("No current tenant")
         member = db.session.get(Account, str(member_id))
         if not member:
             abort(404)
@@ -185,10 +164,6 @@ class DatasetOperatorMemberListApi(Resource):
     @account_initialization_required
     @marshal_with(account_with_role_list_fields)
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant:
-            raise ValueError("No current tenant")
         members = TenantService.get_dataset_operator_members(current_user.current_tenant)
         return {"result": "success", "accounts": members}, 200
 
@@ -209,10 +184,6 @@ class SendOwnerTransferEmailApi(Resource):
             raise EmailSendIpLimitError()
 
         # check if the current user is the owner of the workspace
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant:
-            raise ValueError("No current tenant")
         if not TenantService.is_owner(current_user, current_user.current_tenant):
             raise NotOwnerError()
 
@@ -227,7 +198,7 @@ class SendOwnerTransferEmailApi(Resource):
             account=current_user,
             email=email,
             language=language,
-            workspace_name=current_user.current_tenant.name if current_user.current_tenant else "",
+            workspace_name=current_user.current_tenant.name,
         )
 
         return {"result": "success", "data": token}
@@ -244,10 +215,6 @@ class OwnerTransferCheckApi(Resource):
         parser.add_argument("token", type=str, required=True, nullable=False, location="json")
         args = parser.parse_args()
         # check if the current user is the owner of the workspace
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant:
-            raise ValueError("No current tenant")
         if not TenantService.is_owner(current_user, current_user.current_tenant):
             raise NotOwnerError()
 
@@ -289,10 +256,6 @@ class OwnerTransfer(Resource):
         args = parser.parse_args()
 
         # check if the current user is the owner of the workspace
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant:
-            raise ValueError("No current tenant")
         if not TenantService.is_owner(current_user, current_user.current_tenant):
             raise NotOwnerError()
 
@@ -311,11 +274,9 @@ class OwnerTransfer(Resource):
         member = db.session.get(Account, str(member_id))
         if not member:
             abort(404)
-            return  # Never reached, but helps type checker
-
-        if not current_user.current_tenant:
-            raise ValueError("No current tenant")
-        if not TenantService.is_member(member, current_user.current_tenant):
+        else:
+            member_account = member
+        if not TenantService.is_member(member_account, current_user.current_tenant):
             raise MemberNotInTenantError()
 
         try:
@@ -325,13 +286,13 @@ class OwnerTransfer(Resource):
             AccountService.send_new_owner_transfer_notify_email(
                 account=member,
                 email=member.email,
-                workspace_name=current_user.current_tenant.name if current_user.current_tenant else "",
+                workspace_name=current_user.current_tenant.name,
             )
 
             AccountService.send_old_owner_transfer_notify_email(
                 account=current_user,
                 email=current_user.email,
-                workspace_name=current_user.current_tenant.name if current_user.current_tenant else "",
+                workspace_name=current_user.current_tenant.name,
                 new_owner_email=member.email,
             )
 

api/controllers/console/workspace/model_providers.py

@@ -12,7 +12,6 @@ from core.model_runtime.errors.validate import CredentialsValidateFailedError
 from core.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import StrLen, uuid_value
 from libs.login import login_required
-from models.account import Account
 from services.billing_service import BillingService
 from services.model_provider_service import ModelProviderService
 
@@ -22,10 +21,6 @@ class ModelProviderListApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         tenant_id = current_user.current_tenant_id
 
         parser = reqparse.RequestParser()
@@ -50,10 +45,6 @@ class ModelProviderCredentialApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, provider: str):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         tenant_id = current_user.current_tenant_id
         # if credential_id is not provided, return current used credential
         parser = reqparse.RequestParser()
@@ -71,8 +62,6 @@ class ModelProviderCredentialApi(Resource):
     @login_required
     @account_initialization_required
     def post(self, provider: str):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         if not current_user.is_admin_or_owner:
             raise Forbidden()
 
@@ -83,8 +72,6 @@ class ModelProviderCredentialApi(Resource):
 
         model_provider_service = ModelProviderService()
 
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         try:
             model_provider_service.create_provider_credential(
                 tenant_id=current_user.current_tenant_id,
@@ -101,8 +88,6 @@ class ModelProviderCredentialApi(Resource):
     @login_required
     @account_initialization_required
     def put(self, provider: str):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         if not current_user.is_admin_or_owner:
             raise Forbidden()
 
@@ -114,8 +99,6 @@ class ModelProviderCredentialApi(Resource):
 
         model_provider_service = ModelProviderService()
 
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         try:
             model_provider_service.update_provider_credential(
                 tenant_id=current_user.current_tenant_id,
@@ -133,16 +116,12 @@ class ModelProviderCredentialApi(Resource):
     @login_required
     @account_initialization_required
     def delete(self, provider: str):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         if not current_user.is_admin_or_owner:
             raise Forbidden()
         parser = reqparse.RequestParser()
         parser.add_argument("credential_id", type=uuid_value, required=True, nullable=False, location="json")
         args = parser.parse_args()
 
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         model_provider_service = ModelProviderService()
         model_provider_service.remove_provider_credential(
             tenant_id=current_user.current_tenant_id, provider=provider, credential_id=args["credential_id"]
@@ -156,16 +135,12 @@ class ModelProviderCredentialSwitchApi(Resource):
     @login_required
     @account_initialization_required
     def post(self, provider: str):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         if not current_user.is_admin_or_owner:
             raise Forbidden()
         parser = reqparse.RequestParser()
         parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
         args = parser.parse_args()
 
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         service = ModelProviderService()
         service.switch_active_provider_credential(
             tenant_id=current_user.current_tenant_id,
@@ -180,14 +155,10 @@ class ModelProviderValidateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self, provider: str):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
         args = parser.parse_args()
 
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         tenant_id = current_user.current_tenant_id
 
         model_provider_service = ModelProviderService()
@@ -234,13 +205,9 @@ class PreferredProviderTypeUpdateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self, provider: str):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         if not current_user.is_admin_or_owner:
             raise Forbidden()
 
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         tenant_id = current_user.current_tenant_id
 
         parser = reqparse.RequestParser()
@@ -269,11 +236,7 @@ class ModelProviderPaymentCheckoutUrlApi(Resource):
     def get(self, provider: str):
         if provider != "anthropic":
             raise ValueError(f"provider name {provider} is invalid")
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         BillingService.is_tenant_owner_or_admin(current_user)
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         data = BillingService.get_model_provider_payment_link(
             provider_name=provider,
             tenant_id=current_user.current_tenant_id,

api/controllers/console/workspace/tool_providers.py

@@ -21,11 +21,11 @@ from core.mcp.auth.auth_provider import OAuthClientProvider
 from core.mcp.error import MCPAuthError, MCPError
 from core.mcp.mcp_client import MCPClient
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.entities.plugin import ToolProviderID
 from core.plugin.impl.oauth import OAuthHandler
 from core.tools.entities.tool_entities import CredentialType
 from libs.helper import StrLen, alphanumeric, uuid_value
 from libs.login import login_required
+from models.provider_ids import ToolProviderID
 from services.plugin.oauth_service import OAuthProxyService
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService
@@ -865,7 +865,6 @@ class ToolProviderMCPApi(Resource):
         parser.add_argument(
             "sse_read_timeout", type=float, required=False, nullable=False, location="json", default=300
         )
-        parser.add_argument("headers", type=dict, required=False, nullable=True, location="json", default={})
         args = parser.parse_args()
         user = current_user
         if not is_valid_url(args["server_url"]):
@@ -882,7 +881,6 @@ class ToolProviderMCPApi(Resource):
                 server_identifier=args["server_identifier"],
                 timeout=args["timeout"],
                 sse_read_timeout=args["sse_read_timeout"],
-                headers=args["headers"],
             )
         )
 
@@ -900,7 +898,6 @@ class ToolProviderMCPApi(Resource):
         parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
         parser.add_argument("timeout", type=float, required=False, nullable=True, location="json")
         parser.add_argument("sse_read_timeout", type=float, required=False, nullable=True, location="json")
-        parser.add_argument("headers", type=dict, required=False, nullable=True, location="json")
         args = parser.parse_args()
         if not is_valid_url(args["server_url"]):
             if "[__HIDDEN__]" in args["server_url"]:
@@ -918,7 +915,6 @@ class ToolProviderMCPApi(Resource):
             server_identifier=args["server_identifier"],
             timeout=args.get("timeout"),
             sse_read_timeout=args.get("sse_read_timeout"),
-            headers=args.get("headers"),
         )
         return {"result": "success"}
 
@@ -955,9 +951,6 @@ class ToolMCPAuthApi(Resource):
                 authed=False,
                 authorization_code=args["authorization_code"],
                 for_list=True,
-                headers=provider.decrypted_headers,
-                timeout=provider.timeout,
-                sse_read_timeout=provider.sse_read_timeout,
             ):
                 MCPToolManageService.update_mcp_provider_credentials(
                     mcp_provider=provider,

api/controllers/console/workspace/workspace.py

@@ -25,7 +25,7 @@ from controllers.console.wraps import (
 from extensions.ext_database import db
 from libs.helper import TimestampField
 from libs.login import login_required
-from models.account import Account, Tenant, TenantStatus
+from models.account import Tenant, TenantStatus
 from services.account_service import TenantService
 from services.feature_service import FeatureService
 from services.file_service import FileService
@@ -70,8 +70,6 @@ class TenantListApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         tenants = TenantService.get_join_tenants(current_user)
         tenant_dicts = []
 
@@ -85,7 +83,7 @@ class TenantListApi(Resource):
                 "status": tenant.status,
                 "created_at": tenant.created_at,
                 "plan": features.billing.subscription.plan if features.billing.enabled else "sandbox",
-                "current": tenant.id == current_user.current_tenant_id if current_user.current_tenant_id else False,
+                "current": tenant.id == current_user.current_tenant_id,
             }
 
             tenant_dicts.append(tenant_dict)
@@ -127,11 +125,7 @@ class TenantApi(Resource):
         if request.path == "/info":
             logger.warning("Deprecated URL /info was used.")
 
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         tenant = current_user.current_tenant
-        if not tenant:
-            raise ValueError("No current tenant")
 
         if tenant.status == TenantStatus.ARCHIVE:
             tenants = TenantService.get_join_tenants(current_user)
@@ -143,8 +137,6 @@ class TenantApi(Resource):
             else:
                 raise Unauthorized("workspace is archived")
 
-        if not tenant:
-            raise ValueError("No tenant available")
         return WorkspaceService.get_tenant_info(tenant), 200
 
 
@@ -153,8 +145,6 @@ class SwitchWorkspaceApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("tenant_id", type=str, required=True, location="json")
         args = parser.parse_args()
@@ -178,15 +168,11 @@ class CustomConfigWorkspaceApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("workspace_custom")
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("remove_webapp_brand", type=bool, location="json")
         parser.add_argument("replace_webapp_logo", type=str, location="json")
         args = parser.parse_args()
 
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         tenant = db.get_or_404(Tenant, current_user.current_tenant_id)
 
         custom_config_dict = {
@@ -208,8 +194,6 @@ class WebappLogoWorkspaceApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("workspace_custom")
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
@@ -227,7 +211,7 @@ class WebappLogoWorkspaceApi(Resource):
             raise UnsupportedFileTypeError()
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,
@@ -248,14 +232,10 @@ class WorkspaceInfoApi(Resource):
     @account_initialization_required
     # Change workspace name
     def post(self):
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, location="json")
         args = parser.parse_args()
 
-        if not current_user.current_tenant_id:
-            raise ValueError("No current tenant")
         tenant = db.get_or_404(Tenant, current_user.current_tenant_id)
         tenant.name = args["name"]
         db.session.commit()

api/controllers/console/wraps.py

@@ -266,3 +266,14 @@ def is_allow_transfer_owner(view: Callable[P, R]):
         abort(403)
 
     return decorated
+
+
+def knowledge_pipeline_publish_enabled(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_features(current_user.current_tenant_id)
+        if features.knowledge_pipeline.publish_enabled:
+            return view(*args, **kwargs)
+        abort(403)
+
+    return decorated

api/controllers/files/__init__.py

@@ -10,10 +10,11 @@ api = ExternalApi(
     version="1.0",
     title="Files API",
     description="API for file operations including upload and preview",
+    doc="/docs",  # Enable Swagger UI at /files/docs
 )
 
 files_ns = Namespace("files", description="File operations", path="/")
 
-from . import image_preview, tool_files, upload  # pyright: ignore[reportUnusedImport]
+from . import image_preview, tool_files, upload
 
 api.add_namespace(files_ns)

api/controllers/files/image_preview.py

@@ -7,6 +7,7 @@ from werkzeug.exceptions import NotFound
 import services
 from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import files_ns
+from extensions.ext_database import db
 from services.account_service import TenantService
 from services.file_service import FileService
 
@@ -28,7 +29,7 @@ class ImagePreviewApi(Resource):
             return {"content": "Invalid request."}, 400
 
         try:
-            generator, mimetype = FileService.get_image_preview(
+            generator, mimetype = FileService(db.engine).get_image_preview(
                 file_id=file_id,
                 timestamp=timestamp,
                 nonce=nonce,
@@ -57,7 +58,7 @@ class FilePreviewApi(Resource):
             return {"content": "Invalid request."}, 400
 
         try:
-            generator, upload_file = FileService.get_file_generator_by_file_id(
+            generator, upload_file = FileService(db.engine).get_file_generator_by_file_id(
                 file_id=file_id,
                 timestamp=args["timestamp"],
                 nonce=args["nonce"],
@@ -108,7 +109,7 @@ class WorkspaceWebappLogoApi(Resource):
             raise NotFound("webapp logo is not found")
 
         try:
-            generator, mimetype = FileService.get_public_image_preview(
+            generator, mimetype = FileService(db.engine).get_public_image_preview(
                 webapp_logo_file_id,
             )
         except services.errors.file.UnsupportedFileTypeError:

api/controllers/files/tool_files.py

@@ -8,7 +8,7 @@ from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import files_ns
 from core.tools.signature import verify_tool_file_signature
 from core.tools.tool_file_manager import ToolFileManager
-from models import db as global_db
+from extensions.ext_database import db as global_db
 
 
 @files_ns.route("/tools/<uuid:file_id>.<string:extension>")

api/controllers/inner_api/__init__.py

@@ -10,13 +10,14 @@ api = ExternalApi(
     version="1.0",
     title="Inner API",
     description="Internal APIs for enterprise features, billing, and plugin communication",
+    doc="/docs",  # Enable Swagger UI at /inner/api/docs
 )
 
 # Create namespace
 inner_api_ns = Namespace("inner_api", description="Internal API operations", path="/")
 
-from . import mail as _mail  # pyright: ignore[reportUnusedImport]
-from .plugin import plugin as _plugin  # pyright: ignore[reportUnusedImport]
-from .workspace import workspace as _workspace  # pyright: ignore[reportUnusedImport]
+from . import mail
+from .plugin import plugin
+from .workspace import workspace
 
 api.add_namespace(inner_api_ns)

api/controllers/inner_api/plugin/plugin.py

@@ -37,9 +37,9 @@ from models.model import EndUser
 
 @inner_api_ns.route("/invoke/llm")
 class PluginInvokeLLMApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeLLM)
     @inner_api_ns.doc("plugin_invoke_llm")
     @inner_api_ns.doc(description="Invoke LLM models through plugin interface")
@@ -60,9 +60,9 @@ class PluginInvokeLLMApi(Resource):
 
 @inner_api_ns.route("/invoke/llm/structured-output")
 class PluginInvokeLLMWithStructuredOutputApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeLLMWithStructuredOutput)
     @inner_api_ns.doc("plugin_invoke_llm_structured")
     @inner_api_ns.doc(description="Invoke LLM models with structured output through plugin interface")
@@ -85,9 +85,9 @@ class PluginInvokeLLMWithStructuredOutputApi(Resource):
 
 @inner_api_ns.route("/invoke/text-embedding")
 class PluginInvokeTextEmbeddingApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeTextEmbedding)
     @inner_api_ns.doc("plugin_invoke_text_embedding")
     @inner_api_ns.doc(description="Invoke text embedding models through plugin interface")
@@ -115,9 +115,9 @@ class PluginInvokeTextEmbeddingApi(Resource):
 
 @inner_api_ns.route("/invoke/rerank")
 class PluginInvokeRerankApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeRerank)
     @inner_api_ns.doc("plugin_invoke_rerank")
     @inner_api_ns.doc(description="Invoke rerank models through plugin interface")
@@ -141,9 +141,9 @@ class PluginInvokeRerankApi(Resource):
 
 @inner_api_ns.route("/invoke/tts")
 class PluginInvokeTTSApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeTTS)
     @inner_api_ns.doc("plugin_invoke_tts")
     @inner_api_ns.doc(description="Invoke text-to-speech models through plugin interface")
@@ -168,9 +168,9 @@ class PluginInvokeTTSApi(Resource):
 
 @inner_api_ns.route("/invoke/speech2text")
 class PluginInvokeSpeech2TextApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeSpeech2Text)
     @inner_api_ns.doc("plugin_invoke_speech2text")
     @inner_api_ns.doc(description="Invoke speech-to-text models through plugin interface")
@@ -194,9 +194,9 @@ class PluginInvokeSpeech2TextApi(Resource):
 
 @inner_api_ns.route("/invoke/moderation")
 class PluginInvokeModerationApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeModeration)
     @inner_api_ns.doc("plugin_invoke_moderation")
     @inner_api_ns.doc(description="Invoke moderation models through plugin interface")
@@ -220,9 +220,9 @@ class PluginInvokeModerationApi(Resource):
 
 @inner_api_ns.route("/invoke/tool")
 class PluginInvokeToolApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeTool)
     @inner_api_ns.doc("plugin_invoke_tool")
     @inner_api_ns.doc(description="Invoke tools through plugin interface")
@@ -252,9 +252,9 @@ class PluginInvokeToolApi(Resource):
 
 @inner_api_ns.route("/invoke/parameter-extractor")
 class PluginInvokeParameterExtractorNodeApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeParameterExtractorNode)
     @inner_api_ns.doc("plugin_invoke_parameter_extractor")
     @inner_api_ns.doc(description="Invoke parameter extractor node through plugin interface")
@@ -285,9 +285,9 @@ class PluginInvokeParameterExtractorNodeApi(Resource):
 
 @inner_api_ns.route("/invoke/question-classifier")
 class PluginInvokeQuestionClassifierNodeApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeQuestionClassifierNode)
     @inner_api_ns.doc("plugin_invoke_question_classifier")
     @inner_api_ns.doc(description="Invoke question classifier node through plugin interface")
@@ -318,9 +318,9 @@ class PluginInvokeQuestionClassifierNodeApi(Resource):
 
 @inner_api_ns.route("/invoke/app")
 class PluginInvokeAppApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeApp)
     @inner_api_ns.doc("plugin_invoke_app")
     @inner_api_ns.doc(description="Invoke application through plugin interface")
@@ -348,9 +348,9 @@ class PluginInvokeAppApi(Resource):
 
 @inner_api_ns.route("/invoke/encrypt")
 class PluginInvokeEncryptApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeEncrypt)
     @inner_api_ns.doc("plugin_invoke_encrypt")
     @inner_api_ns.doc(description="Encrypt or decrypt data through plugin interface")
@@ -375,9 +375,9 @@ class PluginInvokeEncryptApi(Resource):
 
 @inner_api_ns.route("/invoke/summary")
 class PluginInvokeSummaryApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestInvokeSummary)
     @inner_api_ns.doc("plugin_invoke_summary")
     @inner_api_ns.doc(description="Invoke summary functionality through plugin interface")
@@ -405,9 +405,9 @@ class PluginInvokeSummaryApi(Resource):
 
 @inner_api_ns.route("/upload/file/request")
 class PluginUploadFileRequestApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestRequestUploadFile)
     @inner_api_ns.doc("plugin_upload_file_request")
     @inner_api_ns.doc(description="Request signed URL for file upload through plugin interface")
@@ -426,9 +426,9 @@ class PluginUploadFileRequestApi(Resource):
 
 @inner_api_ns.route("/fetch/app/info")
 class PluginFetchAppInfoApi(Resource):
-    @get_user_tenant
     @setup_required
     @plugin_inner_api_only
+    @get_user_tenant
     @plugin_data(payload_type=RequestFetchAppInfo)
     @inner_api_ns.doc("plugin_fetch_app_info")
     @inner_api_ns.doc(description="Fetch application information through plugin interface")

api/controllers/inner_api/plugin/wraps.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import Optional, ParamSpec, TypeVar, cast
+from typing import Optional
 
 from flask import current_app, request
 from flask_login import user_logged_in
@@ -8,72 +8,65 @@ from flask_restx import reqparse
 from pydantic import BaseModel
 from sqlalchemy.orm import Session
 
-from core.file.constants import DEFAULT_SERVICE_API_USER_ID
 from extensions.ext_database import db
-from libs.login import current_user
-from models.account import Tenant
+from libs.login import _get_user
+from models.account import Account, Tenant
 from models.model import EndUser
-
-P = ParamSpec("P")
-R = TypeVar("R")
+from services.account_service import AccountService
 
 
-def get_user(tenant_id: str, user_id: str | None) -> EndUser:
-    """
-    Get current user
-
-    NOTE: user_id is not trusted, it could be maliciously set to any value.
-    As a result, it could only be considered as an end user id.
-    """
+def get_user(tenant_id: str, user_id: str | None) -> Account | EndUser:
     try:
         with Session(db.engine) as session:
             if not user_id:
-                user_id = DEFAULT_SERVICE_API_USER_ID
-
-            user_model = (
-                session.query(EndUser)
-                .where(
-                    EndUser.session_id == user_id,
-                    EndUser.tenant_id == tenant_id,
-                )
-                .first()
-            )
-            if not user_model:
-                user_model = EndUser(
-                    tenant_id=tenant_id,
-                    type="service_api",
-                    is_anonymous=user_id == DEFAULT_SERVICE_API_USER_ID,
-                    session_id=user_id,
-                )
-                session.add(user_model)
-                session.commit()
-                session.refresh(user_model)
+                user_id = "DEFAULT-USER"
 
+            if user_id == "DEFAULT-USER":
+                user_model = session.query(EndUser).where(EndUser.session_id == "DEFAULT-USER").first()
+                if not user_model:
+                    user_model = EndUser(
+                        tenant_id=tenant_id,
+                        type="service_api",
+                        is_anonymous=True if user_id == "DEFAULT-USER" else False,
+                        session_id=user_id,
+                    )
+                    session.add(user_model)
+                    session.commit()
+                    session.refresh(user_model)
+            else:
+                user_model = AccountService.load_user(user_id)
+                if not user_model:
+                    user_model = session.query(EndUser).where(EndUser.id == user_id).first()
+                if not user_model:
+                    raise ValueError("user not found")
     except Exception:
         raise ValueError("user not found")
 
     return user_model
 
 
-def get_user_tenant(view: Optional[Callable[P, R]] = None):
-    def decorator(view_func: Callable[P, R]):
+def get_user_tenant(view: Optional[Callable] = None):
+    def decorator(view_func):
         @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        def decorated_view(*args, **kwargs):
             # fetch json body
             parser = reqparse.RequestParser()
             parser.add_argument("tenant_id", type=str, required=True, location="json")
             parser.add_argument("user_id", type=str, required=True, location="json")
 
-            p = parser.parse_args()
+            kwargs = parser.parse_args()
 
-            user_id = cast(str, p.get("user_id"))
-            tenant_id = cast(str, p.get("tenant_id"))
+            user_id = kwargs.get("user_id")
+            tenant_id = kwargs.get("tenant_id")
 
             if not tenant_id:
                 raise ValueError("tenant_id is required")
 
             if not user_id:
-                user_id = DEFAULT_SERVICE_API_USER_ID
+                user_id = "DEFAULT-USER"
+
+            del kwargs["tenant_id"]
+            del kwargs["user_id"]
 
             try:
                 tenant_model = (
@@ -95,7 +88,7 @@ def get_user_tenant(view: Optional[Callable[P, R]] = None):
             kwargs["user_model"] = user
 
             current_app.login_manager._update_request_context_with_user(user)  # type: ignore
-            user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
+            user_logged_in.send(current_app._get_current_object(), user=_get_user())  # type: ignore
 
             return view_func(*args, **kwargs)
 
@@ -107,9 +100,9 @@ def get_user_tenant(view: Optional[Callable[P, R]] = None):
         return decorator(view)
 
 
-def plugin_data(view: Optional[Callable[P, R]] = None, *, payload_type: type[BaseModel]):
-    def decorator(view_func: Callable[P, R]):
-        def decorated_view(*args: P.args, **kwargs: P.kwargs):
+def plugin_data(view: Optional[Callable] = None, *, payload_type: type[BaseModel]):
+    def decorator(view_func):
+        def decorated_view(*args, **kwargs):
             try:
                 data = request.get_json()
             except Exception:

api/controllers/inner_api/wraps.py

@@ -46,9 +46,9 @@ def enterprise_inner_api_only(view: Callable[P, R]):
     return decorated
 
 
-def enterprise_inner_api_user_auth(view: Callable[P, R]):
+def enterprise_inner_api_user_auth(view):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs):
+    def decorated(*args, **kwargs):
         if not dify_config.INNER_API:
             return view(*args, **kwargs)
 

api/controllers/mcp/__init__.py

@@ -10,10 +10,11 @@ api = ExternalApi(
     version="1.0",
     title="MCP API",
     description="API for Model Context Protocol operations",
+    doc="/docs",  # Enable Swagger UI at /mcp/docs
 )
 
 mcp_ns = Namespace("mcp", description="MCP operations", path="/")
 
-from . import mcp  # pyright: ignore[reportUnusedImport]
+from . import mcp
 
 api.add_namespace(mcp_ns)

api/controllers/service_api/__init__.py

@@ -10,31 +10,14 @@ api = ExternalApi(
     version="1.0",
     title="Service API",
     description="API for application services",
+    doc="/docs",  # Enable Swagger UI at /v1/docs
 )
 
 service_api_ns = Namespace("service_api", description="Service operations", path="/")
 
-from . import index  # pyright: ignore[reportUnusedImport]
-from .app import (
-    annotation,  # pyright: ignore[reportUnusedImport]
-    app,  # pyright: ignore[reportUnusedImport]
-    audio,  # pyright: ignore[reportUnusedImport]
-    completion,  # pyright: ignore[reportUnusedImport]
-    conversation,  # pyright: ignore[reportUnusedImport]
-    file,  # pyright: ignore[reportUnusedImport]
-    file_preview,  # pyright: ignore[reportUnusedImport]
-    message,  # pyright: ignore[reportUnusedImport]
-    site,  # pyright: ignore[reportUnusedImport]
-    workflow,  # pyright: ignore[reportUnusedImport]
-)
-from .dataset import (
-    dataset,  # pyright: ignore[reportUnusedImport]
-    document,  # pyright: ignore[reportUnusedImport]
-    hit_testing,  # pyright: ignore[reportUnusedImport]
-    metadata,  # pyright: ignore[reportUnusedImport]
-    segment,  # pyright: ignore[reportUnusedImport]
-    upload_file,  # pyright: ignore[reportUnusedImport]
-)
-from .workspace import models  # pyright: ignore[reportUnusedImport]
+from . import index
+from .app import annotation, app, audio, completion, conversation, file, file_preview, message, site, workflow
+from .dataset import dataset, document, hit_testing, metadata, segment, upload_file
+from .workspace import models
 
 api.add_namespace(service_api_ns)

api/controllers/service_api/app/annotation.py

@@ -165,7 +165,7 @@ class AnnotationUpdateDeleteApi(Resource):
     def put(self, app_model: App, annotation_id):
         """Update an existing annotation."""
         assert isinstance(current_user, Account)
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         annotation_id = str(annotation_id)
@@ -189,7 +189,7 @@ class AnnotationUpdateDeleteApi(Resource):
         """Delete an annotation."""
         assert isinstance(current_user, Account)
 
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         annotation_id = str(annotation_id)

api/controllers/service_api/app/conversation.py

@@ -1,5 +1,4 @@
 from flask_restx import Resource, reqparse
-from flask_restx._http import HTTPStatus
 from flask_restx.inputs import int_range
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, NotFound
@@ -122,7 +121,7 @@ class ConversationDetailApi(Resource):
         }
     )
     @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON))
-    @service_api_ns.marshal_with(build_conversation_delete_model(service_api_ns), code=HTTPStatus.NO_CONTENT)
+    @service_api_ns.marshal_with(build_conversation_delete_model(service_api_ns), code=204)
     def delete(self, app_model: App, end_user: EndUser, c_id):
         """Delete a specific conversation."""
         app_mode = AppMode.value_of(app_model.mode)

api/controllers/service_api/app/file.py

@@ -12,8 +12,9 @@ from controllers.common.errors import (
 )
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
+from extensions.ext_database import db
 from fields.file_fields import build_file_model
-from models.model import App, EndUser
+from models import App, EndUser
 from services.file_service import FileService
 
 
@@ -52,7 +53,7 @@ class FileApi(Resource):
             raise FilenameNotExistsError
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,

api/controllers/service_api/app/workflow.py

@@ -26,7 +26,8 @@ from core.errors.error import (
 )
 from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
+from core.workflow.enums import WorkflowExecutionStatus
+from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import build_workflow_app_log_pagination_model
 from libs import helper
@@ -262,7 +263,12 @@ class WorkflowTaskStopApi(Resource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.SERVICE_API, end_user.id)
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}
 

api/controllers/service_api/dataset/dataset.py

@@ -13,13 +13,13 @@ from controllers.service_api.wraps import (
     validate_dataset_token,
 )
 from core.model_runtime.entities.model_entities import ModelType
-from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
 from fields.dataset_fields import dataset_detail_fields
 from fields.tag_fields import build_dataset_tag_fields
 from libs.login import current_user
 from models.account import Account
 from models.dataset import Dataset, DatasetPermissionEnum
+from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import RetrievalModel
 from services.tag_service import TagService
@@ -559,7 +559,7 @@ class DatasetTagsApi(DatasetApiResource):
     def post(self, _, dataset_id):
         """Add a knowledge type tag."""
         assert isinstance(current_user, Account)
-        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
             raise Forbidden()
 
         args = tag_create_parser.parse_args()
@@ -583,7 +583,7 @@ class DatasetTagsApi(DatasetApiResource):
     @validate_dataset_token
     def patch(self, _, dataset_id):
         assert isinstance(current_user, Account)
-        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
             raise Forbidden()
 
         args = tag_update_parser.parse_args()
@@ -610,7 +610,7 @@ class DatasetTagsApi(DatasetApiResource):
     def delete(self, _, dataset_id):
         """Delete a knowledge type tag."""
         assert isinstance(current_user, Account)
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
         args = tag_delete_parser.parse_args()
         TagService.delete_tag(args.get("tag_id"))
@@ -634,7 +634,7 @@ class DatasetTagBindingApi(DatasetApiResource):
     def post(self, _, dataset_id):
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
         assert isinstance(current_user, Account)
-        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
             raise Forbidden()
 
         args = tag_binding_parser.parse_args()
@@ -660,7 +660,7 @@ class DatasetTagUnbindingApi(DatasetApiResource):
     def post(self, _, dataset_id):
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
         assert isinstance(current_user, Account)
-        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
             raise Forbidden()
 
         args = tag_unbinding_parser.parse_args()

api/controllers/service_api/dataset/document.py

@@ -30,7 +30,6 @@ from extensions.ext_database import db
 from fields.document_fields import document_fields, document_status_fields
 from libs.login import current_user
 from models.dataset import Dataset, Document, DocumentSegment
-from models.model import EndUser
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig
 from services.file_service import FileService
@@ -124,7 +123,7 @@ class DocumentAddByTextApi(DatasetApiResource):
                 args.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
             )
 
-        upload_file = FileService.upload_text(text=str(text), text_name=str(name))
+        upload_file = FileService(db.engine).upload_text(text=str(text), text_name=str(name))
         data_source = {
             "type": "upload_file",
             "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
@@ -134,6 +133,9 @@ class DocumentAddByTextApi(DatasetApiResource):
         # validate args
         DocumentService.document_create_args_validate(knowledge_config)
 
+        if not current_user:
+            raise ValueError("current_user is required")
+
         try:
             documents, batch = DocumentService.save_document_with_dataset_id(
                 dataset=dataset,
@@ -199,7 +201,7 @@ class DocumentUpdateByTextApi(DatasetApiResource):
             name = args.get("name")
             if text is None or name is None:
                 raise ValueError("Both text and name must be strings.")
-            upload_file = FileService.upload_text(text=str(text), text_name=str(name))
+            upload_file = FileService(db.engine).upload_text(text=str(text), text_name=str(name))
             data_source = {
                 "type": "upload_file",
                 "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
@@ -299,10 +301,7 @@ class DocumentAddByFileApi(DatasetApiResource):
         if not file.filename:
             raise FilenameNotExistsError
 
-        if not isinstance(current_user, EndUser):
-            raise ValueError("Invalid user account")
-
-        upload_file = FileService.upload_file(
+        upload_file = FileService(db.engine).upload_file(
             filename=file.filename,
             content=file.read(),
             mimetype=file.mimetype,
@@ -391,9 +390,7 @@ class DocumentUpdateByFileApi(DatasetApiResource):
                 raise FilenameNotExistsError
 
             try:
-                if not isinstance(current_user, EndUser):
-                    raise ValueError("Invalid user account")
-                upload_file = FileService.upload_file(
+                upload_file = FileService(db.engine).upload_file(
                     filename=file.filename,
                     content=file.read(),
                     mimetype=file.mimetype,

api/controllers/service_api/dataset/metadata.py

@@ -174,7 +174,7 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
             MetadataService.enable_built_in_field(dataset)
         elif action == "disable":
             MetadataService.disable_built_in_field(dataset)
-        return {"result": "success"}, 200
+        return 200
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/documents/metadata")
@@ -204,4 +204,4 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
 
         MetadataService.update_documents_metadata(dataset, metadata_args)
 
-        return {"result": "success"}, 200
+        return 200

api/controllers/service_api/workspace/models.py

@@ -19,7 +19,7 @@ class ModelProviderAvailableModelApi(Resource):
         }
     )
     @validate_dataset_token
-    def get(self, _, model_type: str):
+    def get(self, _, model_type):
         """Get available models by model type.
 
         Returns a list of available models for the specified model type.

api/controllers/service_api/wraps.py

@@ -3,7 +3,7 @@ from collections.abc import Callable
 from datetime import timedelta
 from enum import StrEnum, auto
 from functools import wraps
-from typing import Concatenate, Optional, ParamSpec, TypeVar
+from typing import Optional, ParamSpec, TypeVar
 
 from flask import current_app, request
 from flask_login import user_logged_in
@@ -13,11 +13,10 @@ from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
-from core.file.constants import DEFAULT_SERVICE_API_USER_ID
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.datetime_utils import naive_utc_now
-from libs.login import current_user
+from libs.login import _get_user
 from models.account import Account, Tenant, TenantAccountJoin, TenantStatus
 from models.dataset import Dataset, RateLimitLog
 from models.model import ApiToken, App, EndUser
@@ -25,7 +24,6 @@ from services.feature_service import FeatureService
 
 P = ParamSpec("P")
 R = TypeVar("R")
-T = TypeVar("T")
 
 
 class WhereisUserArg(StrEnum):
@@ -43,10 +41,10 @@ class FetchUserArg(BaseModel):
     required: bool = False
 
 
-def validate_app_token(view: Optional[Callable[P, R]] = None, *, fetch_user_arg: Optional[FetchUserArg] = None):
-    def decorator(view_func: Callable[P, R]):
+def validate_app_token(view: Optional[Callable] = None, *, fetch_user_arg: Optional[FetchUserArg] = None):
+    def decorator(view_func):
         @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        def decorated_view(*args, **kwargs):
             api_token = validate_and_get_api_token("app")
 
             app_model = db.session.query(App).where(App.id == api_token.app_id).first()
@@ -190,10 +188,10 @@ def cloud_edition_billing_rate_limit_check(resource: str, api_token_type: str):
     return interceptor
 
 
-def validate_dataset_token(view: Optional[Callable[Concatenate[T, P], R]] = None):
-    def decorator(view: Callable[Concatenate[T, P], R]):
+def validate_dataset_token(view=None):
+    def decorator(view):
         @wraps(view)
-        def decorated(*args: P.args, **kwargs: P.kwargs):
+        def decorated(*args, **kwargs):
             api_token = validate_and_get_api_token("dataset")
             tenant_account_join = (
                 db.session.query(Tenant, TenantAccountJoin)
@@ -210,7 +208,7 @@ def validate_dataset_token(view: Optional[Callable[Concatenate[T, P], R]] = None
                 if account:
                     account.current_tenant = tenant
                     current_app.login_manager._update_request_context_with_user(account)  # type: ignore
-                    user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
+                    user_logged_in.send(current_app._get_current_object(), user=_get_user())  # type: ignore
                 else:
                     raise Unauthorized("Tenant owner account does not exist.")
             else:
@@ -273,7 +271,7 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: Optional[str]
     Create or update session terminal based on user ID.
     """
     if not user_id:
-        user_id = DEFAULT_SERVICE_API_USER_ID
+        user_id = "DEFAULT-USER"
 
     with Session(db.engine, expire_on_commit=False) as session:
         end_user = (
@@ -292,7 +290,7 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: Optional[str]
                 tenant_id=app_model.tenant_id,
                 app_id=app_model.id,
                 type="service_api",
-                is_anonymous=user_id == DEFAULT_SERVICE_API_USER_ID,
+                is_anonymous=user_id == "DEFAULT-USER",
                 session_id=user_id,
             )
             session.add(end_user)

api/controllers/web/__init__.py

@@ -10,26 +10,27 @@ api = ExternalApi(
     version="1.0",
     title="Web API",
     description="Public APIs for web applications including file uploads, chat interactions, and app management",
+    doc="/docs",  # Enable Swagger UI at /api/docs
 )
 
 # Create namespace
 web_ns = Namespace("web", description="Web application API operations", path="/")
 
 from . import (
-    app,  # pyright: ignore[reportUnusedImport]
-    audio,  # pyright: ignore[reportUnusedImport]
-    completion,  # pyright: ignore[reportUnusedImport]
-    conversation,  # pyright: ignore[reportUnusedImport]
-    feature,  # pyright: ignore[reportUnusedImport]
-    files,  # pyright: ignore[reportUnusedImport]
-    forgot_password,  # pyright: ignore[reportUnusedImport]
-    login,  # pyright: ignore[reportUnusedImport]
-    message,  # pyright: ignore[reportUnusedImport]
-    passport,  # pyright: ignore[reportUnusedImport]
-    remote_files,  # pyright: ignore[reportUnusedImport]
-    saved_message,  # pyright: ignore[reportUnusedImport]
-    site,  # pyright: ignore[reportUnusedImport]
-    workflow,  # pyright: ignore[reportUnusedImport]
+    app,
+    audio,
+    completion,
+    conversation,
+    feature,
+    files,
+    forgot_password,
+    login,
+    message,
+    passport,
+    remote_files,
+    saved_message,
+    site,
+    workflow,
 )
 
 api.add_namespace(web_ns)

api/controllers/web/audio.py

@@ -5,7 +5,7 @@ from flask_restx import fields, marshal_with, reqparse
 from werkzeug.exceptions import InternalServerError
 
 import services
-from controllers.web import web_ns
+from controllers.web import api
 from controllers.web.error import (
     AppUnavailableError,
     AudioTooLargeError,
@@ -32,16 +32,15 @@ from services.errors.audio import (
 logger = logging.getLogger(__name__)
 
 
-@web_ns.route("/audio-to-text")
 class AudioApi(WebApiResource):
     audio_to_text_response_fields = {
         "text": fields.String,
     }
 
     @marshal_with(audio_to_text_response_fields)
-    @web_ns.doc("Audio to Text")
-    @web_ns.doc(description="Convert audio file to text using speech-to-text service.")
-    @web_ns.doc(
+    @api.doc("Audio to Text")
+    @api.doc(description="Convert audio file to text using speech-to-text service.")
+    @api.doc(
         responses={
             200: "Success",
             400: "Bad Request",
@@ -86,7 +85,6 @@ class AudioApi(WebApiResource):
             raise InternalServerError()
 
 
-@web_ns.route("/text-to-audio")
 class TextApi(WebApiResource):
     text_to_audio_response_fields = {
         "audio_url": fields.String,
@@ -94,9 +92,9 @@ class TextApi(WebApiResource):
     }
 
     @marshal_with(text_to_audio_response_fields)
-    @web_ns.doc("Text to Audio")
-    @web_ns.doc(description="Convert text to audio using text-to-speech service.")
-    @web_ns.doc(
+    @api.doc("Text to Audio")
+    @api.doc(description="Convert text to audio using text-to-speech service.")
+    @api.doc(
         responses={
             200: "Success",
             400: "Bad Request",
@@ -147,3 +145,7 @@ class TextApi(WebApiResource):
         except Exception as e:
             logger.exception("Failed to handle post request to TextApi")
             raise InternalServerError()
+
+
+api.add_resource(AudioApi, "/audio-to-text")
+api.add_resource(TextApi, "/text-to-audio")

api/controllers/web/completion.py

@@ -4,7 +4,7 @@ from flask_restx import reqparse
 from werkzeug.exceptions import InternalServerError, NotFound
 
 import services
-from controllers.web import web_ns
+from controllers.web import api
 from controllers.web.error import (
     AppUnavailableError,
     CompletionRequestError,
@@ -35,11 +35,10 @@ logger = logging.getLogger(__name__)
 
 
 # define completion api for user
-@web_ns.route("/completion-messages")
 class CompletionApi(WebApiResource):
-    @web_ns.doc("Create Completion Message")
-    @web_ns.doc(description="Create a completion message for text generation applications.")
-    @web_ns.doc(
+    @api.doc("Create Completion Message")
+    @api.doc(description="Create a completion message for text generation applications.")
+    @api.doc(
         params={
             "inputs": {"description": "Input variables for the completion", "type": "object", "required": True},
             "query": {"description": "Query text for completion", "type": "string", "required": False},
@@ -53,7 +52,7 @@ class CompletionApi(WebApiResource):
             "retriever_from": {"description": "Source of retriever", "type": "string", "required": False},
         }
     )
-    @web_ns.doc(
+    @api.doc(
         responses={
             200: "Success",
             400: "Bad Request",
@@ -107,12 +106,11 @@ class CompletionApi(WebApiResource):
             raise InternalServerError()
 
 
-@web_ns.route("/completion-messages/<string:task_id>/stop")
 class CompletionStopApi(WebApiResource):
-    @web_ns.doc("Stop Completion Message")
-    @web_ns.doc(description="Stop a running completion message task.")
-    @web_ns.doc(params={"task_id": {"description": "Task ID to stop", "type": "string", "required": True}})
-    @web_ns.doc(
+    @api.doc("Stop Completion Message")
+    @api.doc(description="Stop a running completion message task.")
+    @api.doc(params={"task_id": {"description": "Task ID to stop", "type": "string", "required": True}})
+    @api.doc(
         responses={
             200: "Success",
             400: "Bad Request",
@@ -131,11 +129,10 @@ class CompletionStopApi(WebApiResource):
         return {"result": "success"}, 200
 
 
-@web_ns.route("/chat-messages")
 class ChatApi(WebApiResource):
-    @web_ns.doc("Create Chat Message")
-    @web_ns.doc(description="Create a chat message for conversational applications.")
-    @web_ns.doc(
+    @api.doc("Create Chat Message")
+    @api.doc(description="Create a chat message for conversational applications.")
+    @api.doc(
         params={
             "inputs": {"description": "Input variables for the chat", "type": "object", "required": True},
             "query": {"description": "User query/message", "type": "string", "required": True},
@@ -151,7 +148,7 @@ class ChatApi(WebApiResource):
             "retriever_from": {"description": "Source of retriever", "type": "string", "required": False},
         }
     )
-    @web_ns.doc(
+    @api.doc(
         responses={
             200: "Success",
             400: "Bad Request",
@@ -210,12 +207,11 @@ class ChatApi(WebApiResource):
             raise InternalServerError()
 
 
-@web_ns.route("/chat-messages/<string:task_id>/stop")
 class ChatStopApi(WebApiResource):
-    @web_ns.doc("Stop Chat Message")
-    @web_ns.doc(description="Stop a running chat message task.")
-    @web_ns.doc(params={"task_id": {"description": "Task ID to stop", "type": "string", "required": True}})
-    @web_ns.doc(
+    @api.doc("Stop Chat Message")
+    @api.doc(description="Stop a running chat message task.")
+    @api.doc(params={"task_id": {"description": "Task ID to stop", "type": "string", "required": True}})
+    @api.doc(
         responses={
             200: "Success",
             400: "Bad Request",
@@ -233,3 +229,9 @@ class ChatStopApi(WebApiResource):
         AppQueueManager.set_stop_flag(task_id, InvokeFrom.WEB_APP, end_user.id)
 
         return {"result": "success"}, 200
+
+
+api.add_resource(CompletionApi, "/completion-messages")
+api.add_resource(CompletionStopApi, "/completion-messages/<string:task_id>/stop")
+api.add_resource(ChatApi, "/chat-messages")
+api.add_resource(ChatStopApi, "/chat-messages/<string:task_id>/stop")

api/controllers/web/conversation.py

@@ -3,7 +3,7 @@ from flask_restx.inputs import int_range
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import NotFound
 
-from controllers.web import web_ns
+from controllers.web import api
 from controllers.web.error import NotChatAppError
 from controllers.web.wraps import WebApiResource
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -16,44 +16,7 @@ from services.errors.conversation import ConversationNotExistsError, LastConvers
 from services.web_conversation_service import WebConversationService
 
 
-@web_ns.route("/conversations")
 class ConversationListApi(WebApiResource):
-    @web_ns.doc("Get Conversation List")
-    @web_ns.doc(description="Retrieve paginated list of conversations for a chat application.")
-    @web_ns.doc(
-        params={
-            "last_id": {"description": "Last conversation ID for pagination", "type": "string", "required": False},
-            "limit": {
-                "description": "Number of conversations to return (1-100)",
-                "type": "integer",
-                "required": False,
-                "default": 20,
-            },
-            "pinned": {
-                "description": "Filter by pinned status",
-                "type": "string",
-                "enum": ["true", "false"],
-                "required": False,
-            },
-            "sort_by": {
-                "description": "Sort order",
-                "type": "string",
-                "enum": ["created_at", "-created_at", "updated_at", "-updated_at"],
-                "required": False,
-                "default": "-updated_at",
-            },
-        }
-    )
-    @web_ns.doc(
-        responses={
-            200: "Success",
-            400: "Bad Request",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "App Not Found or Not a Chat App",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(conversation_infinite_scroll_pagination_fields)
     def get(self, app_model, end_user):
         app_mode = AppMode.value_of(app_model.mode)
@@ -94,25 +57,11 @@ class ConversationListApi(WebApiResource):
             raise NotFound("Last Conversation Not Exists.")
 
 
-@web_ns.route("/conversations/<uuid:c_id>")
 class ConversationApi(WebApiResource):
     delete_response_fields = {
         "result": fields.String,
     }
 
-    @web_ns.doc("Delete Conversation")
-    @web_ns.doc(description="Delete a specific conversation.")
-    @web_ns.doc(params={"c_id": {"description": "Conversation UUID", "type": "string", "required": True}})
-    @web_ns.doc(
-        responses={
-            204: "Conversation deleted successfully",
-            400: "Bad Request",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Conversation Not Found or Not a Chat App",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(delete_response_fields)
     def delete(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
@@ -127,32 +76,7 @@ class ConversationApi(WebApiResource):
         return {"result": "success"}, 204
 
 
-@web_ns.route("/conversations/<uuid:c_id>/name")
 class ConversationRenameApi(WebApiResource):
-    @web_ns.doc("Rename Conversation")
-    @web_ns.doc(description="Rename a specific conversation with a custom name or auto-generate one.")
-    @web_ns.doc(params={"c_id": {"description": "Conversation UUID", "type": "string", "required": True}})
-    @web_ns.doc(
-        params={
-            "name": {"description": "New conversation name", "type": "string", "required": False},
-            "auto_generate": {
-                "description": "Auto-generate conversation name",
-                "type": "boolean",
-                "required": False,
-                "default": False,
-            },
-        }
-    )
-    @web_ns.doc(
-        responses={
-            200: "Conversation renamed successfully",
-            400: "Bad Request",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Conversation Not Found or Not a Chat App",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(simple_conversation_fields)
     def post(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
@@ -172,25 +96,11 @@ class ConversationRenameApi(WebApiResource):
             raise NotFound("Conversation Not Exists.")
 
 
-@web_ns.route("/conversations/<uuid:c_id>/pin")
 class ConversationPinApi(WebApiResource):
     pin_response_fields = {
         "result": fields.String,
     }
 
-    @web_ns.doc("Pin Conversation")
-    @web_ns.doc(description="Pin a specific conversation to keep it at the top of the list.")
-    @web_ns.doc(params={"c_id": {"description": "Conversation UUID", "type": "string", "required": True}})
-    @web_ns.doc(
-        responses={
-            200: "Conversation pinned successfully",
-            400: "Bad Request",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Conversation Not Found or Not a Chat App",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(pin_response_fields)
     def patch(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
@@ -207,25 +117,11 @@ class ConversationPinApi(WebApiResource):
         return {"result": "success"}
 
 
-@web_ns.route("/conversations/<uuid:c_id>/unpin")
 class ConversationUnPinApi(WebApiResource):
     unpin_response_fields = {
         "result": fields.String,
     }
 
-    @web_ns.doc("Unpin Conversation")
-    @web_ns.doc(description="Unpin a specific conversation to remove it from the top of the list.")
-    @web_ns.doc(params={"c_id": {"description": "Conversation UUID", "type": "string", "required": True}})
-    @web_ns.doc(
-        responses={
-            200: "Conversation unpinned successfully",
-            400: "Bad Request",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Conversation Not Found or Not a Chat App",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(unpin_response_fields)
     def patch(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
@@ -236,3 +132,10 @@ class ConversationUnPinApi(WebApiResource):
         WebConversationService.unpin(app_model, conversation_id, end_user)
 
         return {"result": "success"}
+
+
+api.add_resource(ConversationRenameApi, "/conversations/<uuid:c_id>/name", endpoint="web_conversation_name")
+api.add_resource(ConversationListApi, "/conversations")
+api.add_resource(ConversationApi, "/conversations/<uuid:c_id>")
+api.add_resource(ConversationPinApi, "/conversations/<uuid:c_id>/pin")
+api.add_resource(ConversationUnPinApi, "/conversations/<uuid:c_id>/unpin")

api/controllers/web/files.py

@@ -11,6 +11,7 @@ from controllers.common.errors import (
 )
 from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
+from extensions.ext_database import db
 from fields.file_fields import build_file_model
 from services.file_service import FileService
 
@@ -68,7 +69,7 @@ class FileApi(WebApiResource):
             source = None
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,

api/controllers/web/message.py

@@ -4,7 +4,7 @@ from flask_restx import fields, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from werkzeug.exceptions import InternalServerError, NotFound
 
-from controllers.web import web_ns
+from controllers.web import api
 from controllers.web.error import (
     AppMoreLikeThisDisabledError,
     AppSuggestedQuestionsAfterAnswerDisabledError,
@@ -38,7 +38,6 @@ from services.message_service import MessageService
 logger = logging.getLogger(__name__)
 
 
-@web_ns.route("/messages")
 class MessageListApi(WebApiResource):
     message_fields = {
         "id": fields.String,
@@ -63,30 +62,6 @@ class MessageListApi(WebApiResource):
         "data": fields.List(fields.Nested(message_fields)),
     }
 
-    @web_ns.doc("Get Message List")
-    @web_ns.doc(description="Retrieve paginated list of messages from a conversation in a chat application.")
-    @web_ns.doc(
-        params={
-            "conversation_id": {"description": "Conversation UUID", "type": "string", "required": True},
-            "first_id": {"description": "First message ID for pagination", "type": "string", "required": False},
-            "limit": {
-                "description": "Number of messages to return (1-100)",
-                "type": "integer",
-                "required": False,
-                "default": 20,
-            },
-        }
-    )
-    @web_ns.doc(
-        responses={
-            200: "Success",
-            400: "Bad Request",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Conversation Not Found or Not a Chat App",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(message_infinite_scroll_pagination_fields)
     def get(self, app_model, end_user):
         app_mode = AppMode.value_of(app_model.mode)
@@ -109,36 +84,11 @@ class MessageListApi(WebApiResource):
             raise NotFound("First Message Not Exists.")
 
 
-@web_ns.route("/messages/<uuid:message_id>/feedbacks")
 class MessageFeedbackApi(WebApiResource):
     feedback_response_fields = {
         "result": fields.String,
     }
 
-    @web_ns.doc("Create Message Feedback")
-    @web_ns.doc(description="Submit feedback (like/dislike) for a specific message.")
-    @web_ns.doc(params={"message_id": {"description": "Message UUID", "type": "string", "required": True}})
-    @web_ns.doc(
-        params={
-            "rating": {
-                "description": "Feedback rating",
-                "type": "string",
-                "enum": ["like", "dislike"],
-                "required": False,
-            },
-            "content": {"description": "Feedback content/comment", "type": "string", "required": False},
-        }
-    )
-    @web_ns.doc(
-        responses={
-            200: "Feedback submitted successfully",
-            400: "Bad Request",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Message Not Found",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(feedback_response_fields)
     def post(self, app_model, end_user, message_id):
         message_id = str(message_id)
@@ -162,31 +112,7 @@ class MessageFeedbackApi(WebApiResource):
         return {"result": "success"}
 
 
-@web_ns.route("/messages/<uuid:message_id>/more-like-this")
 class MessageMoreLikeThisApi(WebApiResource):
-    @web_ns.doc("Generate More Like This")
-    @web_ns.doc(description="Generate a new completion similar to an existing message (completion apps only).")
-    @web_ns.doc(
-        params={
-            "message_id": {"description": "Message UUID", "type": "string", "required": True},
-            "response_mode": {
-                "description": "Response mode",
-                "type": "string",
-                "enum": ["blocking", "streaming"],
-                "required": True,
-            },
-        }
-    )
-    @web_ns.doc(
-        responses={
-            200: "Success",
-            400: "Bad Request - Not a completion app or feature disabled",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Message Not Found",
-            500: "Internal Server Error",
-        }
-    )
     def get(self, app_model, end_user, message_id):
         if app_model.mode != "completion":
             raise NotCompletionAppError()
@@ -230,25 +156,11 @@ class MessageMoreLikeThisApi(WebApiResource):
             raise InternalServerError()
 
 
-@web_ns.route("/messages/<uuid:message_id>/suggested-questions")
 class MessageSuggestedQuestionApi(WebApiResource):
     suggested_questions_response_fields = {
         "data": fields.List(fields.String),
     }
 
-    @web_ns.doc("Get Suggested Questions")
-    @web_ns.doc(description="Get suggested follow-up questions after a message (chat apps only).")
-    @web_ns.doc(params={"message_id": {"description": "Message UUID", "type": "string", "required": True}})
-    @web_ns.doc(
-        responses={
-            200: "Success",
-            400: "Bad Request - Not a chat app or feature disabled",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Message Not Found or Conversation Not Found",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(suggested_questions_response_fields)
     def get(self, app_model, end_user, message_id):
         app_mode = AppMode.value_of(app_model.mode)
@@ -280,3 +192,9 @@ class MessageSuggestedQuestionApi(WebApiResource):
             raise InternalServerError()
 
         return {"data": questions}
+
+
+api.add_resource(MessageListApi, "/messages")
+api.add_resource(MessageFeedbackApi, "/messages/<uuid:message_id>/feedbacks")
+api.add_resource(MessageMoreLikeThisApi, "/messages/<uuid:message_id>/more-like-this")
+api.add_resource(MessageSuggestedQuestionApi, "/messages/<uuid:message_id>/suggested-questions")

api/controllers/web/remote_files.py

@@ -14,6 +14,7 @@ from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
+from extensions.ext_database import db
 from fields.file_fields import build_file_with_signed_url_model, build_remote_file_info_model
 from services.file_service import FileService
 
@@ -119,7 +120,7 @@ class RemoteFileUploadApi(WebApiResource):
         content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file_info.filename,
                 content=content,
                 mimetype=file_info.mimetype,

api/controllers/web/saved_message.py

@@ -2,7 +2,7 @@ from flask_restx import fields, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from werkzeug.exceptions import NotFound
 
-from controllers.web import web_ns
+from controllers.web import api
 from controllers.web.error import NotCompletionAppError
 from controllers.web.wraps import WebApiResource
 from fields.conversation_fields import message_file_fields
@@ -23,7 +23,6 @@ message_fields = {
 }
 
 
-@web_ns.route("/saved-messages")
 class SavedMessageListApi(WebApiResource):
     saved_message_infinite_scroll_pagination_fields = {
         "limit": fields.Integer,
@@ -35,29 +34,6 @@ class SavedMessageListApi(WebApiResource):
         "result": fields.String,
     }
 
-    @web_ns.doc("Get Saved Messages")
-    @web_ns.doc(description="Retrieve paginated list of saved messages for a completion application.")
-    @web_ns.doc(
-        params={
-            "last_id": {"description": "Last message ID for pagination", "type": "string", "required": False},
-            "limit": {
-                "description": "Number of messages to return (1-100)",
-                "type": "integer",
-                "required": False,
-                "default": 20,
-            },
-        }
-    )
-    @web_ns.doc(
-        responses={
-            200: "Success",
-            400: "Bad Request - Not a completion app",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "App Not Found",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(saved_message_infinite_scroll_pagination_fields)
     def get(self, app_model, end_user):
         if app_model.mode != "completion":
@@ -70,23 +46,6 @@ class SavedMessageListApi(WebApiResource):
 
         return SavedMessageService.pagination_by_last_id(app_model, end_user, args["last_id"], args["limit"])
 
-    @web_ns.doc("Save Message")
-    @web_ns.doc(description="Save a specific message for later reference.")
-    @web_ns.doc(
-        params={
-            "message_id": {"description": "Message UUID to save", "type": "string", "required": True},
-        }
-    )
-    @web_ns.doc(
-        responses={
-            200: "Message saved successfully",
-            400: "Bad Request - Not a completion app",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Message Not Found",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(post_response_fields)
     def post(self, app_model, end_user):
         if app_model.mode != "completion":
@@ -104,25 +63,11 @@ class SavedMessageListApi(WebApiResource):
         return {"result": "success"}
 
 
-@web_ns.route("/saved-messages/<uuid:message_id>")
 class SavedMessageApi(WebApiResource):
     delete_response_fields = {
         "result": fields.String,
     }
 
-    @web_ns.doc("Delete Saved Message")
-    @web_ns.doc(description="Remove a message from saved messages.")
-    @web_ns.doc(params={"message_id": {"description": "Message UUID to delete", "type": "string", "required": True}})
-    @web_ns.doc(
-        responses={
-            204: "Message removed successfully",
-            400: "Bad Request - Not a completion app",
-            401: "Unauthorized",
-            403: "Forbidden",
-            404: "Message Not Found",
-            500: "Internal Server Error",
-        }
-    )
     @marshal_with(delete_response_fields)
     def delete(self, app_model, end_user, message_id):
         message_id = str(message_id)
@@ -133,3 +78,7 @@ class SavedMessageApi(WebApiResource):
         SavedMessageService.delete(app_model, end_user, message_id)
 
         return {"result": "success"}, 204
+
+
+api.add_resource(SavedMessageListApi, "/saved-messages")
+api.add_resource(SavedMessageApi, "/saved-messages/<uuid:message_id>")