Refactor WorkflowService to handle missing default credentials gracefully

.devcontainer/Dockerfile

@@ -1,4 +1,4 @@
-FROM mcr.microsoft.com/devcontainers/python:3.12-bullseye
+FROM mcr.microsoft.com/devcontainers/python:3.12
 
 RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \
      && apt-get -y install libgmp-dev libmpfr-dev libmpc-dev

.github/dependabot.yml

@@ -1,12 +0,0 @@
-version: 2
-updates:
-  - package-ecosystem: "npm"
-    directory: "/web"
-    schedule:
-      interval: "weekly"
-    open-pull-requests-limit: 2
-  - package-ecosystem: "uv"
-    directory: "/api"
-    schedule:
-      interval: "weekly"
-    open-pull-requests-limit: 2

.github/workflows/autofix.yml

@@ -22,33 +22,13 @@ jobs:
           # Fix lint errors
           uv run ruff check --fix .
           # Format code
-          uv run ruff format ..
+          uv run ruff format .
 
       - name: ast-grep
         run: |
           uvx --from ast-grep-cli sg --pattern 'db.session.query($WHATEVER).filter($HERE)' --rewrite 'db.session.query($WHATEVER).where($HERE)' -l py --update-all
           uvx --from ast-grep-cli sg --pattern 'session.query($WHATEVER).filter($HERE)' --rewrite 'session.query($WHATEVER).where($HERE)' -l py --update-all
-          # Convert Optional[T] to T | None (ignoring quoted types)
-          cat > /tmp/optional-rule.yml << 'EOF'
-          id: convert-optional-to-union
-          language: python
-          rule:
-            kind: generic_type
-            all:
-              - has:
-                  kind: identifier
-                  pattern: Optional
-              - has:
-                  kind: type_parameter
-                  has:
-                    kind: type
-                    pattern: $T
-          fix: $T | None
-          EOF
-          uvx --from ast-grep-cli sg scan --inline-rules "$(cat /tmp/optional-rule.yml)" --update-all
-          # Fix forward references that were incorrectly converted (Python doesn't support "Type" | None syntax)
-          find . -name "*.py" -type f -exec sed -i.bak -E 's/"([^"]+)" \| None/Optional["\1"]/g; s/'"'"'([^'"'"']+)'"'"' \| None/Optional['"'"'\1'"'"']/g' {} \;
-          find . -name "*.py.bak" -type f -delete
+
 
       - name: mdformat
         run: |

.github/workflows/build-push.yml

@@ -8,8 +8,6 @@ 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/rag-dev"
+      - "deploy/dev"
     types:
       - completed
 
@@ -12,13 +12,12 @@ jobs:
   deploy:
     runs-on: ubuntu-latest
     if: |
-      github.event.workflow_run.conclusion == 'success' &&
-      github.event.workflow_run.head_branch == 'deploy/rag-dev'
+      github.event.workflow_run.conclusion == 'success'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v0.1.8
         with:
-          host: ${{ secrets.RAG_SSH_HOST }}
+          host: ${{ secrets.SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.github/workflows/style.yml

@@ -12,6 +12,7 @@ permissions:
   statuses: write
   contents: read
 
+
 jobs:
   python-style:
     name: Python Style
@@ -43,10 +44,6 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv sync --project api --dev
 
-      - name: Run Import Linter
-        if: steps.changed-files.outputs.any_changed == 'true'
-        run: uv run --directory api --dev lint-imports
-
       - name: Run Basedpyright Checks
         if: steps.changed-files.outputs.any_changed == 'true'
         run: dev/basedpyright-check
@@ -102,6 +99,7 @@ jobs:
         working-directory: ./web
         run: |
           pnpm run lint
+          pnpm run eslint
 
   docker-compose-template:
     name: Docker Compose Template

AGENTS.md

@@ -1,87 +0,0 @@
-# AGENTS.md
-
-## Project Overview
-
-Dify is an open-source platform for developing LLM applications with an intuitive interface combining agentic AI workflows, RAG pipelines, agent capabilities, and model management.
-
-The codebase consists of:
-
-- **Backend API** (`/api`): Python Flask application with Domain-Driven Design architecture
-- **Frontend Web** (`/web`): Next.js 15 application with TypeScript and React 19
-- **Docker deployment** (`/docker`): Containerized deployment configurations
-
-## Development Commands
-
-### Backend (API)
-
-All Python commands must be prefixed with `uv run --project api`:
-
-```bash
-# Start development servers
-./dev/start-api                   # Start API server
-./dev/start-worker                # Start Celery worker
-
-# Run tests
-uv run --project api pytest      # Run all tests
-uv run --project api pytest tests/unit_tests/     # Unit tests only
-uv run --project api pytest tests/integration_tests/  # Integration tests
-
-# Code quality
-./dev/reformat                    # Run all formatters and linters
-uv run --project api ruff check --fix ./    # Fix linting issues
-uv run --project api ruff format ./         # Format code
-uv run --directory api basedpyright         # Type checking
-```
-
-### Frontend (Web)
-
-```bash
-cd web
-pnpm lint                         # Run ESLint
-pnpm eslint-fix                   # Fix ESLint issues
-pnpm test                         # Run Jest tests
-```
-
-## Testing Guidelines
-
-### Backend Testing
-
-- Use `pytest` for all backend tests
-- Write tests first (TDD approach)
-- Test structure: Arrange-Act-Assert
-
-## Code Style Requirements
-
-### Python
-
-- Use type hints for all functions and class attributes
-- No `Any` types unless absolutely necessary
-- Implement special methods (`__repr__`, `__str__`) appropriately
-
-### TypeScript/JavaScript
-
-- Strict TypeScript configuration
-- ESLint with Prettier integration
-- Avoid `any` type
-
-## Important Notes
-
-- **Environment Variables**: Always use UV for Python commands: `uv run --project api <command>`
-- **Comments**: Only write meaningful comments that explain "why", not "what"
-- **File Creation**: Always prefer editing existing files over creating new ones
-- **Documentation**: Don't create documentation files unless explicitly requested
-- **Code Quality**: Always run `./dev/reformat` before committing backend changes
-
-## Common Development Tasks
-
-### Adding a New API Endpoint
-
-1. Create controller in `/api/controllers/`
-1. Add service logic in `/api/services/`
-1. Update routes in controller's `__init__.py`
-1. Write tests in `/api/tests/`
-
-## Project-Specific Conventions
-
-- All async tasks use Celery with Redis as broker
-- **Internationalization**: Frontend supports multiple languages with English (`web/i18n/en-US/`) as the source. All user-facing text must use i18n keys, no hardcoded strings. Edit corresponding module files in `en-US/` directory for translations.

AGENTS.md

@@ -0,0 +1 @@
+CLAUDE.md

CLAUDE.md

@@ -1 +0,0 @@
-AGENTS.md

CLAUDE.md

@@ -0,0 +1,89 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project Overview
+
+Dify is an open-source platform for developing LLM applications with an intuitive interface combining agentic AI workflows, RAG pipelines, agent capabilities, and model management.
+
+The codebase consists of:
+
+- **Backend API** (`/api`): Python Flask application with Domain-Driven Design architecture
+- **Frontend Web** (`/web`): Next.js 15 application with TypeScript and React 19
+- **Docker deployment** (`/docker`): Containerized deployment configurations
+
+## Development Commands
+
+### Backend (API)
+
+All Python commands must be prefixed with `uv run --project api`:
+
+```bash
+# Start development servers
+./dev/start-api                   # Start API server
+./dev/start-worker                # Start Celery worker
+
+# Run tests
+uv run --project api pytest      # Run all tests
+uv run --project api pytest tests/unit_tests/     # Unit tests only
+uv run --project api pytest tests/integration_tests/  # Integration tests
+
+# Code quality
+./dev/reformat                    # Run all formatters and linters
+uv run --project api ruff check --fix ./    # Fix linting issues
+uv run --project api ruff format ./         # Format code
+uv run --directory api basedpyright         # Type checking
+```
+
+### Frontend (Web)
+
+```bash
+cd web
+pnpm lint                         # Run ESLint
+pnpm eslint-fix                   # Fix ESLint issues
+pnpm test                         # Run Jest tests
+```
+
+## Testing Guidelines
+
+### Backend Testing
+
+- Use `pytest` for all backend tests
+- Write tests first (TDD approach)
+- Test structure: Arrange-Act-Assert
+
+## Code Style Requirements
+
+### Python
+
+- Use type hints for all functions and class attributes
+- No `Any` types unless absolutely necessary
+- Implement special methods (`__repr__`, `__str__`) appropriately
+
+### TypeScript/JavaScript
+
+- Strict TypeScript configuration
+- ESLint with Prettier integration
+- Avoid `any` type
+
+## Important Notes
+
+- **Environment Variables**: Always use UV for Python commands: `uv run --project api <command>`
+- **Comments**: Only write meaningful comments that explain "why", not "what"
+- **File Creation**: Always prefer editing existing files over creating new ones
+- **Documentation**: Don't create documentation files unless explicitly requested
+- **Code Quality**: Always run `./dev/reformat` before committing backend changes
+
+## Common Development Tasks
+
+### Adding a New API Endpoint
+
+1. Create controller in `/api/controllers/`
+1. Add service logic in `/api/services/`
+1. Update routes in controller's `__init__.py`
+1. Write tests in `/api/tests/`
+
+## Project-Specific Conventions
+
+- All async tasks use Celery with Redis as broker
+- **Internationalization**: Frontend supports multiple languages with English (`web/i18n/en-US/`) as the source. All user-facing text must use i18n keys, no hardcoded strings. Edit corresponding module files in `en-US/` directory for translations.

api/.env.example

@@ -76,7 +76,6 @@ DB_HOST=localhost
 DB_PORT=5432
 DB_DATABASE=dify
 SQLALCHEMY_POOL_PRE_PING=true
-SQLALCHEMY_POOL_TIMEOUT=30
 
 # Storage configuration
 # use for store upload files, private keys...
@@ -329,7 +328,7 @@ MATRIXONE_DATABASE=dify
 LINDORM_URL=http://ld-*******************-proxy-search-pub.lindorm.aliyuncs.com:30070
 LINDORM_USERNAME=admin
 LINDORM_PASSWORD=admin
-LINDORM_USING_UGC=True
+USING_UGC_INDEX=False
 LINDORM_QUERY_TIMEOUT=1
 
 # OceanBase Vector configuration
@@ -462,16 +461,6 @@ 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)

api/.importlinter

@@ -1,105 +0,0 @@
-[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.loop.loop_node -> core.workflow.graph_events
-
-    core.workflow.nodes.node_factory -> core.workflow.graph
-    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_engine
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine.command_channels
-
-[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_handler
-    graph_traversal
-    graph_state_manager
-    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: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

@@ -5,7 +5,7 @@ line-length = 120
 quote-style = "double"
 
 [lint]
-preview = true
+preview = false
 select = [
     "B",       # flake8-bugbear rules
     "C4",      # flake8-comprehensions
@@ -65,7 +65,6 @@ ignore = [
     "B006",    # mutable-argument-default
     "B007",    # unused-loop-control-variable
     "B026",    # star-arg-unpacking-after-keyword-arg
-    "B901",    # allow return in yield
     "B903",    # class-as-data-structure
     "B904",    # raise-without-from-inside-except
     "B905",    # zip-without-explicit-strict

api/app.py

@@ -1,3 +1,4 @@
+import os
 import sys
 
 
@@ -16,20 +17,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
-    #
-    # # gevent
-    # monkey.patch_all()
-    #
-    # from grpc.experimental import gevent as grpc_gevent  # type: ignore
-    #
-    # # grpc gevent
-    # grpc_gevent.init_gevent()
+    if (flask_debug := os.environ.get("FLASK_DEBUG", "0")) and flask_debug.lower() in {"false", "0", "no"}:
+        from gevent import monkey
 
-    # import psycogreen.gevent  # type: ignore
-    #
-    # psycogreen.gevent.patch_psycopg()
+        # 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()
 
     from app_factory import create_app
 

api/celery_entrypoint.py

@@ -1,22 +0,0 @@
-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

@@ -2,7 +2,7 @@ import base64
 import json
 import logging
 import secrets
-from typing import Any
+from typing import Any, Optional
 
 import click
 import sqlalchemy as sa
@@ -13,37 +13,29 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from constants.languages import languages
-from core.helper import encrypter
-from core.plugin.impl.plugin import PluginInstaller
+from core.plugin.entities.plugin import ToolProviderID
 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
 from extensions.ext_redis import redis_client
 from extensions.ext_storage import storage
-from extensions.storage.opendal_storage import OpenDALStorage
-from extensions.storage.storage_type import StorageType
 from libs.helper import email as email_validate
 from libs.password import hash_password, password_pattern, valid_password
 from libs.rsa import generate_key_pair
 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, UploadFile
-from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
+from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation
 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__)
@@ -647,7 +639,7 @@ def old_metadata_migration():
 @click.option("--email", prompt=True, help="Tenant account email.")
 @click.option("--name", prompt=True, help="Workspace name.")
 @click.option("--language", prompt=True, help="Account language, default: en-US.")
-def create_tenant(email: str, language: str | None = None, name: str | None = None):
+def create_tenant(email: str, language: Optional[str] = None, name: Optional[str] = None):
     """
     Create tenant account
     """
@@ -961,7 +953,7 @@ def clear_orphaned_file_records(force: bool):
             click.echo(click.style("- Deleting orphaned message_files records", fg="white"))
             query = "DELETE FROM message_files WHERE id IN :ids"
             with db.engine.begin() as conn:
-                conn.execute(sa.text(query), {"ids": tuple(record["id"] for record in orphaned_message_files)})
+                conn.execute(sa.text(query), {"ids": tuple([record["id"] for record in orphaned_message_files])})
             click.echo(
                 click.style(f"Removed {len(orphaned_message_files)} orphaned message_files records.", fg="green")
             )
@@ -1253,17 +1245,15 @@ 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, including associated file counts.
+    Count orphaned draft variables by app.
 
     Returns:
-        Dictionary with statistics about orphaned variables and files
+        Dictionary with statistics about orphaned variables
     """
-    # Count orphaned variables by app
-    variables_query = """
+    query = """
         SELECT
             wdv.app_id,
-            COUNT(*) as variable_count,
-            COUNT(wdv.file_id) as file_count
+            COUNT(*) as variable_count
         FROM workflow_draft_variables AS wdv
         WHERE NOT EXISTS(
             SELECT 1 FROM apps WHERE apps.id = wdv.app_id
@@ -1273,21 +1263,14 @@ def _count_orphaned_draft_variables() -> dict[str, Any]:
     """
 
     with db.engine.connect() as conn:
-        result = conn.execute(sa.text(variables_query))
-        orphaned_by_app = {}
-        total_files = 0
+        result = conn.execute(sa.text(query))
+        orphaned_by_app = {row[0]: row[1] for row in result}
 
-        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())
+        total_orphaned = sum(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,
         }
@@ -1316,7 +1299,6 @@ 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:
@@ -1325,10 +1307,10 @@ def cleanup_orphaned_draft_variables(
 
     if dry_run:
         logger.info("DRY RUN: Would delete the following:")
-        for app_id, data in sorted(stats["orphaned_by_app"].items(), key=lambda x: x[1]["variables"], reverse=True)[
+        for app_id, count in sorted(stats["orphaned_by_app"].items(), key=lambda x: x[1], reverse=True)[
             :10
         ]:  # Show top 10
-            logger.info("  App %s: %s variables, %s files", app_id, data["variables"], data["files"])
+            logger.info("  App %s: %s variables", app_id, count)
         if len(stats["orphaned_by_app"]) > 10:
             logger.info("  ... and %s more apps", len(stats["orphaned_by_app"]) - 10)
         return
@@ -1337,8 +1319,7 @@ 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 and {stats['total_orphaned_files']} associated files "
-            f"from {stats['orphaned_app_count']} apps?",
+            f"orphaned draft variables from {stats['orphaned_app_count']} apps?",
             abort=True,
         )
 
@@ -1371,425 +1352,3 @@ 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)  # pyright: ignore[reportPrivateUsage]
-        firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)  # pyright: ignore[reportPrivateUsage]
-        jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)  # pyright: ignore[reportPrivateUsage]
-        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 notion_credential in notion_credentials:
-                tenant_id = notion_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(notion_credential)
-            for tenant_id, notion_tenant_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 notion_tenant_credential in notion_tenant_credentials:
-                    auth_count += 1
-                    # get credential oauth params
-                    access_token = notion_tenant_credential.access_token
-                    # notion info
-                    notion_info = notion_tenant_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 firecrawl_credential in firecrawl_credentials:
-                tenant_id = firecrawl_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(firecrawl_credential)
-            for tenant_id, firecrawl_tenant_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 firecrawl_tenant_credential in firecrawl_tenant_credentials:
-                    auth_count += 1
-                    # get credential api key
-                    credentials_json = json.loads(firecrawl_tenant_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 jina_credential in jina_credentials:
-                tenant_id = jina_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(jina_credential)
-            for tenant_id, jina_tenant_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 jina_tenant_credential in jina_tenant_credentials:
-                    auth_count += 1
-                    # get credential api key
-                    credentials_json = json.loads(jina_tenant_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"))
-
-
-@click.command(
-    "migrate-oss",
-    help="Migrate files from Local or OpenDAL source to a cloud OSS storage (destination must NOT be local/opendal).",
-)
-@click.option(
-    "--path",
-    "paths",
-    multiple=True,
-    help="Storage path prefixes to migrate (repeatable). Defaults: privkeys, upload_files, image_files,"
-    " tools, website_files, keyword_files, ops_trace",
-)
-@click.option(
-    "--source",
-    type=click.Choice(["local", "opendal"], case_sensitive=False),
-    default="opendal",
-    show_default=True,
-    help="Source storage type to read from",
-)
-@click.option("--overwrite", is_flag=True, default=False, help="Overwrite destination if file already exists")
-@click.option("--dry-run", is_flag=True, default=False, help="Show what would be migrated without uploading")
-@click.option("-f", "--force", is_flag=True, help="Skip confirmation and run without prompts")
-@click.option(
-    "--update-db/--no-update-db",
-    default=True,
-    help="Update upload_files.storage_type from source type to current storage after migration",
-)
-def migrate_oss(
-    paths: tuple[str, ...],
-    source: str,
-    overwrite: bool,
-    dry_run: bool,
-    force: bool,
-    update_db: bool,
-):
-    """
-    Copy all files under selected prefixes from a source storage
-    (Local filesystem or OpenDAL-backed) into the currently configured
-    destination storage backend, then optionally update DB records.
-
-    Expected usage: set STORAGE_TYPE (and its credentials) to your target backend.
-    """
-    # Ensure target storage is not local/opendal
-    if dify_config.STORAGE_TYPE in (StorageType.LOCAL, StorageType.OPENDAL):
-        click.echo(
-            click.style(
-                "Target STORAGE_TYPE must be a cloud OSS (not 'local' or 'opendal').\n"
-                "Please set STORAGE_TYPE to one of: s3, aliyun-oss, azure-blob, google-storage, tencent-cos, \n"
-                "volcengine-tos, supabase, oci-storage, huawei-obs, baidu-obs, clickzetta-volume.",
-                fg="red",
-            )
-        )
-        return
-
-    # Default paths if none specified
-    default_paths = ("privkeys", "upload_files", "image_files", "tools", "website_files", "keyword_files", "ops_trace")
-    path_list = list(paths) if paths else list(default_paths)
-    is_source_local = source.lower() == "local"
-
-    click.echo(click.style("Preparing migration to target storage.", fg="yellow"))
-    click.echo(click.style(f"Target storage type: {dify_config.STORAGE_TYPE}", fg="white"))
-    if is_source_local:
-        src_root = dify_config.STORAGE_LOCAL_PATH
-        click.echo(click.style(f"Source: local fs, root: {src_root}", fg="white"))
-    else:
-        click.echo(click.style(f"Source: opendal scheme={dify_config.OPENDAL_SCHEME}", fg="white"))
-    click.echo(click.style(f"Paths to migrate: {', '.join(path_list)}", fg="white"))
-    click.echo("")
-
-    if not force:
-        click.confirm("Proceed with migration?", abort=True)
-
-    # Instantiate source storage
-    try:
-        if is_source_local:
-            src_root = dify_config.STORAGE_LOCAL_PATH
-            source_storage = OpenDALStorage(scheme="fs", root=src_root)
-        else:
-            source_storage = OpenDALStorage(scheme=dify_config.OPENDAL_SCHEME)
-    except Exception as e:
-        click.echo(click.style(f"Failed to initialize source storage: {str(e)}", fg="red"))
-        return
-
-    total_files = 0
-    copied_files = 0
-    skipped_files = 0
-    errored_files = 0
-    copied_upload_file_keys: list[str] = []
-
-    for prefix in path_list:
-        click.echo(click.style(f"Scanning source path: {prefix}", fg="white"))
-        try:
-            keys = source_storage.scan(path=prefix, files=True, directories=False)
-        except FileNotFoundError:
-            click.echo(click.style(f"  -> Skipping missing path: {prefix}", fg="yellow"))
-            continue
-        except NotImplementedError:
-            click.echo(click.style("  -> Source storage does not support scanning.", fg="red"))
-            return
-        except Exception as e:
-            click.echo(click.style(f"  -> Error scanning '{prefix}': {str(e)}", fg="red"))
-            continue
-
-        click.echo(click.style(f"Found {len(keys)} files under {prefix}", fg="white"))
-
-        for key in keys:
-            total_files += 1
-
-            # check destination existence
-            if not overwrite:
-                try:
-                    if storage.exists(key):
-                        skipped_files += 1
-                        continue
-                except Exception as e:
-                    # existence check failures should not block migration attempt
-                    # but should be surfaced to user as a warning for visibility
-                    click.echo(
-                        click.style(
-                            f"  -> Warning: failed target existence check for {key}: {str(e)}",
-                            fg="yellow",
-                        )
-                    )
-
-            if dry_run:
-                copied_files += 1
-                continue
-
-            # read from source and write to destination
-            try:
-                data = source_storage.load_once(key)
-            except FileNotFoundError:
-                errored_files += 1
-                click.echo(click.style(f"  -> Missing on source: {key}", fg="yellow"))
-                continue
-            except Exception as e:
-                errored_files += 1
-                click.echo(click.style(f"  -> Error reading {key}: {str(e)}", fg="red"))
-                continue
-
-            try:
-                storage.save(key, data)
-                copied_files += 1
-                if prefix == "upload_files":
-                    copied_upload_file_keys.append(key)
-            except Exception as e:
-                errored_files += 1
-                click.echo(click.style(f"  -> Error writing {key} to target: {str(e)}", fg="red"))
-                continue
-
-    click.echo("")
-    click.echo(click.style("Migration summary:", fg="yellow"))
-    click.echo(click.style(f"  Total:   {total_files}", fg="white"))
-    click.echo(click.style(f"  Copied:  {copied_files}", fg="green"))
-    click.echo(click.style(f"  Skipped: {skipped_files}", fg="white"))
-    if errored_files:
-        click.echo(click.style(f"  Errors:  {errored_files}", fg="red"))
-
-    if dry_run:
-        click.echo(click.style("Dry-run complete. No changes were made.", fg="green"))
-        return
-
-    if errored_files:
-        click.echo(
-            click.style(
-                "Some files failed to migrate. Review errors above before updating DB records.",
-                fg="yellow",
-            )
-        )
-        if update_db and not force:
-            if not click.confirm("Proceed to update DB storage_type despite errors?", default=False):
-                update_db = False
-
-    # Optionally update DB records for upload_files.storage_type (only for successfully copied upload_files)
-    if update_db:
-        if not copied_upload_file_keys:
-            click.echo(click.style("No upload_files copied. Skipping DB storage_type update.", fg="yellow"))
-        else:
-            try:
-                source_storage_type = StorageType.LOCAL if is_source_local else StorageType.OPENDAL
-                updated = (
-                    db.session.query(UploadFile)
-                    .where(
-                        UploadFile.storage_type == source_storage_type,
-                        UploadFile.key.in_(copied_upload_file_keys),
-                    )
-                    .update({UploadFile.storage_type: dify_config.STORAGE_TYPE}, synchronize_session=False)
-                )
-                db.session.commit()
-                click.echo(click.style(f"Updated storage_type for {updated} upload_files records.", fg="green"))
-            except Exception as e:
-                db.session.rollback()
-                click.echo(click.style(f"Failed to update DB storage_type: {str(e)}", fg="red"))

api/configs/__init__.py

@@ -1,3 +1,3 @@
 from .app_config import DifyConfig
 
-dify_config = DifyConfig()  # type: ignore
+dify_config = DifyConfig()

api/configs/extra/notion_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,28 +9,28 @@ class NotionConfig(BaseSettings):
     Configuration settings for Notion integration
     """
 
-    NOTION_CLIENT_ID: str | None = Field(
+    NOTION_CLIENT_ID: Optional[str] = Field(
         description="Client ID for Notion API authentication. Required for OAuth 2.0 flow.",
         default=None,
     )
 
-    NOTION_CLIENT_SECRET: str | None = Field(
+    NOTION_CLIENT_SECRET: Optional[str] = Field(
         description="Client secret for Notion API authentication. Required for OAuth 2.0 flow.",
         default=None,
     )
 
-    NOTION_INTEGRATION_TYPE: str | None = Field(
+    NOTION_INTEGRATION_TYPE: Optional[str] = Field(
         description="Type of Notion integration."
         " Set to 'internal' for internal integrations, or None for public integrations.",
         default=None,
     )
 
-    NOTION_INTERNAL_SECRET: str | None = Field(
+    NOTION_INTERNAL_SECRET: Optional[str] = Field(
         description="Secret key for internal Notion integrations. Required when NOTION_INTEGRATION_TYPE is 'internal'.",
         default=None,
     )
 
-    NOTION_INTEGRATION_TOKEN: str | None = Field(
+    NOTION_INTEGRATION_TOKEN: Optional[str] = Field(
         description="Integration token for Notion API access. Used for direct API calls without OAuth flow.",
         default=None,
     )

api/configs/extra/sentry_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, NonNegativeFloat
 from pydantic_settings import BaseSettings
 
@@ -7,7 +9,7 @@ class SentryConfig(BaseSettings):
     Configuration settings for Sentry error tracking and performance monitoring
     """
 
-    SENTRY_DSN: str | None = Field(
+    SENTRY_DSN: Optional[str] = Field(
         description="Sentry Data Source Name (DSN)."
         " This is the unique identifier of your Sentry project, used to send events to the correct project.",
         default=None,

api/configs/feature/__init__.py

@@ -1,5 +1,4 @@
-from enum import StrEnum
-from typing import Literal
+from typing import Literal, Optional
 
 from pydantic import (
     AliasChoices,
@@ -58,7 +57,7 @@ class SecurityConfig(BaseSettings):
         default=False,
     )
 
-    ADMIN_API_KEY: str | None = Field(
+    ADMIN_API_KEY: Optional[str] = Field(
         description="admin api key for authentication",
         default=None,
     )
@@ -98,17 +97,17 @@ class CodeExecutionSandboxConfig(BaseSettings):
         default="dify-sandbox",
     )
 
-    CODE_EXECUTION_CONNECT_TIMEOUT: float | None = Field(
+    CODE_EXECUTION_CONNECT_TIMEOUT: Optional[float] = Field(
         description="Connection timeout in seconds for code execution requests",
         default=10.0,
     )
 
-    CODE_EXECUTION_READ_TIMEOUT: float | None = Field(
+    CODE_EXECUTION_READ_TIMEOUT: Optional[float] = Field(
         description="Read timeout in seconds for code execution requests",
         default=60.0,
     )
 
-    CODE_EXECUTION_WRITE_TIMEOUT: float | None = Field(
+    CODE_EXECUTION_WRITE_TIMEOUT: Optional[float] = Field(
         description="Write timeout in seconds for code execution request",
         default=10.0,
     )
@@ -369,17 +368,17 @@ class HttpConfig(BaseSettings):
         default=3,
     )
 
-    SSRF_PROXY_ALL_URL: str | None = Field(
+    SSRF_PROXY_ALL_URL: Optional[str] = Field(
         description="Proxy URL for HTTP or HTTPS requests to prevent Server-Side Request Forgery (SSRF)",
         default=None,
     )
 
-    SSRF_PROXY_HTTP_URL: str | None = Field(
+    SSRF_PROXY_HTTP_URL: Optional[str] = Field(
         description="Proxy URL for HTTP requests to prevent Server-Side Request Forgery (SSRF)",
         default=None,
     )
 
-    SSRF_PROXY_HTTPS_URL: str | None = Field(
+    SSRF_PROXY_HTTPS_URL: Optional[str] = Field(
         description="Proxy URL for HTTPS requests to prevent Server-Side Request Forgery (SSRF)",
         default=None,
     )
@@ -421,7 +420,7 @@ class InnerAPIConfig(BaseSettings):
         default=False,
     )
 
-    INNER_API_KEY: str | None = Field(
+    INNER_API_KEY: Optional[str] = Field(
         description="API key for accessing the internal API",
         default=None,
     )
@@ -437,7 +436,7 @@ class LoggingConfig(BaseSettings):
         default="INFO",
     )
 
-    LOG_FILE: str | None = Field(
+    LOG_FILE: Optional[str] = Field(
         description="File path for log output.",
         default=None,
     )
@@ -457,12 +456,12 @@ class LoggingConfig(BaseSettings):
         default="%(asctime)s.%(msecs)03d %(levelname)s [%(threadName)s] [%(filename)s:%(lineno)d] - %(message)s",
     )
 
-    LOG_DATEFORMAT: str | None = Field(
+    LOG_DATEFORMAT: Optional[str] = Field(
         description="Date format string for log timestamps",
         default=None,
     )
 
-    LOG_TZ: str | None = Field(
+    LOG_TZ: Optional[str] = Field(
         description="Timezone for log timestamps (e.g., 'America/New_York')",
         default="UTC",
     )
@@ -506,22 +505,6 @@ 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(
-        100000,
-        description="maximum length for string to trigger tuncation, measure in number of characters",
-    )
-    WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH: PositiveInt = Field(
-        1000,
-        description="maximum length for array to trigger truncation.",
-    )
-
-
 class WorkflowConfig(BaseSettings):
     """
     Configuration for workflow execution
@@ -552,28 +535,6 @@ 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):
     """
@@ -634,22 +595,22 @@ class AuthConfig(BaseSettings):
         default="/console/api/oauth/authorize",
     )
 
-    GITHUB_CLIENT_ID: str | None = Field(
+    GITHUB_CLIENT_ID: Optional[str] = Field(
         description="GitHub OAuth client ID",
         default=None,
     )
 
-    GITHUB_CLIENT_SECRET: str | None = Field(
+    GITHUB_CLIENT_SECRET: Optional[str] = Field(
         description="GitHub OAuth client secret",
         default=None,
     )
 
-    GOOGLE_CLIENT_ID: str | None = Field(
+    GOOGLE_CLIENT_ID: Optional[str] = Field(
         description="Google OAuth client ID",
         default=None,
     )
 
-    GOOGLE_CLIENT_SECRET: str | None = Field(
+    GOOGLE_CLIENT_SECRET: Optional[str] = Field(
         description="Google OAuth client secret",
         default=None,
     )
@@ -712,71 +673,47 @@ class ToolConfig(BaseSettings):
     )
 
 
-class TemplateMode(StrEnum):
-    # unsafe mode allows flexible operations in templates, but may cause security vulnerabilities
-    UNSAFE = "unsafe"
-
-    # sandbox mode restricts some unsafe operations like accessing __class__.
-    # however, it is still not 100% safe, for example, cpu exploitation can happen.
-    SANDBOX = "sandbox"
-
-    # templating is disabled
-    DISABLED = "disabled"
-
-
 class MailConfig(BaseSettings):
     """
     Configuration for email services
     """
 
-    MAIL_TEMPLATING_MODE: TemplateMode = Field(
-        description="Template mode for email services",
-        default=TemplateMode.SANDBOX,
-    )
-
-    MAIL_TEMPLATING_TIMEOUT: int = Field(
-        description="""
-        Timeout for email templating in seconds. Used to prevent infinite loops in malicious templates. 
-        Only available in sandbox mode.""",
-        default=3,
-    )
-
-    MAIL_TYPE: str | None = Field(
+    MAIL_TYPE: Optional[str] = Field(
         description="Email service provider type ('smtp' or 'resend' or 'sendGrid), default to None.",
         default=None,
     )
 
-    MAIL_DEFAULT_SEND_FROM: str | None = Field(
+    MAIL_DEFAULT_SEND_FROM: Optional[str] = Field(
         description="Default email address to use as the sender",
         default=None,
     )
 
-    RESEND_API_KEY: str | None = Field(
+    RESEND_API_KEY: Optional[str] = Field(
         description="API key for Resend email service",
         default=None,
     )
 
-    RESEND_API_URL: str | None = Field(
+    RESEND_API_URL: Optional[str] = Field(
         description="API URL for Resend email service",
         default=None,
     )
 
-    SMTP_SERVER: str | None = Field(
+    SMTP_SERVER: Optional[str] = Field(
         description="SMTP server hostname",
         default=None,
     )
 
-    SMTP_PORT: int | None = Field(
+    SMTP_PORT: Optional[int] = Field(
         description="SMTP server port number",
         default=465,
     )
 
-    SMTP_USERNAME: str | None = Field(
+    SMTP_USERNAME: Optional[str] = Field(
         description="Username for SMTP authentication",
         default=None,
     )
 
-    SMTP_PASSWORD: str | None = Field(
+    SMTP_PASSWORD: Optional[str] = Field(
         description="Password for SMTP authentication",
         default=None,
     )
@@ -796,7 +733,7 @@ class MailConfig(BaseSettings):
         default=50,
     )
 
-    SENDGRID_API_KEY: str | None = Field(
+    SENDGRID_API_KEY: Optional[str] = Field(
         description="API key for SendGrid service",
         default=None,
     )
@@ -819,17 +756,17 @@ class RagEtlConfig(BaseSettings):
         default="database",
     )
 
-    UNSTRUCTURED_API_URL: str | None = Field(
+    UNSTRUCTURED_API_URL: Optional[str] = Field(
         description="API URL for Unstructured.io service",
         default=None,
     )
 
-    UNSTRUCTURED_API_KEY: str | None = Field(
+    UNSTRUCTURED_API_KEY: Optional[str] = Field(
         description="API key for Unstructured.io service",
         default="",
     )
 
-    SCARF_NO_ANALYTICS: str | None = Field(
+    SCARF_NO_ANALYTICS: Optional[str] = Field(
         description="This is about whether to disable Scarf analytics in Unstructured library.",
         default="false",
     )
@@ -1104,6 +1041,5 @@ class FeatureConfig(
     CeleryBeatConfig,
     CeleryScheduleTasksConfig,
     WorkflowLogConfig,
-    WorkflowVariableTruncationConfig,
 ):
     pass

api/configs/feature/hosted_service/__init__.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, NonNegativeInt
 from pydantic_settings import BaseSettings
 
@@ -38,17 +40,17 @@ class HostedOpenAiConfig(BaseSettings):
     Configuration for hosted OpenAI service
     """
 
-    HOSTED_OPENAI_API_KEY: str | None = Field(
+    HOSTED_OPENAI_API_KEY: Optional[str] = Field(
         description="API key for hosted OpenAI service",
         default=None,
     )
 
-    HOSTED_OPENAI_API_BASE: str | None = Field(
+    HOSTED_OPENAI_API_BASE: Optional[str] = Field(
         description="Base URL for hosted OpenAI API",
         default=None,
     )
 
-    HOSTED_OPENAI_API_ORGANIZATION: str | None = Field(
+    HOSTED_OPENAI_API_ORGANIZATION: Optional[str] = Field(
         description="Organization ID for hosted OpenAI service",
         default=None,
     )
@@ -108,12 +110,12 @@ class HostedAzureOpenAiConfig(BaseSettings):
         default=False,
     )
 
-    HOSTED_AZURE_OPENAI_API_KEY: str | None = Field(
+    HOSTED_AZURE_OPENAI_API_KEY: Optional[str] = Field(
         description="API key for hosted Azure OpenAI service",
         default=None,
     )
 
-    HOSTED_AZURE_OPENAI_API_BASE: str | None = Field(
+    HOSTED_AZURE_OPENAI_API_BASE: Optional[str] = Field(
         description="Base URL for hosted Azure OpenAI API",
         default=None,
     )
@@ -129,12 +131,12 @@ class HostedAnthropicConfig(BaseSettings):
     Configuration for hosted Anthropic service
     """
 
-    HOSTED_ANTHROPIC_API_BASE: str | None = Field(
+    HOSTED_ANTHROPIC_API_BASE: Optional[str] = Field(
         description="Base URL for hosted Anthropic API",
         default=None,
     )
 
-    HOSTED_ANTHROPIC_API_KEY: str | None = Field(
+    HOSTED_ANTHROPIC_API_KEY: Optional[str] = Field(
         description="API key for hosted Anthropic service",
         default=None,
     )
@@ -220,28 +222,11 @@ 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/configs/middleware/__init__.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Literal
+from typing import Any, Literal, Optional
 from urllib.parse import parse_qsl, quote_plus
 
 from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveFloat, PositiveInt, computed_field
@@ -78,18 +78,18 @@ class StorageConfig(BaseSettings):
 
 
 class VectorStoreConfig(BaseSettings):
-    VECTOR_STORE: str | None = Field(
+    VECTOR_STORE: Optional[str] = Field(
         description="Type of vector store to use for efficient similarity search."
         " Set to None if not using a vector store.",
         default=None,
     )
 
-    VECTOR_STORE_WHITELIST_ENABLE: bool | None = Field(
+    VECTOR_STORE_WHITELIST_ENABLE: Optional[bool] = Field(
         description="Enable whitelist for vector store.",
         default=False,
     )
 
-    VECTOR_INDEX_NAME_PREFIX: str | None = Field(
+    VECTOR_INDEX_NAME_PREFIX: Optional[str] = Field(
         description="Prefix used to create collection name in vector database",
         default="Vector_index",
     )
@@ -187,11 +187,6 @@ class DatabaseConfig(BaseSettings):
         default=False,
     )
 
-    SQLALCHEMY_POOL_TIMEOUT: NonNegativeInt = Field(
-        description="Number of seconds to wait for a connection from the pool before raising a timeout error.",
-        default=30,
-    )
-
     RETRIEVAL_SERVICE_EXECUTORS: NonNegativeInt = Field(
         description="Number of processes for the retrieval service, default to CPU cores.",
         default=os.cpu_count() or 1,
@@ -221,7 +216,6 @@ class DatabaseConfig(BaseSettings):
             "connect_args": connect_args,
             "pool_use_lifo": self.SQLALCHEMY_POOL_USE_LIFO,
             "pool_reset_on_return": None,
-            "pool_timeout": self.SQLALCHEMY_POOL_TIMEOUT,
         }
 
 
@@ -231,26 +225,26 @@ class CeleryConfig(DatabaseConfig):
         default="redis",
     )
 
-    CELERY_BROKER_URL: str | None = Field(
+    CELERY_BROKER_URL: Optional[str] = Field(
         description="URL of the message broker for Celery tasks.",
         default=None,
     )
 
-    CELERY_USE_SENTINEL: bool | None = Field(
+    CELERY_USE_SENTINEL: Optional[bool] = Field(
         description="Whether to use Redis Sentinel for high availability.",
         default=False,
     )
 
-    CELERY_SENTINEL_MASTER_NAME: str | None = Field(
+    CELERY_SENTINEL_MASTER_NAME: Optional[str] = Field(
         description="Name of the Redis Sentinel master.",
         default=None,
     )
 
-    CELERY_SENTINEL_PASSWORD: str | None = Field(
+    CELERY_SENTINEL_PASSWORD: Optional[str] = Field(
         description="Password of the Redis Sentinel master.",
         default=None,
     )
-    CELERY_SENTINEL_SOCKET_TIMEOUT: PositiveFloat | None = Field(
+    CELERY_SENTINEL_SOCKET_TIMEOUT: Optional[PositiveFloat] = Field(
         description="Timeout for Redis Sentinel socket operations in seconds.",
         default=0.1,
     )
@@ -274,12 +268,12 @@ class InternalTestConfig(BaseSettings):
     Configuration settings for Internal Test
     """
 
-    AWS_SECRET_ACCESS_KEY: str | None = Field(
+    AWS_SECRET_ACCESS_KEY: Optional[str] = Field(
         description="Internal test AWS secret access key",
         default=None,
     )
 
-    AWS_ACCESS_KEY_ID: str | None = Field(
+    AWS_ACCESS_KEY_ID: Optional[str] = Field(
         description="Internal test AWS access key ID",
         default=None,
     )
@@ -290,15 +284,15 @@ class DatasetQueueMonitorConfig(BaseSettings):
     Configuration settings for Dataset Queue Monitor
     """
 
-    QUEUE_MONITOR_THRESHOLD: NonNegativeInt | None = Field(
+    QUEUE_MONITOR_THRESHOLD: Optional[NonNegativeInt] = Field(
         description="Threshold for dataset queue monitor",
         default=200,
     )
-    QUEUE_MONITOR_ALERT_EMAILS: str | None = Field(
+    QUEUE_MONITOR_ALERT_EMAILS: Optional[str] = Field(
         description="Emails for dataset queue monitor alert, separated by commas",
         default=None,
     )
-    QUEUE_MONITOR_INTERVAL: NonNegativeFloat | None = Field(
+    QUEUE_MONITOR_INTERVAL: Optional[NonNegativeFloat] = Field(
         description="Interval for dataset queue monitor in minutes",
         default=30,
     )

api/configs/middleware/cache/redis_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, NonNegativeInt, PositiveFloat, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -17,12 +19,12 @@ class RedisConfig(BaseSettings):
         default=6379,
     )
 
-    REDIS_USERNAME: str | None = Field(
+    REDIS_USERNAME: Optional[str] = Field(
         description="Username for Redis authentication (if required)",
         default=None,
     )
 
-    REDIS_PASSWORD: str | None = Field(
+    REDIS_PASSWORD: Optional[str] = Field(
         description="Password for Redis authentication (if required)",
         default=None,
     )
@@ -42,47 +44,47 @@ class RedisConfig(BaseSettings):
         default="CERT_NONE",
     )
 
-    REDIS_SSL_CA_CERTS: str | None = Field(
+    REDIS_SSL_CA_CERTS: Optional[str] = Field(
         description="Path to the CA certificate file for SSL verification",
         default=None,
     )
 
-    REDIS_SSL_CERTFILE: str | None = Field(
+    REDIS_SSL_CERTFILE: Optional[str] = Field(
         description="Path to the client certificate file for SSL authentication",
         default=None,
     )
 
-    REDIS_SSL_KEYFILE: str | None = Field(
+    REDIS_SSL_KEYFILE: Optional[str] = Field(
         description="Path to the client private key file for SSL authentication",
         default=None,
     )
 
-    REDIS_USE_SENTINEL: bool | None = Field(
+    REDIS_USE_SENTINEL: Optional[bool] = Field(
         description="Enable Redis Sentinel mode for high availability",
         default=False,
     )
 
-    REDIS_SENTINELS: str | None = Field(
+    REDIS_SENTINELS: Optional[str] = Field(
         description="Comma-separated list of Redis Sentinel nodes (host:port)",
         default=None,
     )
 
-    REDIS_SENTINEL_SERVICE_NAME: str | None = Field(
+    REDIS_SENTINEL_SERVICE_NAME: Optional[str] = Field(
         description="Name of the Redis Sentinel service to monitor",
         default=None,
     )
 
-    REDIS_SENTINEL_USERNAME: str | None = Field(
+    REDIS_SENTINEL_USERNAME: Optional[str] = Field(
         description="Username for Redis Sentinel authentication (if required)",
         default=None,
     )
 
-    REDIS_SENTINEL_PASSWORD: str | None = Field(
+    REDIS_SENTINEL_PASSWORD: Optional[str] = Field(
         description="Password for Redis Sentinel authentication (if required)",
         default=None,
     )
 
-    REDIS_SENTINEL_SOCKET_TIMEOUT: PositiveFloat | None = Field(
+    REDIS_SENTINEL_SOCKET_TIMEOUT: Optional[PositiveFloat] = Field(
         description="Socket timeout in seconds for Redis Sentinel connections",
         default=0.1,
     )
@@ -92,12 +94,12 @@ class RedisConfig(BaseSettings):
         default=False,
     )
 
-    REDIS_CLUSTERS: str | None = Field(
+    REDIS_CLUSTERS: Optional[str] = Field(
         description="Comma-separated list of Redis Clusters nodes (host:port)",
         default=None,
     )
 
-    REDIS_CLUSTERS_PASSWORD: str | None = Field(
+    REDIS_CLUSTERS_PASSWORD: Optional[str] = Field(
         description="Password for Redis Clusters authentication (if required)",
         default=None,
     )

api/configs/middleware/storage/aliyun_oss_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,37 +9,37 @@ class AliyunOSSStorageConfig(BaseSettings):
     Configuration settings for Aliyun Object Storage Service (OSS)
     """
 
-    ALIYUN_OSS_BUCKET_NAME: str | None = Field(
+    ALIYUN_OSS_BUCKET_NAME: Optional[str] = Field(
         description="Name of the Aliyun OSS bucket to store and retrieve objects",
         default=None,
     )
 
-    ALIYUN_OSS_ACCESS_KEY: str | None = Field(
+    ALIYUN_OSS_ACCESS_KEY: Optional[str] = Field(
         description="Access key ID for authenticating with Aliyun OSS",
         default=None,
     )
 
-    ALIYUN_OSS_SECRET_KEY: str | None = Field(
+    ALIYUN_OSS_SECRET_KEY: Optional[str] = Field(
         description="Secret access key for authenticating with Aliyun OSS",
         default=None,
     )
 
-    ALIYUN_OSS_ENDPOINT: str | None = Field(
+    ALIYUN_OSS_ENDPOINT: Optional[str] = Field(
         description="URL of the Aliyun OSS endpoint for your chosen region",
         default=None,
     )
 
-    ALIYUN_OSS_REGION: str | None = Field(
+    ALIYUN_OSS_REGION: Optional[str] = Field(
         description="Aliyun OSS region where your bucket is located (e.g., 'oss-cn-hangzhou')",
         default=None,
     )
 
-    ALIYUN_OSS_AUTH_VERSION: str | None = Field(
+    ALIYUN_OSS_AUTH_VERSION: Optional[str] = Field(
         description="Version of the authentication protocol to use with Aliyun OSS (e.g., 'v4')",
         default=None,
     )
 
-    ALIYUN_OSS_PATH: str | None = Field(
+    ALIYUN_OSS_PATH: Optional[str] = Field(
         description="Base path within the bucket to store objects (e.g., 'my-app-data/')",
         default=None,
     )

api/configs/middleware/storage/amazon_s3_storage_config.py

@@ -1,4 +1,4 @@
-from typing import Literal
+from typing import Literal, Optional
 
 from pydantic import Field
 from pydantic_settings import BaseSettings
@@ -9,27 +9,27 @@ class S3StorageConfig(BaseSettings):
     Configuration settings for S3-compatible object storage
     """
 
-    S3_ENDPOINT: str | None = Field(
+    S3_ENDPOINT: Optional[str] = Field(
         description="URL of the S3-compatible storage endpoint (e.g., 'https://s3.amazonaws.com')",
         default=None,
     )
 
-    S3_REGION: str | None = Field(
+    S3_REGION: Optional[str] = Field(
         description="Region where the S3 bucket is located (e.g., 'us-east-1')",
         default=None,
     )
 
-    S3_BUCKET_NAME: str | None = Field(
+    S3_BUCKET_NAME: Optional[str] = Field(
         description="Name of the S3 bucket to store and retrieve objects",
         default=None,
     )
 
-    S3_ACCESS_KEY: str | None = Field(
+    S3_ACCESS_KEY: Optional[str] = Field(
         description="Access key ID for authenticating with the S3 service",
         default=None,
     )
 
-    S3_SECRET_KEY: str | None = Field(
+    S3_SECRET_KEY: Optional[str] = Field(
         description="Secret access key for authenticating with the S3 service",
         default=None,
     )

api/configs/middleware/storage/azure_blob_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,22 +9,22 @@ class AzureBlobStorageConfig(BaseSettings):
     Configuration settings for Azure Blob Storage
     """
 
-    AZURE_BLOB_ACCOUNT_NAME: str | None = Field(
+    AZURE_BLOB_ACCOUNT_NAME: Optional[str] = Field(
         description="Name of the Azure Storage account (e.g., 'mystorageaccount')",
         default=None,
     )
 
-    AZURE_BLOB_ACCOUNT_KEY: str | None = Field(
+    AZURE_BLOB_ACCOUNT_KEY: Optional[str] = Field(
         description="Access key for authenticating with the Azure Storage account",
         default=None,
     )
 
-    AZURE_BLOB_CONTAINER_NAME: str | None = Field(
+    AZURE_BLOB_CONTAINER_NAME: Optional[str] = Field(
         description="Name of the Azure Blob container to store and retrieve objects",
         default=None,
     )
 
-    AZURE_BLOB_ACCOUNT_URL: str | None = Field(
+    AZURE_BLOB_ACCOUNT_URL: Optional[str] = Field(
         description="URL of the Azure Blob storage endpoint (e.g., 'https://mystorageaccount.blob.core.windows.net')",
         default=None,
     )

api/configs/middleware/storage/baidu_obs_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,22 +9,22 @@ class BaiduOBSStorageConfig(BaseSettings):
     Configuration settings for Baidu Object Storage Service (OBS)
     """
 
-    BAIDU_OBS_BUCKET_NAME: str | None = Field(
+    BAIDU_OBS_BUCKET_NAME: Optional[str] = Field(
         description="Name of the Baidu OBS bucket to store and retrieve objects (e.g., 'my-obs-bucket')",
         default=None,
     )
 
-    BAIDU_OBS_ACCESS_KEY: str | None = Field(
+    BAIDU_OBS_ACCESS_KEY: Optional[str] = Field(
         description="Access Key ID for authenticating with Baidu OBS",
         default=None,
     )
 
-    BAIDU_OBS_SECRET_KEY: str | None = Field(
+    BAIDU_OBS_SECRET_KEY: Optional[str] = Field(
         description="Secret Access Key for authenticating with Baidu OBS",
         default=None,
     )
 
-    BAIDU_OBS_ENDPOINT: str | None = Field(
+    BAIDU_OBS_ENDPOINT: Optional[str] = Field(
         description="URL of the Baidu OSS endpoint for your chosen region (e.g., 'https://.bj.bcebos.com')",
         default=None,
     )

api/configs/middleware/storage/clickzetta_volume_storage_config.py

@@ -1,5 +1,7 @@
 """ClickZetta Volume Storage Configuration"""
 
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,17 +9,17 @@ from pydantic_settings import BaseSettings
 class ClickZettaVolumeStorageConfig(BaseSettings):
     """Configuration for ClickZetta Volume storage."""
 
-    CLICKZETTA_VOLUME_USERNAME: str | None = Field(
+    CLICKZETTA_VOLUME_USERNAME: Optional[str] = Field(
         description="Username for ClickZetta Volume authentication",
         default=None,
     )
 
-    CLICKZETTA_VOLUME_PASSWORD: str | None = Field(
+    CLICKZETTA_VOLUME_PASSWORD: Optional[str] = Field(
         description="Password for ClickZetta Volume authentication",
         default=None,
     )
 
-    CLICKZETTA_VOLUME_INSTANCE: str | None = Field(
+    CLICKZETTA_VOLUME_INSTANCE: Optional[str] = Field(
         description="ClickZetta instance identifier",
         default=None,
     )
@@ -47,7 +49,7 @@ class ClickZettaVolumeStorageConfig(BaseSettings):
         default="user",
     )
 
-    CLICKZETTA_VOLUME_NAME: str | None = Field(
+    CLICKZETTA_VOLUME_NAME: Optional[str] = Field(
         description="ClickZetta volume name for external volumes",
         default=None,
     )

api/configs/middleware/storage/google_cloud_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,12 +9,12 @@ class GoogleCloudStorageConfig(BaseSettings):
     Configuration settings for Google Cloud Storage
     """
 
-    GOOGLE_STORAGE_BUCKET_NAME: str | None = Field(
+    GOOGLE_STORAGE_BUCKET_NAME: Optional[str] = Field(
         description="Name of the Google Cloud Storage bucket to store and retrieve objects (e.g., 'my-gcs-bucket')",
         default=None,
     )
 
-    GOOGLE_STORAGE_SERVICE_ACCOUNT_JSON_BASE64: str | None = Field(
+    GOOGLE_STORAGE_SERVICE_ACCOUNT_JSON_BASE64: Optional[str] = Field(
         description="Base64-encoded JSON key file for Google Cloud service account authentication",
         default=None,
     )

api/configs/middleware/storage/huawei_obs_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,22 +9,22 @@ class HuaweiCloudOBSStorageConfig(BaseSettings):
     Configuration settings for Huawei Cloud Object Storage Service (OBS)
     """
 
-    HUAWEI_OBS_BUCKET_NAME: str | None = Field(
+    HUAWEI_OBS_BUCKET_NAME: Optional[str] = Field(
         description="Name of the Huawei Cloud OBS bucket to store and retrieve objects (e.g., 'my-obs-bucket')",
         default=None,
     )
 
-    HUAWEI_OBS_ACCESS_KEY: str | None = Field(
+    HUAWEI_OBS_ACCESS_KEY: Optional[str] = Field(
         description="Access Key ID for authenticating with Huawei Cloud OBS",
         default=None,
     )
 
-    HUAWEI_OBS_SECRET_KEY: str | None = Field(
+    HUAWEI_OBS_SECRET_KEY: Optional[str] = Field(
         description="Secret Access Key for authenticating with Huawei Cloud OBS",
         default=None,
     )
 
-    HUAWEI_OBS_SERVER: str | None = Field(
+    HUAWEI_OBS_SERVER: Optional[str] = Field(
         description="Endpoint URL for Huawei Cloud OBS (e.g., 'https://obs.cn-north-4.myhuaweicloud.com')",
         default=None,
     )

api/configs/middleware/storage/oci_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,27 +9,27 @@ class OCIStorageConfig(BaseSettings):
     Configuration settings for Oracle Cloud Infrastructure (OCI) Object Storage
     """
 
-    OCI_ENDPOINT: str | None = Field(
+    OCI_ENDPOINT: Optional[str] = Field(
         description="URL of the OCI Object Storage endpoint (e.g., 'https://objectstorage.us-phoenix-1.oraclecloud.com')",
         default=None,
     )
 
-    OCI_REGION: str | None = Field(
+    OCI_REGION: Optional[str] = Field(
         description="OCI region where the bucket is located (e.g., 'us-phoenix-1')",
         default=None,
     )
 
-    OCI_BUCKET_NAME: str | None = Field(
+    OCI_BUCKET_NAME: Optional[str] = Field(
         description="Name of the OCI Object Storage bucket to store and retrieve objects (e.g., 'my-oci-bucket')",
         default=None,
     )
 
-    OCI_ACCESS_KEY: str | None = Field(
+    OCI_ACCESS_KEY: Optional[str] = Field(
         description="Access key (also known as API key) for authenticating with OCI Object Storage",
         default=None,
     )
 
-    OCI_SECRET_KEY: str | None = Field(
+    OCI_SECRET_KEY: Optional[str] = Field(
         description="Secret key associated with the access key for authenticating with OCI Object Storage",
         default=None,
     )

api/configs/middleware/storage/supabase_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,17 +9,17 @@ class SupabaseStorageConfig(BaseSettings):
     Configuration settings for Supabase Object Storage Service
     """
 
-    SUPABASE_BUCKET_NAME: str | None = Field(
+    SUPABASE_BUCKET_NAME: Optional[str] = Field(
         description="Name of the Supabase bucket to store and retrieve objects (e.g., 'dify-bucket')",
         default=None,
     )
 
-    SUPABASE_API_KEY: str | None = Field(
+    SUPABASE_API_KEY: Optional[str] = Field(
         description="API KEY for authenticating with Supabase",
         default=None,
     )
 
-    SUPABASE_URL: str | None = Field(
+    SUPABASE_URL: Optional[str] = Field(
         description="URL of the Supabase",
         default=None,
     )

api/configs/middleware/storage/tencent_cos_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,27 +9,27 @@ class TencentCloudCOSStorageConfig(BaseSettings):
     Configuration settings for Tencent Cloud Object Storage (COS)
     """
 
-    TENCENT_COS_BUCKET_NAME: str | None = Field(
+    TENCENT_COS_BUCKET_NAME: Optional[str] = Field(
         description="Name of the Tencent Cloud COS bucket to store and retrieve objects",
         default=None,
     )
 
-    TENCENT_COS_REGION: str | None = Field(
+    TENCENT_COS_REGION: Optional[str] = Field(
         description="Tencent Cloud region where the COS bucket is located (e.g., 'ap-guangzhou')",
         default=None,
     )
 
-    TENCENT_COS_SECRET_ID: str | None = Field(
+    TENCENT_COS_SECRET_ID: Optional[str] = Field(
         description="SecretId for authenticating with Tencent Cloud COS (part of API credentials)",
         default=None,
     )
 
-    TENCENT_COS_SECRET_KEY: str | None = Field(
+    TENCENT_COS_SECRET_KEY: Optional[str] = Field(
         description="SecretKey for authenticating with Tencent Cloud COS (part of API credentials)",
         default=None,
     )
 
-    TENCENT_COS_SCHEME: str | None = Field(
+    TENCENT_COS_SCHEME: Optional[str] = Field(
         description="Protocol scheme for COS requests: 'https' (recommended) or 'http'",
         default=None,
     )

api/configs/middleware/storage/volcengine_tos_storage_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,27 +9,27 @@ class VolcengineTOSStorageConfig(BaseSettings):
     Configuration settings for Volcengine Tinder Object Storage (TOS)
     """
 
-    VOLCENGINE_TOS_BUCKET_NAME: str | None = Field(
+    VOLCENGINE_TOS_BUCKET_NAME: Optional[str] = Field(
         description="Name of the Volcengine TOS bucket to store and retrieve objects (e.g., 'my-tos-bucket')",
         default=None,
     )
 
-    VOLCENGINE_TOS_ACCESS_KEY: str | None = Field(
+    VOLCENGINE_TOS_ACCESS_KEY: Optional[str] = Field(
         description="Access Key ID for authenticating with Volcengine TOS",
         default=None,
     )
 
-    VOLCENGINE_TOS_SECRET_KEY: str | None = Field(
+    VOLCENGINE_TOS_SECRET_KEY: Optional[str] = Field(
         description="Secret Access Key for authenticating with Volcengine TOS",
         default=None,
     )
 
-    VOLCENGINE_TOS_ENDPOINT: str | None = Field(
+    VOLCENGINE_TOS_ENDPOINT: Optional[str] = Field(
         description="URL of the Volcengine TOS endpoint (e.g., 'https://tos-cn-beijing.volces.com')",
         default=None,
     )
 
-    VOLCENGINE_TOS_REGION: str | None = Field(
+    VOLCENGINE_TOS_REGION: Optional[str] = Field(
         description="Volcengine region where the TOS bucket is located (e.g., 'cn-beijing')",
         default=None,
     )

api/configs/middleware/vdb/analyticdb_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -9,37 +11,37 @@ class AnalyticdbConfig(BaseSettings):
     https://www.alibabacloud.com/help/en/analyticdb-for-postgresql/getting-started/create-an-instance-instances-with-vector-engine-optimization-enabled
     """
 
-    ANALYTICDB_KEY_ID: str | None = Field(
+    ANALYTICDB_KEY_ID: Optional[str] = Field(
         default=None, description="The Access Key ID provided by Alibaba Cloud for API authentication."
     )
-    ANALYTICDB_KEY_SECRET: str | None = Field(
+    ANALYTICDB_KEY_SECRET: Optional[str] = Field(
         default=None, description="The Secret Access Key corresponding to the Access Key ID for secure API access."
     )
-    ANALYTICDB_REGION_ID: str | None = Field(
+    ANALYTICDB_REGION_ID: Optional[str] = Field(
         default=None,
         description="The region where the AnalyticDB instance is deployed (e.g., 'cn-hangzhou', 'ap-southeast-1').",
     )
-    ANALYTICDB_INSTANCE_ID: str | None = Field(
+    ANALYTICDB_INSTANCE_ID: Optional[str] = Field(
         default=None,
         description="The unique identifier of the AnalyticDB instance you want to connect to.",
     )
-    ANALYTICDB_ACCOUNT: str | None = Field(
+    ANALYTICDB_ACCOUNT: Optional[str] = Field(
         default=None,
         description="The account name used to log in to the AnalyticDB instance"
         " (usually the initial account created with the instance).",
     )
-    ANALYTICDB_PASSWORD: str | None = Field(
+    ANALYTICDB_PASSWORD: Optional[str] = Field(
         default=None, description="The password associated with the AnalyticDB account for database authentication."
     )
-    ANALYTICDB_NAMESPACE: str | None = Field(
+    ANALYTICDB_NAMESPACE: Optional[str] = Field(
         default=None, description="The namespace within AnalyticDB for schema isolation (if using namespace feature)."
     )
-    ANALYTICDB_NAMESPACE_PASSWORD: str | None = Field(
+    ANALYTICDB_NAMESPACE_PASSWORD: Optional[str] = Field(
         default=None,
         description="The password for accessing the specified namespace within the AnalyticDB instance"
         " (if namespace feature is enabled).",
     )
-    ANALYTICDB_HOST: str | None = Field(
+    ANALYTICDB_HOST: Optional[str] = Field(
         default=None, description="The host of the AnalyticDB instance you want to connect to."
     )
     ANALYTICDB_PORT: PositiveInt = Field(

api/configs/middleware/vdb/baidu_vector_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, NonNegativeInt, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,7 +9,7 @@ class BaiduVectorDBConfig(BaseSettings):
     Configuration settings for Baidu Vector Database
     """
 
-    BAIDU_VECTOR_DB_ENDPOINT: str | None = Field(
+    BAIDU_VECTOR_DB_ENDPOINT: Optional[str] = Field(
         description="URL of the Baidu Vector Database service (e.g., 'http://vdb.bj.baidubce.com')",
         default=None,
     )
@@ -17,17 +19,17 @@ class BaiduVectorDBConfig(BaseSettings):
         default=30000,
     )
 
-    BAIDU_VECTOR_DB_ACCOUNT: str | None = Field(
+    BAIDU_VECTOR_DB_ACCOUNT: Optional[str] = Field(
         description="Account for authenticating with the Baidu Vector Database",
         default=None,
     )
 
-    BAIDU_VECTOR_DB_API_KEY: str | None = Field(
+    BAIDU_VECTOR_DB_API_KEY: Optional[str] = Field(
         description="API key for authenticating with the Baidu Vector Database service",
         default=None,
     )
 
-    BAIDU_VECTOR_DB_DATABASE: str | None = Field(
+    BAIDU_VECTOR_DB_DATABASE: Optional[str] = Field(
         description="Name of the specific Baidu Vector Database to connect to",
         default=None,
     )

api/configs/middleware/vdb/chroma_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,7 +9,7 @@ class ChromaConfig(BaseSettings):
     Configuration settings for Chroma vector database
     """
 
-    CHROMA_HOST: str | None = Field(
+    CHROMA_HOST: Optional[str] = Field(
         description="Hostname or IP address of the Chroma server (e.g., 'localhost' or '192.168.1.100')",
         default=None,
     )
@@ -17,22 +19,22 @@ class ChromaConfig(BaseSettings):
         default=8000,
     )
 
-    CHROMA_TENANT: str | None = Field(
+    CHROMA_TENANT: Optional[str] = Field(
         description="Tenant identifier for multi-tenancy support in Chroma",
         default=None,
     )
 
-    CHROMA_DATABASE: str | None = Field(
+    CHROMA_DATABASE: Optional[str] = Field(
         description="Name of the Chroma database to connect to",
         default=None,
     )
 
-    CHROMA_AUTH_PROVIDER: str | None = Field(
+    CHROMA_AUTH_PROVIDER: Optional[str] = Field(
         description="Authentication provider for Chroma (e.g., 'basic', 'token', or a custom provider)",
         default=None,
     )
 
-    CHROMA_AUTH_CREDENTIALS: str | None = Field(
+    CHROMA_AUTH_CREDENTIALS: Optional[str] = Field(
         description="Authentication credentials for Chroma (format depends on the auth provider)",
         default=None,
     )

api/configs/middleware/vdb/clickzetta_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,62 +9,62 @@ class ClickzettaConfig(BaseSettings):
     Clickzetta Lakehouse vector database configuration
     """
 
-    CLICKZETTA_USERNAME: str | None = Field(
+    CLICKZETTA_USERNAME: Optional[str] = Field(
         description="Username for authenticating with Clickzetta Lakehouse",
         default=None,
     )
 
-    CLICKZETTA_PASSWORD: str | None = Field(
+    CLICKZETTA_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with Clickzetta Lakehouse",
         default=None,
     )
 
-    CLICKZETTA_INSTANCE: str | None = Field(
+    CLICKZETTA_INSTANCE: Optional[str] = Field(
         description="Clickzetta Lakehouse instance ID",
         default=None,
     )
 
-    CLICKZETTA_SERVICE: str | None = Field(
+    CLICKZETTA_SERVICE: Optional[str] = Field(
         description="Clickzetta API service endpoint (e.g., 'api.clickzetta.com')",
         default="api.clickzetta.com",
     )
 
-    CLICKZETTA_WORKSPACE: str | None = Field(
+    CLICKZETTA_WORKSPACE: Optional[str] = Field(
         description="Clickzetta workspace name",
         default="default",
     )
 
-    CLICKZETTA_VCLUSTER: str | None = Field(
+    CLICKZETTA_VCLUSTER: Optional[str] = Field(
         description="Clickzetta virtual cluster name",
         default="default_ap",
     )
 
-    CLICKZETTA_SCHEMA: str | None = Field(
+    CLICKZETTA_SCHEMA: Optional[str] = Field(
         description="Database schema name in Clickzetta",
         default="public",
     )
 
-    CLICKZETTA_BATCH_SIZE: int | None = Field(
+    CLICKZETTA_BATCH_SIZE: Optional[int] = Field(
         description="Batch size for bulk insert operations",
         default=100,
     )
 
-    CLICKZETTA_ENABLE_INVERTED_INDEX: bool | None = Field(
+    CLICKZETTA_ENABLE_INVERTED_INDEX: Optional[bool] = Field(
         description="Enable inverted index for full-text search capabilities",
         default=True,
     )
 
-    CLICKZETTA_ANALYZER_TYPE: str | None = Field(
+    CLICKZETTA_ANALYZER_TYPE: Optional[str] = Field(
         description="Analyzer type for full-text search: keyword, english, chinese, unicode",
         default="chinese",
     )
 
-    CLICKZETTA_ANALYZER_MODE: str | None = Field(
+    CLICKZETTA_ANALYZER_MODE: Optional[str] = Field(
         description="Analyzer mode for tokenization: max_word (fine-grained) or smart (intelligent)",
         default="smart",
     )
 
-    CLICKZETTA_VECTOR_DISTANCE_FUNCTION: str | None = Field(
+    CLICKZETTA_VECTOR_DISTANCE_FUNCTION: Optional[str] = Field(
         description="Distance function for vector similarity: l2_distance or cosine_distance",
         default="cosine_distance",
     )

api/configs/middleware/vdb/couchbase_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,27 +9,27 @@ class CouchbaseConfig(BaseSettings):
     Couchbase configs
     """
 
-    COUCHBASE_CONNECTION_STRING: str | None = Field(
+    COUCHBASE_CONNECTION_STRING: Optional[str] = Field(
         description="COUCHBASE connection string",
         default=None,
     )
 
-    COUCHBASE_USER: str | None = Field(
+    COUCHBASE_USER: Optional[str] = Field(
         description="COUCHBASE user",
         default=None,
     )
 
-    COUCHBASE_PASSWORD: str | None = Field(
+    COUCHBASE_PASSWORD: Optional[str] = Field(
         description="COUCHBASE password",
         default=None,
     )
 
-    COUCHBASE_BUCKET_NAME: str | None = Field(
+    COUCHBASE_BUCKET_NAME: Optional[str] = Field(
         description="COUCHBASE bucket name",
         default=None,
     )
 
-    COUCHBASE_SCOPE_NAME: str | None = Field(
+    COUCHBASE_SCOPE_NAME: Optional[str] = Field(
         description="COUCHBASE scope name",
         default=None,
     )

api/configs/middleware/vdb/elasticsearch_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt, model_validator
 from pydantic_settings import BaseSettings
 
@@ -8,7 +10,7 @@ class ElasticsearchConfig(BaseSettings):
     Can load from environment variables or .env files.
     """
 
-    ELASTICSEARCH_HOST: str | None = Field(
+    ELASTICSEARCH_HOST: Optional[str] = Field(
         description="Hostname or IP address of the Elasticsearch server (e.g., 'localhost' or '192.168.1.100')",
         default="127.0.0.1",
     )
@@ -18,28 +20,30 @@ class ElasticsearchConfig(BaseSettings):
         default=9200,
     )
 
-    ELASTICSEARCH_USERNAME: str | None = Field(
+    ELASTICSEARCH_USERNAME: Optional[str] = Field(
         description="Username for authenticating with Elasticsearch (default is 'elastic')",
         default="elastic",
     )
 
-    ELASTICSEARCH_PASSWORD: str | None = Field(
+    ELASTICSEARCH_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with Elasticsearch (default is 'elastic')",
         default="elastic",
     )
 
     # Elastic Cloud (optional)
-    ELASTICSEARCH_USE_CLOUD: bool | None = Field(
+    ELASTICSEARCH_USE_CLOUD: Optional[bool] = Field(
         description="Set to True to use Elastic Cloud instead of self-hosted Elasticsearch", default=False
     )
-    ELASTICSEARCH_CLOUD_URL: str | None = Field(
+    ELASTICSEARCH_CLOUD_URL: Optional[str] = Field(
         description="Full URL for Elastic Cloud deployment (e.g., 'https://example.es.region.aws.found.io:443')",
         default=None,
     )
-    ELASTICSEARCH_API_KEY: str | None = Field(description="API key for authenticating with Elastic Cloud", default=None)
+    ELASTICSEARCH_API_KEY: Optional[str] = Field(
+        description="API key for authenticating with Elastic Cloud", default=None
+    )
 
     # Common options
-    ELASTICSEARCH_CA_CERTS: str | None = Field(
+    ELASTICSEARCH_CA_CERTS: Optional[str] = Field(
         description="Path to CA certificate file for SSL verification", default=None
     )
     ELASTICSEARCH_VERIFY_CERTS: bool = Field(

api/configs/middleware/vdb/huawei_cloud_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,17 +9,17 @@ class HuaweiCloudConfig(BaseSettings):
     Configuration settings for Huawei cloud search service
     """
 
-    HUAWEI_CLOUD_HOSTS: str | None = Field(
+    HUAWEI_CLOUD_HOSTS: Optional[str] = Field(
         description="Hostname or IP address of the Huawei cloud search service instance",
         default=None,
     )
 
-    HUAWEI_CLOUD_USER: str | None = Field(
+    HUAWEI_CLOUD_USER: Optional[str] = Field(
         description="Username for authenticating with Huawei cloud search service",
         default=None,
     )
 
-    HUAWEI_CLOUD_PASSWORD: str | None = Field(
+    HUAWEI_CLOUD_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with Huawei cloud search service",
         default=None,
     )

api/configs/middleware/vdb/lindorm_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,27 +9,27 @@ class LindormConfig(BaseSettings):
     Lindorm configs
     """
 
-    LINDORM_URL: str | None = Field(
+    LINDORM_URL: Optional[str] = Field(
         description="Lindorm url",
         default=None,
     )
-    LINDORM_USERNAME: str | None = Field(
+    LINDORM_USERNAME: Optional[str] = Field(
         description="Lindorm user",
         default=None,
     )
-    LINDORM_PASSWORD: str | None = Field(
+    LINDORM_PASSWORD: Optional[str] = Field(
         description="Lindorm password",
         default=None,
     )
-    LINDORM_INDEX_TYPE: str | None = Field(
+    DEFAULT_INDEX_TYPE: Optional[str] = Field(
         description="Lindorm Vector Index Type, hnsw or flat is available in dify",
         default="hnsw",
     )
-    LINDORM_DISTANCE_TYPE: str | None = Field(
+    DEFAULT_DISTANCE_TYPE: Optional[str] = Field(
         description="Vector Distance Type, support l2, cosinesimil, innerproduct", default="l2"
     )
-    LINDORM_USING_UGC: bool | None = Field(
-        description="Using UGC index will store indexes with the same IndexType/Dimension in a single big index.",
-        default=True,
+    USING_UGC_INDEX: Optional[bool] = Field(
+        description="Using UGC index will store the same type of Index in a single index but can retrieve separately.",
+        default=False,
     )
-    LINDORM_QUERY_TIMEOUT: float | None = Field(description="The lindorm search request timeout (s)", default=2.0)
+    LINDORM_QUERY_TIMEOUT: Optional[float] = Field(description="The lindorm search request timeout (s)", default=2.0)

api/configs/middleware/vdb/milvus_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,22 +9,22 @@ class MilvusConfig(BaseSettings):
     Configuration settings for Milvus vector database
     """
 
-    MILVUS_URI: str | None = Field(
+    MILVUS_URI: Optional[str] = Field(
         description="URI for connecting to the Milvus server (e.g., 'http://localhost:19530' or 'https://milvus-instance.example.com:19530')",
         default="http://127.0.0.1:19530",
     )
 
-    MILVUS_TOKEN: str | None = Field(
+    MILVUS_TOKEN: Optional[str] = Field(
         description="Authentication token for Milvus, if token-based authentication is enabled",
         default=None,
     )
 
-    MILVUS_USER: str | None = Field(
+    MILVUS_USER: Optional[str] = Field(
         description="Username for authenticating with Milvus, if username/password authentication is enabled",
         default=None,
     )
 
-    MILVUS_PASSWORD: str | None = Field(
+    MILVUS_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with Milvus, if username/password authentication is enabled",
         default=None,
     )
@@ -38,7 +40,7 @@ class MilvusConfig(BaseSettings):
         default=True,
     )
 
-    MILVUS_ANALYZER_PARAMS: str | None = Field(
+    MILVUS_ANALYZER_PARAMS: Optional[str] = Field(
         description='Milvus text analyzer parameters, e.g., {"type": "chinese"} for Chinese segmentation support.',
         default=None,
     )

api/configs/middleware/vdb/oceanbase_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,27 +9,27 @@ class OceanBaseVectorConfig(BaseSettings):
     Configuration settings for OceanBase Vector database
     """
 
-    OCEANBASE_VECTOR_HOST: str | None = Field(
+    OCEANBASE_VECTOR_HOST: Optional[str] = Field(
         description="Hostname or IP address of the OceanBase Vector server (e.g. 'localhost')",
         default=None,
     )
 
-    OCEANBASE_VECTOR_PORT: PositiveInt | None = Field(
+    OCEANBASE_VECTOR_PORT: Optional[PositiveInt] = Field(
         description="Port number on which the OceanBase Vector server is listening (default is 2881)",
         default=2881,
     )
 
-    OCEANBASE_VECTOR_USER: str | None = Field(
+    OCEANBASE_VECTOR_USER: Optional[str] = Field(
         description="Username for authenticating with the OceanBase Vector database",
         default=None,
     )
 
-    OCEANBASE_VECTOR_PASSWORD: str | None = Field(
+    OCEANBASE_VECTOR_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the OceanBase Vector database",
         default=None,
     )
 
-    OCEANBASE_VECTOR_DATABASE: str | None = Field(
+    OCEANBASE_VECTOR_DATABASE: Optional[str] = Field(
         description="Name of the OceanBase Vector database to connect to",
         default=None,
     )

api/configs/middleware/vdb/opengauss_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,7 +9,7 @@ class OpenGaussConfig(BaseSettings):
     Configuration settings for OpenGauss
     """
 
-    OPENGAUSS_HOST: str | None = Field(
+    OPENGAUSS_HOST: Optional[str] = Field(
         description="Hostname or IP address of the OpenGauss server(e.g., 'localhost')",
         default=None,
     )
@@ -17,17 +19,17 @@ class OpenGaussConfig(BaseSettings):
         default=6600,
     )
 
-    OPENGAUSS_USER: str | None = Field(
+    OPENGAUSS_USER: Optional[str] = Field(
         description="Username for authenticating with the OpenGauss database",
         default=None,
     )
 
-    OPENGAUSS_PASSWORD: str | None = Field(
+    OPENGAUSS_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the OpenGauss database",
         default=None,
     )
 
-    OPENGAUSS_DATABASE: str | None = Field(
+    OPENGAUSS_DATABASE: Optional[str] = Field(
         description="Name of the OpenGauss database to connect to",
         default=None,
     )

api/configs/middleware/vdb/opensearch_config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Literal
+from typing import Literal, Optional
 
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
@@ -18,7 +18,7 @@ class OpenSearchConfig(BaseSettings):
         BASIC = "basic"
         AWS_MANAGED_IAM = "aws_managed_iam"
 
-    OPENSEARCH_HOST: str | None = Field(
+    OPENSEARCH_HOST: Optional[str] = Field(
         description="Hostname or IP address of the OpenSearch server (e.g., 'localhost' or 'opensearch.example.com')",
         default=None,
     )
@@ -43,21 +43,21 @@ class OpenSearchConfig(BaseSettings):
         default=AuthMethod.BASIC,
     )
 
-    OPENSEARCH_USER: str | None = Field(
+    OPENSEARCH_USER: Optional[str] = Field(
         description="Username for authenticating with OpenSearch",
         default=None,
     )
 
-    OPENSEARCH_PASSWORD: str | None = Field(
+    OPENSEARCH_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with OpenSearch",
         default=None,
     )
 
-    OPENSEARCH_AWS_REGION: str | None = Field(
+    OPENSEARCH_AWS_REGION: Optional[str] = Field(
         description="AWS region for OpenSearch (e.g. 'us-west-2')",
         default=None,
     )
 
-    OPENSEARCH_AWS_SERVICE: Literal["es", "aoss"] | None = Field(
+    OPENSEARCH_AWS_SERVICE: Optional[Literal["es", "aoss"]] = Field(
         description="AWS service for OpenSearch (e.g. 'aoss' for OpenSearch Serverless)", default=None
     )

api/configs/middleware/vdb/oracle_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,33 +9,33 @@ class OracleConfig(BaseSettings):
     Configuration settings for Oracle database
     """
 
-    ORACLE_USER: str | None = Field(
+    ORACLE_USER: Optional[str] = Field(
         description="Username for authenticating with the Oracle database",
         default=None,
     )
 
-    ORACLE_PASSWORD: str | None = Field(
+    ORACLE_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the Oracle database",
         default=None,
     )
 
-    ORACLE_DSN: str | None = Field(
+    ORACLE_DSN: Optional[str] = Field(
         description="Oracle database connection string. For traditional database, use format 'host:port/service_name'. "
         "For autonomous database, use the service name from tnsnames.ora in the wallet",
         default=None,
     )
 
-    ORACLE_CONFIG_DIR: str | None = Field(
+    ORACLE_CONFIG_DIR: Optional[str] = Field(
         description="Directory containing the tnsnames.ora configuration file. Only used in thin mode connection",
         default=None,
     )
 
-    ORACLE_WALLET_LOCATION: str | None = Field(
+    ORACLE_WALLET_LOCATION: Optional[str] = Field(
         description="Oracle wallet directory path containing the wallet files for secure connection",
         default=None,
     )
 
-    ORACLE_WALLET_PASSWORD: str | None = Field(
+    ORACLE_WALLET_PASSWORD: Optional[str] = Field(
         description="Password to decrypt the Oracle wallet, if it is encrypted",
         default=None,
     )

api/configs/middleware/vdb/pgvector_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,7 +9,7 @@ class PGVectorConfig(BaseSettings):
     Configuration settings for PGVector (PostgreSQL with vector extension)
     """
 
-    PGVECTOR_HOST: str | None = Field(
+    PGVECTOR_HOST: Optional[str] = Field(
         description="Hostname or IP address of the PostgreSQL server with PGVector extension (e.g., 'localhost')",
         default=None,
     )
@@ -17,17 +19,17 @@ class PGVectorConfig(BaseSettings):
         default=5433,
     )
 
-    PGVECTOR_USER: str | None = Field(
+    PGVECTOR_USER: Optional[str] = Field(
         description="Username for authenticating with the PostgreSQL database",
         default=None,
     )
 
-    PGVECTOR_PASSWORD: str | None = Field(
+    PGVECTOR_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the PostgreSQL database",
         default=None,
     )
 
-    PGVECTOR_DATABASE: str | None = Field(
+    PGVECTOR_DATABASE: Optional[str] = Field(
         description="Name of the PostgreSQL database to connect to",
         default=None,
     )

api/configs/middleware/vdb/pgvectors_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,7 +9,7 @@ class PGVectoRSConfig(BaseSettings):
     Configuration settings for PGVecto.RS (Rust-based vector extension for PostgreSQL)
     """
 
-    PGVECTO_RS_HOST: str | None = Field(
+    PGVECTO_RS_HOST: Optional[str] = Field(
         description="Hostname or IP address of the PostgreSQL server with PGVecto.RS extension (e.g., 'localhost')",
         default=None,
     )
@@ -17,17 +19,17 @@ class PGVectoRSConfig(BaseSettings):
         default=5431,
     )
 
-    PGVECTO_RS_USER: str | None = Field(
+    PGVECTO_RS_USER: Optional[str] = Field(
         description="Username for authenticating with the PostgreSQL database using PGVecto.RS",
         default=None,
     )
 
-    PGVECTO_RS_PASSWORD: str | None = Field(
+    PGVECTO_RS_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the PostgreSQL database using PGVecto.RS",
         default=None,
     )
 
-    PGVECTO_RS_DATABASE: str | None = Field(
+    PGVECTO_RS_DATABASE: Optional[str] = Field(
         description="Name of the PostgreSQL database with PGVecto.RS extension to connect to",
         default=None,
     )

api/configs/middleware/vdb/qdrant_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, NonNegativeInt, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,12 +9,12 @@ class QdrantConfig(BaseSettings):
     Configuration settings for Qdrant vector database
     """
 
-    QDRANT_URL: str | None = Field(
+    QDRANT_URL: Optional[str] = Field(
         description="URL of the Qdrant server (e.g., 'http://localhost:6333' or 'https://qdrant.example.com')",
         default=None,
     )
 
-    QDRANT_API_KEY: str | None = Field(
+    QDRANT_API_KEY: Optional[str] = Field(
         description="API key for authenticating with the Qdrant server",
         default=None,
     )

api/configs/middleware/vdb/relyt_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,7 +9,7 @@ class RelytConfig(BaseSettings):
     Configuration settings for Relyt database
     """
 
-    RELYT_HOST: str | None = Field(
+    RELYT_HOST: Optional[str] = Field(
         description="Hostname or IP address of the Relyt server (e.g., 'localhost' or 'relyt.example.com')",
         default=None,
     )
@@ -17,17 +19,17 @@ class RelytConfig(BaseSettings):
         default=9200,
     )
 
-    RELYT_USER: str | None = Field(
+    RELYT_USER: Optional[str] = Field(
         description="Username for authenticating with the Relyt database",
         default=None,
     )
 
-    RELYT_PASSWORD: str | None = Field(
+    RELYT_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the Relyt database",
         default=None,
     )
 
-    RELYT_DATABASE: str | None = Field(
+    RELYT_DATABASE: Optional[str] = Field(
         description="Name of the Relyt database to connect to (default is 'default')",
         default="default",
     )

api/configs/middleware/vdb/tablestore_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,22 +9,22 @@ class TableStoreConfig(BaseSettings):
     Configuration settings for TableStore.
     """
 
-    TABLESTORE_ENDPOINT: str | None = Field(
+    TABLESTORE_ENDPOINT: Optional[str] = Field(
         description="Endpoint address of the TableStore server (e.g. 'https://instance-name.cn-hangzhou.ots.aliyuncs.com')",
         default=None,
     )
 
-    TABLESTORE_INSTANCE_NAME: str | None = Field(
+    TABLESTORE_INSTANCE_NAME: Optional[str] = Field(
         description="Instance name to access TableStore server (eg. 'instance-name')",
         default=None,
     )
 
-    TABLESTORE_ACCESS_KEY_ID: str | None = Field(
+    TABLESTORE_ACCESS_KEY_ID: Optional[str] = Field(
         description="AccessKey id for the instance name",
         default=None,
     )
 
-    TABLESTORE_ACCESS_KEY_SECRET: str | None = Field(
+    TABLESTORE_ACCESS_KEY_SECRET: Optional[str] = Field(
         description="AccessKey secret for the instance name",
         default=None,
     )

api/configs/middleware/vdb/tencent_vector_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, NonNegativeInt, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,12 +9,12 @@ class TencentVectorDBConfig(BaseSettings):
     Configuration settings for Tencent Vector Database
     """
 
-    TENCENT_VECTOR_DB_URL: str | None = Field(
+    TENCENT_VECTOR_DB_URL: Optional[str] = Field(
         description="URL of the Tencent Vector Database service (e.g., 'https://vectordb.tencentcloudapi.com')",
         default=None,
     )
 
-    TENCENT_VECTOR_DB_API_KEY: str | None = Field(
+    TENCENT_VECTOR_DB_API_KEY: Optional[str] = Field(
         description="API key for authenticating with the Tencent Vector Database service",
         default=None,
     )
@@ -22,12 +24,12 @@ class TencentVectorDBConfig(BaseSettings):
         default=30,
     )
 
-    TENCENT_VECTOR_DB_USERNAME: str | None = Field(
+    TENCENT_VECTOR_DB_USERNAME: Optional[str] = Field(
         description="Username for authenticating with the Tencent Vector Database (if required)",
         default=None,
     )
 
-    TENCENT_VECTOR_DB_PASSWORD: str | None = Field(
+    TENCENT_VECTOR_DB_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the Tencent Vector Database (if required)",
         default=None,
     )
@@ -42,7 +44,7 @@ class TencentVectorDBConfig(BaseSettings):
         default=2,
     )
 
-    TENCENT_VECTOR_DB_DATABASE: str | None = Field(
+    TENCENT_VECTOR_DB_DATABASE: Optional[str] = Field(
         description="Name of the specific Tencent Vector Database to connect to",
         default=None,
     )

api/configs/middleware/vdb/tidb_on_qdrant_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, NonNegativeInt, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,12 +9,12 @@ class TidbOnQdrantConfig(BaseSettings):
     Tidb on Qdrant configs
     """
 
-    TIDB_ON_QDRANT_URL: str | None = Field(
+    TIDB_ON_QDRANT_URL: Optional[str] = Field(
         description="Tidb on Qdrant url",
         default=None,
     )
 
-    TIDB_ON_QDRANT_API_KEY: str | None = Field(
+    TIDB_ON_QDRANT_API_KEY: Optional[str] = Field(
         description="Tidb on Qdrant api key",
         default=None,
     )
@@ -32,37 +34,37 @@ class TidbOnQdrantConfig(BaseSettings):
         default=6334,
     )
 
-    TIDB_PUBLIC_KEY: str | None = Field(
+    TIDB_PUBLIC_KEY: Optional[str] = Field(
         description="Tidb account public key",
         default=None,
     )
 
-    TIDB_PRIVATE_KEY: str | None = Field(
+    TIDB_PRIVATE_KEY: Optional[str] = Field(
         description="Tidb account private key",
         default=None,
     )
 
-    TIDB_API_URL: str | None = Field(
+    TIDB_API_URL: Optional[str] = Field(
         description="Tidb API url",
         default=None,
     )
 
-    TIDB_IAM_API_URL: str | None = Field(
+    TIDB_IAM_API_URL: Optional[str] = Field(
         description="Tidb IAM API url",
         default=None,
     )
 
-    TIDB_REGION: str | None = Field(
+    TIDB_REGION: Optional[str] = Field(
         description="Tidb serverless region",
         default="regions/aws-us-east-1",
     )
 
-    TIDB_PROJECT_ID: str | None = Field(
+    TIDB_PROJECT_ID: Optional[str] = Field(
         description="Tidb project id",
         default=None,
     )
 
-    TIDB_SPEND_LIMIT: int | None = Field(
+    TIDB_SPEND_LIMIT: Optional[int] = Field(
         description="Tidb spend limit",
         default=100,
     )

api/configs/middleware/vdb/tidb_vector_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,27 +9,27 @@ class TiDBVectorConfig(BaseSettings):
     Configuration settings for TiDB Vector database
     """
 
-    TIDB_VECTOR_HOST: str | None = Field(
+    TIDB_VECTOR_HOST: Optional[str] = Field(
         description="Hostname or IP address of the TiDB Vector server (e.g., 'localhost' or 'tidb.example.com')",
         default=None,
     )
 
-    TIDB_VECTOR_PORT: PositiveInt | None = Field(
+    TIDB_VECTOR_PORT: Optional[PositiveInt] = Field(
         description="Port number on which the TiDB Vector server is listening (default is 4000)",
         default=4000,
     )
 
-    TIDB_VECTOR_USER: str | None = Field(
+    TIDB_VECTOR_USER: Optional[str] = Field(
         description="Username for authenticating with the TiDB Vector database",
         default=None,
     )
 
-    TIDB_VECTOR_PASSWORD: str | None = Field(
+    TIDB_VECTOR_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the TiDB Vector database",
         default=None,
     )
 
-    TIDB_VECTOR_DATABASE: str | None = Field(
+    TIDB_VECTOR_DATABASE: Optional[str] = Field(
         description="Name of the TiDB Vector database to connect to",
         default=None,
     )

api/configs/middleware/vdb/upstash_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -7,12 +9,12 @@ class UpstashConfig(BaseSettings):
     Configuration settings for Upstash vector database
     """
 
-    UPSTASH_VECTOR_URL: str | None = Field(
+    UPSTASH_VECTOR_URL: Optional[str] = Field(
         description="URL of the upstash server (e.g., 'https://vector.upstash.io')",
         default=None,
     )
 
-    UPSTASH_VECTOR_TOKEN: str | None = Field(
+    UPSTASH_VECTOR_TOKEN: Optional[str] = Field(
         description="Token for authenticating with the upstash server",
         default=None,
     )

api/configs/middleware/vdb/vastbase_vector_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,7 +9,7 @@ class VastbaseVectorConfig(BaseSettings):
     Configuration settings for Vector (Vastbase with vector extension)
     """
 
-    VASTBASE_HOST: str | None = Field(
+    VASTBASE_HOST: Optional[str] = Field(
         description="Hostname or IP address of the Vastbase server with Vector extension (e.g., 'localhost')",
         default=None,
     )
@@ -17,17 +19,17 @@ class VastbaseVectorConfig(BaseSettings):
         default=5432,
     )
 
-    VASTBASE_USER: str | None = Field(
+    VASTBASE_USER: Optional[str] = Field(
         description="Username for authenticating with the Vastbase database",
         default=None,
     )
 
-    VASTBASE_PASSWORD: str | None = Field(
+    VASTBASE_PASSWORD: Optional[str] = Field(
         description="Password for authenticating with the Vastbase database",
         default=None,
     )
 
-    VASTBASE_DATABASE: str | None = Field(
+    VASTBASE_DATABASE: Optional[str] = Field(
         description="Name of the Vastbase database to connect to",
         default=None,
     )

api/configs/middleware/vdb/vikingdb_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -9,14 +11,14 @@ class VikingDBConfig(BaseSettings):
     https://www.volcengine.com/docs/6291/65568
     """
 
-    VIKINGDB_ACCESS_KEY: str | None = Field(
+    VIKINGDB_ACCESS_KEY: Optional[str] = Field(
         description="The Access Key provided by Volcengine VikingDB for API authentication."
         "Refer to the following documentation for details on obtaining credentials:"
         "https://www.volcengine.com/docs/6291/65568",
         default=None,
     )
 
-    VIKINGDB_SECRET_KEY: str | None = Field(
+    VIKINGDB_SECRET_KEY: Optional[str] = Field(
         description="The Secret Key provided by Volcengine VikingDB for API authentication.",
         default=None,
     )

api/configs/middleware/vdb/weaviate_config.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -7,12 +9,12 @@ class WeaviateConfig(BaseSettings):
     Configuration settings for Weaviate vector database
     """
 
-    WEAVIATE_ENDPOINT: str | None = Field(
+    WEAVIATE_ENDPOINT: Optional[str] = Field(
         description="URL of the Weaviate server (e.g., 'http://localhost:8080' or 'https://weaviate.example.com')",
         default=None,
     )
 
-    WEAVIATE_API_KEY: str | None = Field(
+    WEAVIATE_API_KEY: Optional[str] = Field(
         description="API key for authenticating with the Weaviate server",
         default=None,
     )

api/configs/remote_settings_sources/apollo/__init__.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping
-from typing import Any
+from typing import Any, Optional
 
 from pydantic import Field
 from pydantic.fields import FieldInfo
@@ -15,22 +15,22 @@ class ApolloSettingsSourceInfo(BaseSettings):
     Packaging build information
     """
 
-    APOLLO_APP_ID: str | None = Field(
+    APOLLO_APP_ID: Optional[str] = Field(
         description="apollo app_id",
         default=None,
     )
 
-    APOLLO_CLUSTER: str | None = Field(
+    APOLLO_CLUSTER: Optional[str] = Field(
         description="apollo cluster",
         default=None,
     )
 
-    APOLLO_CONFIG_URL: str | None = Field(
+    APOLLO_CONFIG_URL: Optional[str] = Field(
         description="apollo config url",
         default=None,
     )
 
-    APOLLO_NAMESPACE: str | None = Field(
+    APOLLO_NAMESPACE: Optional[str] = Field(
         description="apollo namespace",
         default=None,
     )

api/configs/remote_settings_sources/apollo/utils.py

@@ -29,7 +29,7 @@ def no_key_cache_key(namespace: str, key: str) -> str:
 
 
 # Returns whether the obtained value is obtained, and None if it does not
-def get_value_from_dict(namespace_cache: dict[str, Any] | None, key: str) -> Any:
+def get_value_from_dict(namespace_cache: dict[str, Any] | None, key: str) -> Any | None:
     if namespace_cache:
         kv_data = namespace_cache.get(CONFIGURATIONS)
         if kv_data is None:

api/contexts/__init__.py

@@ -5,7 +5,6 @@ from typing import TYPE_CHECKING
 from contexts.wrapper import RecyclableContextVar
 
 if TYPE_CHECKING:
-    from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
     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
@@ -33,11 +32,3 @@ 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

@@ -61,7 +61,6 @@ from . import (
     init_validate,
     ping,
     setup,
-    spec,
     version,
 )
 
@@ -115,15 +114,6 @@ from .datasets import (
     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 (
@@ -248,8 +238,6 @@ __all__ = [
     "datasets",
     "datasets_document",
     "datasets_segments",
-    "datasource_auth",
-    "datasource_content_preview",
     "email_register",
     "endpoint",
     "extension",
@@ -275,16 +263,10 @@ __all__ = [
     "parameter",
     "ping",
     "plugin",
-    "rag_pipeline",
-    "rag_pipeline_datasets",
-    "rag_pipeline_draft_variable",
-    "rag_pipeline_import",
-    "rag_pipeline_workflow",
     "recommended_app",
     "saved_message",
     "setup",
     "site",
-    "spec",
     "statistic",
     "tags",
     "tool_providers",

api/controllers/console/apikey.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 import flask_restx
 from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with
@@ -48,7 +50,7 @@ class BaseApiKeyListResource(Resource):
     method_decorators = [account_initialization_required, login_required, setup_required]
 
     resource_type: str | None = None
-    resource_model: type | None = None
+    resource_model: Optional[type] = None
     resource_id_field: str | None = None
     token_prefix: str | None = None
     max_keys = 10
@@ -101,7 +103,7 @@ class BaseApiKeyResource(Resource):
     method_decorators = [account_initialization_required, login_required, setup_required]
 
     resource_type: str | None = None
-    resource_model: type | None = None
+    resource_model: Optional[type] = None
     resource_id_field: str | None = None
 
     def delete(self, resource_id, api_key_id):

api/controllers/console/app/generator.py

@@ -16,10 +16,7 @@ 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
 
 
 @console_ns.route("/rule-generate")
@@ -208,6 +205,9 @@ 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
@@ -261,7 +261,6 @@ class InstructionGenerateApi(Resource):
                     instruction=args["instruction"],
                     model_config=args["model_config"],
                     ideal_output=args["ideal_output"],
-                    workflow_service=WorkflowService(),
                 )
             return {"error": "incompatible parameters"}, 400
         except ProviderTokenNotInitError as ex:

api/controllers/console/app/workflow.py

@@ -4,13 +4,13 @@ from collections.abc import Sequence
 from typing import cast
 
 from flask import abort, request
-from flask_restx import Resource, fields, inputs, marshal_with, reqparse
+from flask_restx import Resource, inputs, marshal_with, reqparse
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
 from configs import dify_config
-from controllers.console import api, console_ns
+from controllers.console import api
 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
@@ -20,7 +20,6 @@ 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
@@ -58,13 +57,7 @@ def _parse_file(workflow: Workflow, files: list[dict] | None = None) -> Sequence
     return file_objs
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft")
 class DraftWorkflowApi(Resource):
-    @api.doc("get_draft_workflow")
-    @api.doc(description="Get draft workflow for an application")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Draft workflow retrieved successfully", workflow_fields)
-    @api.response(404, "Draft workflow not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -93,23 +86,6 @@ class DraftWorkflowApi(Resource):
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @api.doc("sync_draft_workflow")
-    @api.doc(description="Sync draft workflow configuration")
-    @api.expect(
-        api.model(
-            "SyncDraftWorkflowRequest",
-            {
-                "graph": fields.Raw(required=True, description="Workflow graph configuration"),
-                "features": fields.Raw(required=True, description="Workflow features configuration"),
-                "hash": fields.String(description="Workflow hash for validation"),
-                "environment_variables": fields.List(fields.Raw, required=True, description="Environment variables"),
-                "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
-            },
-        )
-    )
-    @api.response(200, "Draft workflow synced successfully", workflow_fields)
-    @api.response(400, "Invalid workflow configuration")
-    @api.response(403, "Permission denied")
     def post(self, app_model: App):
         """
         Sync draft workflow
@@ -183,25 +159,7 @@ class DraftWorkflowApi(Resource):
         }
 
 
-@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/run")
 class AdvancedChatDraftWorkflowRunApi(Resource):
-    @api.doc("run_advanced_chat_draft_workflow")
-    @api.doc(description="Run draft workflow for advanced chat application")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "AdvancedChatWorkflowRunRequest",
-            {
-                "query": fields.String(required=True, description="User query"),
-                "inputs": fields.Raw(description="Input variables"),
-                "files": fields.List(fields.Raw, description="File uploads"),
-                "conversation_id": fields.String(description="Conversation ID"),
-            },
-        )
-    )
-    @api.response(200, "Workflow run started successfully")
-    @api.response(400, "Invalid request parameters")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -250,23 +208,7 @@ class AdvancedChatDraftWorkflowRunApi(Resource):
             raise InternalServerError()
 
 
-@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/iteration/nodes/<string:node_id>/run")
 class AdvancedChatDraftRunIterationNodeApi(Resource):
-    @api.doc("run_advanced_chat_draft_iteration_node")
-    @api.doc(description="Run draft workflow iteration node for advanced chat")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.expect(
-        api.model(
-            "IterationNodeRunRequest",
-            {
-                "task_id": fields.String(required=True, description="Task ID"),
-                "inputs": fields.Raw(description="Input variables"),
-            },
-        )
-    )
-    @api.response(200, "Iteration node run started successfully")
-    @api.response(403, "Permission denied")
-    @api.response(404, "Node not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -302,23 +244,7 @@ class AdvancedChatDraftRunIterationNodeApi(Resource):
             raise InternalServerError()
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/iteration/nodes/<string:node_id>/run")
 class WorkflowDraftRunIterationNodeApi(Resource):
-    @api.doc("run_workflow_draft_iteration_node")
-    @api.doc(description="Run draft workflow iteration node")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.expect(
-        api.model(
-            "WorkflowIterationNodeRunRequest",
-            {
-                "task_id": fields.String(required=True, description="Task ID"),
-                "inputs": fields.Raw(description="Input variables"),
-            },
-        )
-    )
-    @api.response(200, "Workflow iteration node run started successfully")
-    @api.response(403, "Permission denied")
-    @api.response(404, "Node not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -354,23 +280,7 @@ class WorkflowDraftRunIterationNodeApi(Resource):
             raise InternalServerError()
 
 
-@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/loop/nodes/<string:node_id>/run")
 class AdvancedChatDraftRunLoopNodeApi(Resource):
-    @api.doc("run_advanced_chat_draft_loop_node")
-    @api.doc(description="Run draft workflow loop node for advanced chat")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.expect(
-        api.model(
-            "LoopNodeRunRequest",
-            {
-                "task_id": fields.String(required=True, description="Task ID"),
-                "inputs": fields.Raw(description="Input variables"),
-            },
-        )
-    )
-    @api.response(200, "Loop node run started successfully")
-    @api.response(403, "Permission denied")
-    @api.response(404, "Node not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -407,23 +317,7 @@ class AdvancedChatDraftRunLoopNodeApi(Resource):
             raise InternalServerError()
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/loop/nodes/<string:node_id>/run")
 class WorkflowDraftRunLoopNodeApi(Resource):
-    @api.doc("run_workflow_draft_loop_node")
-    @api.doc(description="Run draft workflow loop node")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.expect(
-        api.model(
-            "WorkflowLoopNodeRunRequest",
-            {
-                "task_id": fields.String(required=True, description="Task ID"),
-                "inputs": fields.Raw(description="Input variables"),
-            },
-        )
-    )
-    @api.response(200, "Workflow loop node run started successfully")
-    @api.response(403, "Permission denied")
-    @api.response(404, "Node not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -460,22 +354,7 @@ class WorkflowDraftRunLoopNodeApi(Resource):
             raise InternalServerError()
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/run")
 class DraftWorkflowRunApi(Resource):
-    @api.doc("run_draft_workflow")
-    @api.doc(description="Run draft workflow")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "DraftWorkflowRunRequest",
-            {
-                "inputs": fields.Raw(required=True, description="Input variables"),
-                "files": fields.List(fields.Raw, description="File uploads"),
-            },
-        )
-    )
-    @api.response(200, "Draft workflow run started successfully")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -514,14 +393,7 @@ class DraftWorkflowRunApi(Resource):
             raise InvokeRateLimitHttpError(ex.description)
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow-runs/tasks/<string:task_id>/stop")
 class WorkflowTaskStopApi(Resource):
-    @api.doc("stop_workflow_task")
-    @api.doc(description="Stop running workflow task")
-    @api.doc(params={"app_id": "Application ID", "task_id": "Task ID"})
-    @api.response(200, "Task stopped successfully")
-    @api.response(404, "Task not found")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -537,32 +409,12 @@ class WorkflowTaskStopApi(Resource):
         if not current_user.has_edit_permission:
             raise Forbidden()
 
-        # 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)
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
 
         return {"result": "success"}
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/run")
 class DraftWorkflowNodeRunApi(Resource):
-    @api.doc("run_draft_workflow_node")
-    @api.doc(description="Run draft workflow node")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.expect(
-        api.model(
-            "DraftWorkflowNodeRunRequest",
-            {
-                "inputs": fields.Raw(description="Input variables"),
-            },
-        )
-    )
-    @api.response(200, "Node run started successfully", workflow_run_node_execution_fields)
-    @api.response(403, "Permission denied")
-    @api.response(404, "Node not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -610,13 +462,7 @@ class DraftWorkflowNodeRunApi(Resource):
         return workflow_node_execution
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/publish")
 class PublishedWorkflowApi(Resource):
-    @api.doc("get_published_workflow")
-    @api.doc(description="Get published workflow for an application")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Published workflow retrieved successfully", workflow_fields)
-    @api.response(404, "Published workflow not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -688,12 +534,7 @@ class PublishedWorkflowApi(Resource):
         }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/default-workflow-block-configs")
 class DefaultBlockConfigsApi(Resource):
-    @api.doc("get_default_block_configs")
-    @api.doc(description="Get default block configurations for workflow")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Default block configurations retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -714,13 +555,7 @@ class DefaultBlockConfigsApi(Resource):
         return workflow_service.get_default_block_configs()
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>")
 class DefaultBlockConfigApi(Resource):
-    @api.doc("get_default_block_config")
-    @api.doc(description="Get default block configuration by type")
-    @api.doc(params={"app_id": "Application ID", "block_type": "Block type"})
-    @api.response(200, "Default block configuration retrieved successfully")
-    @api.response(404, "Block type not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -753,14 +588,7 @@ class DefaultBlockConfigApi(Resource):
         return workflow_service.get_default_block_config(node_type=block_type, filters=filters)
 
 
-@console_ns.route("/apps/<uuid:app_id>/convert-to-workflow")
 class ConvertToWorkflowApi(Resource):
-    @api.doc("convert_to_workflow")
-    @api.doc(description="Convert application to workflow mode")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Application converted to workflow successfully")
-    @api.response(400, "Application cannot be converted")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -797,14 +625,9 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/config")
 class WorkflowConfigApi(Resource):
     """Resource for workflow configuration."""
 
-    @api.doc("get_workflow_config")
-    @api.doc(description="Get workflow configuration")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Workflow configuration retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -815,12 +638,7 @@ class WorkflowConfigApi(Resource):
         }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
-    @api.doc("get_all_published_workflows")
-    @api.doc(description="Get all published workflows for an application")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Published workflows retrieved successfully", workflow_pagination_fields)
     @setup_required
     @login_required
     @account_initialization_required
@@ -871,23 +689,7 @@ class PublishedAllWorkflowApi(Resource):
             }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>")
 class WorkflowByIdApi(Resource):
-    @api.doc("update_workflow_by_id")
-    @api.doc(description="Update workflow by ID")
-    @api.doc(params={"app_id": "Application ID", "workflow_id": "Workflow ID"})
-    @api.expect(
-        api.model(
-            "UpdateWorkflowRequest",
-            {
-                "environment_variables": fields.List(fields.Raw, description="Environment variables"),
-                "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
-            },
-        )
-    )
-    @api.response(200, "Workflow updated successfully", workflow_fields)
-    @api.response(404, "Workflow not found")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -978,14 +780,7 @@ class WorkflowByIdApi(Resource):
         return None, 204
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/last-run")
 class DraftWorkflowNodeLastRunApi(Resource):
-    @api.doc("get_draft_workflow_node_last_run")
-    @api.doc(description="Get last run result for draft workflow node")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.response(200, "Node last run retrieved successfully", workflow_run_node_execution_fields)
-    @api.response(404, "Node last run not found")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -1004,3 +799,73 @@ class DraftWorkflowNodeLastRunApi(Resource):
         if node_exec is None:
             raise NotFound("last run not found")
         return node_exec
+
+
+api.add_resource(
+    DraftWorkflowApi,
+    "/apps/<uuid:app_id>/workflows/draft",
+)
+api.add_resource(
+    WorkflowConfigApi,
+    "/apps/<uuid:app_id>/workflows/draft/config",
+)
+api.add_resource(
+    AdvancedChatDraftWorkflowRunApi,
+    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/run",
+)
+api.add_resource(
+    DraftWorkflowRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/run",
+)
+api.add_resource(
+    WorkflowTaskStopApi,
+    "/apps/<uuid:app_id>/workflow-runs/tasks/<string:task_id>/stop",
+)
+api.add_resource(
+    DraftWorkflowNodeRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    AdvancedChatDraftRunIterationNodeApi,
+    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/iteration/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    WorkflowDraftRunIterationNodeApi,
+    "/apps/<uuid:app_id>/workflows/draft/iteration/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    AdvancedChatDraftRunLoopNodeApi,
+    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/loop/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    WorkflowDraftRunLoopNodeApi,
+    "/apps/<uuid:app_id>/workflows/draft/loop/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    PublishedWorkflowApi,
+    "/apps/<uuid:app_id>/workflows/publish",
+)
+api.add_resource(
+    PublishedAllWorkflowApi,
+    "/apps/<uuid:app_id>/workflows",
+)
+api.add_resource(
+    DefaultBlockConfigsApi,
+    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs",
+)
+api.add_resource(
+    DefaultBlockConfigApi,
+    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>",
+)
+api.add_resource(
+    ConvertToWorkflowApi,
+    "/apps/<uuid:app_id>/convert-to-workflow",
+)
+api.add_resource(
+    WorkflowByIdApi,
+    "/apps/<uuid:app_id>/workflows/<string:workflow_id>",
+)
+api.add_resource(
+    DraftWorkflowNodeLastRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/last-run",
+)

api/controllers/console/app/workflow_app_log.py

@@ -3,10 +3,10 @@ from flask_restx import Resource, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from sqlalchemy.orm import Session
 
-from controllers.console import api, console_ns
+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.enums import WorkflowExecutionStatus
+from core.workflow.entities.workflow_execution 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
@@ -15,24 +15,7 @@ from models.model import AppMode
 from services.workflow_app_service import WorkflowAppService
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow-app-logs")
 class WorkflowAppLogApi(Resource):
-    @api.doc("get_workflow_app_logs")
-    @api.doc(description="Get workflow application execution logs")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.doc(
-        params={
-            "keyword": "Search keyword for filtering logs",
-            "status": "Filter by execution status (succeeded, failed, stopped, partial-succeeded)",
-            "created_at__before": "Filter logs created before this timestamp",
-            "created_at__after": "Filter logs created after this timestamp",
-            "created_by_end_user_session_id": "Filter by end user session ID",
-            "created_by_account": "Filter by account",
-            "page": "Page number (1-99999)",
-            "limit": "Number of items per page (1-100)",
-        }
-    )
-    @api.response(200, "Workflow app logs retrieved successfully", workflow_app_log_pagination_fields)
     @setup_required
     @login_required
     @account_initialization_required
@@ -95,3 +78,6 @@ class WorkflowAppLogApi(Resource):
             )
 
             return workflow_app_log_pagination
+
+
+api.add_resource(WorkflowAppLogApi, "/apps/<uuid:app_id>/workflow-app-logs")

api/controllers/console/app/workflow_draft_variable.py

@@ -6,23 +6,21 @@ from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqpars
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden
 
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.app.error import (
     DraftWorkflowNotExist,
 )
 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
+from models import App, AppMode, db
 from models.account import Account
 from models.workflow import WorkflowDraftVariable
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
@@ -76,22 +74,6 @@ 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()),
@@ -101,13 +83,11 @@ _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 = {
@@ -164,13 +144,7 @@ def _api_prerequisite(f):
     return wrapper
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/variables")
 class WorkflowVariableCollectionApi(Resource):
-    @api.doc("get_workflow_variables")
-    @api.doc(description="Get draft workflow variables")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.doc(params={"page": "Page number (1-100000)", "limit": "Number of items per page (1-100)"})
-    @api.response(200, "Workflow variables retrieved successfully", _WORKFLOW_DRAFT_VARIABLE_LIST_WITHOUT_VALUE_FIELDS)
     @_api_prerequisite
     @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_WITHOUT_VALUE_FIELDS)
     def get(self, app_model: App):
@@ -199,9 +173,6 @@ class WorkflowVariableCollectionApi(Resource):
 
         return workflow_vars
 
-    @api.doc("delete_workflow_variables")
-    @api.doc(description="Delete all draft workflow variables")
-    @api.response(204, "Workflow variables deleted successfully")
     @_api_prerequisite
     def delete(self, app_model: App):
         draft_var_srv = WorkflowDraftVariableService(
@@ -230,12 +201,7 @@ def validate_node_id(node_id: str) -> NoReturn | None:
     return None
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
 class NodeVariableCollectionApi(Resource):
-    @api.doc("get_node_variables")
-    @api.doc(description="Get variables for a specific node")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.response(200, "Node variables retrieved successfully", _WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
     @_api_prerequisite
     @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
     def get(self, app_model: App, node_id: str):
@@ -248,9 +214,6 @@ class NodeVariableCollectionApi(Resource):
 
         return node_vars
 
-    @api.doc("delete_node_variables")
-    @api.doc(description="Delete all variables for a specific node")
-    @api.response(204, "Node variables deleted successfully")
     @_api_prerequisite
     def delete(self, app_model: App, node_id: str):
         validate_node_id(node_id)
@@ -260,16 +223,10 @@ class NodeVariableCollectionApi(Resource):
         return Response("", 204)
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>")
 class VariableApi(Resource):
     _PATCH_NAME_FIELD = "name"
     _PATCH_VALUE_FIELD = "value"
 
-    @api.doc("get_variable")
-    @api.doc(description="Get a specific workflow variable")
-    @api.doc(params={"app_id": "Application ID", "variable_id": "Variable ID"})
-    @api.response(200, "Variable retrieved successfully", _WORKFLOW_DRAFT_VARIABLE_FIELDS)
-    @api.response(404, "Variable not found")
     @_api_prerequisite
     @marshal_with(_WORKFLOW_DRAFT_VARIABLE_FIELDS)
     def get(self, app_model: App, variable_id: str):
@@ -283,19 +240,6 @@ class VariableApi(Resource):
             raise NotFoundError(description=f"variable not found, id={variable_id}")
         return variable
 
-    @api.doc("update_variable")
-    @api.doc(description="Update a workflow variable")
-    @api.expect(
-        api.model(
-            "UpdateVariableRequest",
-            {
-                "name": fields.String(description="Variable name"),
-                "value": fields.Raw(description="Variable value"),
-            },
-        )
-    )
-    @api.response(200, "Variable updated successfully", _WORKFLOW_DRAFT_VARIABLE_FIELDS)
-    @api.response(404, "Variable not found")
     @_api_prerequisite
     @marshal_with(_WORKFLOW_DRAFT_VARIABLE_FIELDS)
     def patch(self, app_model: App, variable_id: str):
@@ -358,10 +302,6 @@ class VariableApi(Resource):
         db.session.commit()
         return variable
 
-    @api.doc("delete_variable")
-    @api.doc(description="Delete a workflow variable")
-    @api.response(204, "Variable deleted successfully")
-    @api.response(404, "Variable not found")
     @_api_prerequisite
     def delete(self, app_model: App, variable_id: str):
         draft_var_srv = WorkflowDraftVariableService(
@@ -377,14 +317,7 @@ class VariableApi(Resource):
         return Response("", 204)
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>/reset")
 class VariableResetApi(Resource):
-    @api.doc("reset_variable")
-    @api.doc(description="Reset a workflow variable to its default value")
-    @api.doc(params={"app_id": "Application ID", "variable_id": "Variable ID"})
-    @api.response(200, "Variable reset successfully", _WORKFLOW_DRAFT_VARIABLE_FIELDS)
-    @api.response(204, "Variable reset (no content)")
-    @api.response(404, "Variable not found")
     @_api_prerequisite
     def put(self, app_model: App, variable_id: str):
         draft_var_srv = WorkflowDraftVariableService(
@@ -425,13 +358,7 @@ def _get_variable_list(app_model: App, node_id) -> WorkflowDraftVariableList:
     return draft_vars
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/conversation-variables")
 class ConversationVariableCollectionApi(Resource):
-    @api.doc("get_conversation_variables")
-    @api.doc(description="Get conversation variables for workflow")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Conversation variables retrieved successfully", _WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
-    @api.response(404, "Draft workflow not found")
     @_api_prerequisite
     @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
     def get(self, app_model: App):
@@ -447,25 +374,14 @@ class ConversationVariableCollectionApi(Resource):
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/system-variables")
 class SystemVariableCollectionApi(Resource):
-    @api.doc("get_system_variables")
-    @api.doc(description="Get system variables for workflow")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "System variables retrieved successfully", _WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
     @_api_prerequisite
     @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
     def get(self, app_model: App):
         return _get_variable_list(app_model, SYSTEM_VARIABLE_NODE_ID)
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/environment-variables")
 class EnvironmentVariableCollectionApi(Resource):
-    @api.doc("get_environment_variables")
-    @api.doc(description="Get environment variables for workflow")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Environment variables retrieved successfully")
-    @api.response(404, "Draft workflow not found")
     @_api_prerequisite
     def get(self, app_model: App):
         """
@@ -497,3 +413,16 @@ class EnvironmentVariableCollectionApi(Resource):
             )
 
         return {"items": env_vars_list}
+
+
+api.add_resource(
+    WorkflowVariableCollectionApi,
+    "/apps/<uuid:app_id>/workflows/draft/variables",
+)
+api.add_resource(NodeVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
+api.add_resource(VariableApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>")
+api.add_resource(VariableResetApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>/reset")
+
+api.add_resource(ConversationVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/conversation-variables")
+api.add_resource(SystemVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/system-variables")
+api.add_resource(EnvironmentVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/environment-variables")

api/controllers/console/app/workflow_run.py

@@ -4,7 +4,7 @@ from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 from flask_restx.inputs import int_range
 
-from controllers.console import api, console_ns
+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 fields.workflow_run_fields import (
@@ -19,13 +19,7 @@ from models import Account, App, AppMode, EndUser
 from services.workflow_run_service import WorkflowRunService
 
 
-@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflow-runs")
 class AdvancedChatAppWorkflowRunListApi(Resource):
-    @api.doc("get_advanced_chat_workflow_runs")
-    @api.doc(description="Get advanced chat workflow run list")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.doc(params={"last_id": "Last run ID for pagination", "limit": "Number of items per page (1-100)"})
-    @api.response(200, "Workflow runs retrieved successfully", advanced_chat_workflow_run_pagination_fields)
     @setup_required
     @login_required
     @account_initialization_required
@@ -46,13 +40,7 @@ class AdvancedChatAppWorkflowRunListApi(Resource):
         return result
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow-runs")
 class WorkflowRunListApi(Resource):
-    @api.doc("get_workflow_runs")
-    @api.doc(description="Get workflow run list")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.doc(params={"last_id": "Last run ID for pagination", "limit": "Number of items per page (1-100)"})
-    @api.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_fields)
     @setup_required
     @login_required
     @account_initialization_required
@@ -73,13 +61,7 @@ class WorkflowRunListApi(Resource):
         return result
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>")
 class WorkflowRunDetailApi(Resource):
-    @api.doc("get_workflow_run_detail")
-    @api.doc(description="Get workflow run detail")
-    @api.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
-    @api.response(200, "Workflow run detail retrieved successfully", workflow_run_detail_fields)
-    @api.response(404, "Workflow run not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -97,13 +79,7 @@ class WorkflowRunDetailApi(Resource):
         return workflow_run
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>/node-executions")
 class WorkflowRunNodeExecutionListApi(Resource):
-    @api.doc("get_workflow_run_node_executions")
-    @api.doc(description="Get workflow run node execution list")
-    @api.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
-    @api.response(200, "Node executions retrieved successfully", workflow_run_node_execution_list_fields)
-    @api.response(404, "Workflow run not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -124,3 +100,9 @@ class WorkflowRunNodeExecutionListApi(Resource):
         )
 
         return {"data": node_executions}
+
+
+api.add_resource(AdvancedChatAppWorkflowRunListApi, "/apps/<uuid:app_id>/advanced-chat/workflow-runs")
+api.add_resource(WorkflowRunListApi, "/apps/<uuid:app_id>/workflow-runs")
+api.add_resource(WorkflowRunDetailApi, "/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>")
+api.add_resource(WorkflowRunNodeExecutionListApi, "/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>/node-executions")

api/controllers/console/app/workflow_statistic.py

@@ -7,7 +7,7 @@ from flask import jsonify
 from flask_login import current_user
 from flask_restx import Resource, reqparse
 
-from controllers.console import api, console_ns
+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 extensions.ext_database import db
@@ -17,13 +17,7 @@ from models.enums import WorkflowRunTriggeredFrom
 from models.model import AppMode
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow/statistics/daily-conversations")
 class WorkflowDailyRunsStatistic(Resource):
-    @api.doc("get_workflow_daily_runs_statistic")
-    @api.doc(description="Get workflow daily runs statistics")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.doc(params={"start": "Start date and time (YYYY-MM-DD HH:MM)", "end": "End date and time (YYYY-MM-DD HH:MM)"})
-    @api.response(200, "Daily runs statistics retrieved successfully")
     @get_app_model
     @setup_required
     @login_required
@@ -85,13 +79,7 @@ WHERE
         return jsonify({"data": response_data})
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow/statistics/daily-terminals")
 class WorkflowDailyTerminalsStatistic(Resource):
-    @api.doc("get_workflow_daily_terminals_statistic")
-    @api.doc(description="Get workflow daily terminals statistics")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.doc(params={"start": "Start date and time (YYYY-MM-DD HH:MM)", "end": "End date and time (YYYY-MM-DD HH:MM)"})
-    @api.response(200, "Daily terminals statistics retrieved successfully")
     @get_app_model
     @setup_required
     @login_required
@@ -153,13 +141,7 @@ WHERE
         return jsonify({"data": response_data})
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow/statistics/token-costs")
 class WorkflowDailyTokenCostStatistic(Resource):
-    @api.doc("get_workflow_daily_token_cost_statistic")
-    @api.doc(description="Get workflow daily token cost statistics")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.doc(params={"start": "Start date and time (YYYY-MM-DD HH:MM)", "end": "End date and time (YYYY-MM-DD HH:MM)"})
-    @api.response(200, "Daily token cost statistics retrieved successfully")
     @get_app_model
     @setup_required
     @login_required
@@ -226,13 +208,7 @@ WHERE
         return jsonify({"data": response_data})
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflow/statistics/average-app-interactions")
 class WorkflowAverageAppInteractionStatistic(Resource):
-    @api.doc("get_workflow_average_app_interaction_statistic")
-    @api.doc(description="Get workflow average app interaction statistics")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.doc(params={"start": "Start date and time (YYYY-MM-DD HH:MM)", "end": "End date and time (YYYY-MM-DD HH:MM)"})
-    @api.response(200, "Average app interaction statistics retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -309,3 +285,11 @@ GROUP BY
                 )
 
         return jsonify({"data": response_data})
+
+
+api.add_resource(WorkflowDailyRunsStatistic, "/apps/<uuid:app_id>/workflow/statistics/daily-conversations")
+api.add_resource(WorkflowDailyTerminalsStatistic, "/apps/<uuid:app_id>/workflow/statistics/daily-terminals")
+api.add_resource(WorkflowDailyTokenCostStatistic, "/apps/<uuid:app_id>/workflow/statistics/token-costs")
+api.add_resource(
+    WorkflowAverageAppInteractionStatistic, "/apps/<uuid:app_id>/workflow/statistics/average-app-interactions"
+)

api/controllers/console/app/wraps.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import ParamSpec, TypeVar, Union
+from typing import Optional, ParamSpec, TypeVar, Union
 
 from controllers.console.app.error import AppNotFoundError
 from extensions.ext_database import db
@@ -12,7 +12,7 @@ P = ParamSpec("P")
 R = TypeVar("R")
 
 
-def _load_app_model(app_id: str) -> App | None:
+def _load_app_model(app_id: str) -> Optional[App]:
     assert isinstance(current_user, Account)
     app_model = (
         db.session.query(App)
@@ -22,7 +22,7 @@ def _load_app_model(app_id: str) -> App | None:
     return app_model
 
 
-def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
+def get_app_model(view: Optional[Callable[P, R]] = None, *, mode: Union[AppMode, list[AppMode], None] = None):
     def decorator(view_func: Callable[P, R]):
         @wraps(view_func)
         def decorated_view(*args: P.args, **kwargs: P.kwargs):

api/controllers/console/auth/oauth.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Optional
 
 import requests
 from flask import current_app, redirect, request
@@ -156,8 +157,8 @@ class OAuthCallback(Resource):
         )
 
 
-def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) -> Account | None:
-    account: Account | None = Account.get_by_openid(provider, user_info.id)
+def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) -> Optional[Account]:
+    account: Optional[Account] = Account.get_by_openid(provider, user_info.id)
 
     if not account:
         with Session(db.engine) as session:

api/controllers/console/datasets/data_source.py

@@ -1,6 +1,4 @@
 import json
-from collections.abc import Generator
-from typing import cast
 
 from flask import request
 from flask_login import current_user
@@ -11,8 +9,6 @@ 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
@@ -23,7 +19,6 @@ 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
 
 
@@ -116,18 +111,6 @@ 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
@@ -151,49 +134,31 @@ 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
-            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(),
+            data_source_bindings = session.scalars(
+                select(DataSourceOauthBinding).filter_by(
+                    tenant_id=current_user.current_tenant_id, provider="notion", disabled=False
                 )
-            )
-            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
+            ).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
 
 
 class DataSourceNotionApi(Resource):
@@ -201,25 +166,27 @@ 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=credential.get("integration_secret"),
+            notion_access_token=data_source_binding.access_token,
             tenant_id=current_user.current_tenant_id,
         )
 
@@ -244,12 +211,10 @@ 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"],

api/controllers/console/datasets/datasets.py

@@ -1,13 +1,13 @@
 import flask_restx
 from flask import request
 from flask_login import current_user
-from flask_restx import Resource, fields, marshal, marshal_with, reqparse
+from flask_restx import Resource, marshal, marshal_with, reqparse
 from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
 from configs import dify_config
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.apikey import api_key_fields, api_key_list
 from controllers.console.app.error import ProviderNotInitializeError
 from controllers.console.datasets.error import DatasetInUseError, DatasetNameDuplicateError, IndexingEstimateError
@@ -20,6 +20,7 @@ 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
@@ -32,7 +33,6 @@ 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
 
 
@@ -48,21 +48,7 @@ def _validate_description_length(description):
     return description
 
 
-@console_ns.route("/datasets")
 class DatasetListApi(Resource):
-    @api.doc("get_datasets")
-    @api.doc(description="Get list of datasets")
-    @api.doc(
-        params={
-            "page": "Page number (default: 1)",
-            "limit": "Number of items per page (default: 20)",
-            "ids": "Filter by dataset IDs (list)",
-            "keyword": "Search keyword",
-            "tag_ids": "Filter by tag IDs (list)",
-            "include_all": "Include all datasets (default: false)",
-        }
-    )
-    @api.response(200, "Datasets retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -114,24 +100,6 @@ class DatasetListApi(Resource):
         response = {"data": data, "has_more": len(datasets) == limit, "limit": limit, "total": total, "page": page}
         return response, 200
 
-    @api.doc("create_dataset")
-    @api.doc(description="Create a new dataset")
-    @api.expect(
-        api.model(
-            "CreateDatasetRequest",
-            {
-                "name": fields.String(required=True, description="Dataset name (1-40 characters)"),
-                "description": fields.String(description="Dataset description (max 400 characters)"),
-                "indexing_technique": fields.String(description="Indexing technique"),
-                "permission": fields.String(description="Dataset permission"),
-                "provider": fields.String(description="Provider"),
-                "external_knowledge_api_id": fields.String(description="External knowledge API ID"),
-                "external_knowledge_id": fields.String(description="External knowledge ID"),
-            },
-        )
-    )
-    @api.response(201, "Dataset created successfully")
-    @api.response(400, "Invalid request parameters")
     @setup_required
     @login_required
     @account_initialization_required
@@ -204,14 +172,7 @@ class DatasetListApi(Resource):
         return marshal(dataset, dataset_detail_fields), 201
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>")
 class DatasetApi(Resource):
-    @api.doc("get_dataset")
-    @api.doc(description="Get dataset details")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.response(200, "Dataset retrieved successfully", dataset_detail_fields)
-    @api.response(404, "Dataset not found")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -254,23 +215,6 @@ class DatasetApi(Resource):
 
         return data, 200
 
-    @api.doc("update_dataset")
-    @api.doc(description="Update dataset details")
-    @api.expect(
-        api.model(
-            "UpdateDatasetRequest",
-            {
-                "name": fields.String(description="Dataset name"),
-                "description": fields.String(description="Dataset description"),
-                "permission": fields.String(description="Dataset permission"),
-                "indexing_technique": fields.String(description="Indexing technique"),
-                "external_retrieval_model": fields.Raw(description="External retrieval model settings"),
-            },
-        )
-    )
-    @api.response(200, "Dataset updated successfully", dataset_detail_fields)
-    @api.response(404, "Dataset not found")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -337,15 +281,6 @@ 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()
 
@@ -396,7 +331,7 @@ class DatasetApi(Resource):
         dataset_id_str = str(dataset_id)
 
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not (current_user.is_editor or current_user.is_dataset_operator):
+        if not current_user.is_editor or current_user.is_dataset_operator:
             raise Forbidden()
 
         try:
@@ -409,12 +344,7 @@ class DatasetApi(Resource):
             raise DatasetInUseError()
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/use-check")
 class DatasetUseCheckApi(Resource):
-    @api.doc("check_dataset_use")
-    @api.doc(description="Check if dataset is in use")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.response(200, "Dataset use status retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -425,12 +355,7 @@ class DatasetUseCheckApi(Resource):
         return {"is_using": dataset_is_using}, 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/queries")
 class DatasetQueryApi(Resource):
-    @api.doc("get_dataset_queries")
-    @api.doc(description="Get dataset query history")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.response(200, "Query history retrieved successfully", dataset_query_detail_fields)
     @setup_required
     @login_required
     @account_initialization_required
@@ -460,11 +385,7 @@ class DatasetQueryApi(Resource):
         return response, 200
 
 
-@console_ns.route("/datasets/indexing-estimate")
 class DatasetIndexingEstimateApi(Resource):
-    @api.doc("estimate_dataset_indexing")
-    @api.doc(description="Estimate dataset indexing cost")
-    @api.response(200, "Indexing estimate calculated successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -512,12 +433,10 @@ 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"],
@@ -567,12 +486,7 @@ class DatasetIndexingEstimateApi(Resource):
         return response.model_dump(), 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/related-apps")
 class DatasetRelatedAppListApi(Resource):
-    @api.doc("get_dataset_related_apps")
-    @api.doc(description="Get applications related to dataset")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.response(200, "Related apps retrieved successfully", related_app_list)
     @setup_required
     @login_required
     @account_initialization_required
@@ -599,12 +513,7 @@ class DatasetRelatedAppListApi(Resource):
         return {"data": related_apps, "total": len(related_apps)}, 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/indexing-status")
 class DatasetIndexingStatusApi(Resource):
-    @api.doc("get_dataset_indexing_status")
-    @api.doc(description="Get dataset indexing status")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.response(200, "Indexing status retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -651,15 +560,11 @@ class DatasetIndexingStatusApi(Resource):
         return data, 200
 
 
-@console_ns.route("/datasets/api-keys")
 class DatasetApiKeyApi(Resource):
     max_keys = 10
     token_prefix = "dataset-"
     resource_type = "dataset"
 
-    @api.doc("get_dataset_api_keys")
-    @api.doc(description="Get dataset API keys")
-    @api.response(200, "API keys retrieved successfully", api_key_list)
     @setup_required
     @login_required
     @account_initialization_required
@@ -704,14 +609,9 @@ class DatasetApiKeyApi(Resource):
         return api_token, 200
 
 
-@console_ns.route("/datasets/api-keys/<uuid:api_key_id>")
 class DatasetApiDeleteApi(Resource):
     resource_type = "dataset"
 
-    @api.doc("delete_dataset_api_key")
-    @api.doc(description="Delete dataset API key")
-    @api.doc(params={"api_key_id": "API key ID"})
-    @api.response(204, "API key deleted successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -741,24 +641,7 @@ class DatasetApiDeleteApi(Resource):
         return {"result": "success"}, 204
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/api-keys/<string:status>")
-class DatasetEnableApiApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, dataset_id, status):
-        dataset_id_str = str(dataset_id)
-
-        DatasetService.update_dataset_api_status(dataset_id_str, status == "enable")
-
-        return {"result": "success"}, 200
-
-
-@console_ns.route("/datasets/api-base-info")
 class DatasetApiBaseUrlApi(Resource):
-    @api.doc("get_dataset_api_base_info")
-    @api.doc(description="Get dataset API base information")
-    @api.response(200, "API base info retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -766,11 +649,7 @@ class DatasetApiBaseUrlApi(Resource):
         return {"api_base_url": (dify_config.SERVICE_API_URL or request.host_url.rstrip("/")) + "/v1"}
 
 
-@console_ns.route("/datasets/retrieval-setting")
 class DatasetRetrievalSettingApi(Resource):
-    @api.doc("get_dataset_retrieval_setting")
-    @api.doc(description="Get dataset retrieval settings")
-    @api.response(200, "Retrieval settings retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -821,12 +700,7 @@ class DatasetRetrievalSettingApi(Resource):
                 raise ValueError(f"Unsupported vector db type {vector_type}.")
 
 
-@console_ns.route("/datasets/retrieval-setting/<string:vector_type>")
 class DatasetRetrievalSettingMockApi(Resource):
-    @api.doc("get_dataset_retrieval_setting_mock")
-    @api.doc(description="Get mock dataset retrieval settings by vector type")
-    @api.doc(params={"vector_type": "Vector store type"})
-    @api.response(200, "Mock retrieval settings retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -875,13 +749,7 @@ class DatasetRetrievalSettingMockApi(Resource):
                 raise ValueError(f"Unsupported vector db type {vector_type}.")
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/error-docs")
 class DatasetErrorDocs(Resource):
-    @api.doc("get_dataset_error_docs")
-    @api.doc(description="Get dataset error documents")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.response(200, "Error documents retrieved successfully")
-    @api.response(404, "Dataset not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -895,14 +763,7 @@ class DatasetErrorDocs(Resource):
         return {"data": [marshal(item, document_status_fields) for item in results], "total": len(results)}, 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/permission-part-users")
 class DatasetPermissionUserListApi(Resource):
-    @api.doc("get_dataset_permission_users")
-    @api.doc(description="Get dataset permission user list")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.response(200, "Permission users retrieved successfully")
-    @api.response(404, "Dataset not found")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -923,13 +784,7 @@ class DatasetPermissionUserListApi(Resource):
         }, 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/auto-disable-logs")
 class DatasetAutoDisableLogApi(Resource):
-    @api.doc("get_dataset_auto_disable_logs")
-    @api.doc(description="Get dataset auto disable logs")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.response(200, "Auto disable logs retrieved successfully")
-    @api.response(404, "Dataset not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -939,3 +794,20 @@ class DatasetAutoDisableLogApi(Resource):
         if dataset is None:
             raise NotFound("Dataset not found.")
         return DatasetService.get_dataset_auto_disable_logs(dataset_id_str), 200
+
+
+api.add_resource(DatasetListApi, "/datasets")
+api.add_resource(DatasetApi, "/datasets/<uuid:dataset_id>")
+api.add_resource(DatasetUseCheckApi, "/datasets/<uuid:dataset_id>/use-check")
+api.add_resource(DatasetQueryApi, "/datasets/<uuid:dataset_id>/queries")
+api.add_resource(DatasetErrorDocs, "/datasets/<uuid:dataset_id>/error-docs")
+api.add_resource(DatasetIndexingEstimateApi, "/datasets/indexing-estimate")
+api.add_resource(DatasetRelatedAppListApi, "/datasets/<uuid:dataset_id>/related-apps")
+api.add_resource(DatasetIndexingStatusApi, "/datasets/<uuid:dataset_id>/indexing-status")
+api.add_resource(DatasetApiKeyApi, "/datasets/api-keys")
+api.add_resource(DatasetApiDeleteApi, "/datasets/api-keys/<uuid:api_key_id>")
+api.add_resource(DatasetApiBaseUrlApi, "/datasets/api-base-info")
+api.add_resource(DatasetRetrievalSettingApi, "/datasets/retrieval-setting")
+api.add_resource(DatasetRetrievalSettingMockApi, "/datasets/retrieval-setting/<string:vector_type>")
+api.add_resource(DatasetPermissionUserListApi, "/datasets/<uuid:dataset_id>/permission-part-users")
+api.add_resource(DatasetAutoDisableLogApi, "/datasets/<uuid:dataset_id>/auto-disable-logs")

api/controllers/console/datasets/datasets_document.py

@@ -1,4 +1,3 @@
-import json
 import logging
 from argparse import ArgumentTypeError
 from collections.abc import Sequence
@@ -6,12 +5,12 @@ from typing import Literal, cast
 
 from flask import request
 from flask_login import current_user
-from flask_restx import Resource, fields, marshal, marshal_with, reqparse
+from flask_restx import Resource, marshal, marshal_with, reqparse
 from sqlalchemy import asc, desc, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.app.error import (
     ProviderModelCurrentlyNotSupportError,
     ProviderNotInitializeError,
@@ -54,7 +53,6 @@ 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
 
@@ -100,12 +98,7 @@ class DocumentResource(Resource):
         return documents
 
 
-@console_ns.route("/datasets/process-rule")
 class GetProcessRuleApi(Resource):
-    @api.doc("get_process_rule")
-    @api.doc(description="Get dataset document processing rules")
-    @api.doc(params={"document_id": "Document ID (optional)"})
-    @api.response(200, "Process rules retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -147,21 +140,7 @@ class GetProcessRuleApi(Resource):
         return {"mode": mode, "rules": rules, "limits": limits}
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents")
 class DatasetDocumentListApi(Resource):
-    @api.doc("get_dataset_documents")
-    @api.doc(description="Get documents in a dataset")
-    @api.doc(
-        params={
-            "dataset_id": "Dataset ID",
-            "page": "Page number (default: 1)",
-            "limit": "Number of items per page (default: 20)",
-            "keyword": "Search keyword",
-            "sort": "Sort order (default: -created_at)",
-            "fetch": "Fetch full details (default: false)",
-        }
-    )
-    @api.response(200, "Documents retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -345,23 +324,7 @@ class DatasetDocumentListApi(Resource):
         return {"result": "success"}, 204
 
 
-@console_ns.route("/datasets/init")
 class DatasetInitApi(Resource):
-    @api.doc("init_dataset")
-    @api.doc(description="Initialize dataset with documents")
-    @api.expect(
-        api.model(
-            "DatasetInitRequest",
-            {
-                "upload_file_id": fields.String(required=True, description="Upload file ID"),
-                "indexing_technique": fields.String(description="Indexing technique"),
-                "process_rule": fields.Raw(description="Processing rules"),
-                "data_source": fields.Raw(description="Data source configuration"),
-            },
-        )
-    )
-    @api.response(201, "Dataset initialized successfully", dataset_and_document_fields)
-    @api.response(400, "Invalid request parameters")
     @setup_required
     @login_required
     @account_initialization_required
@@ -431,14 +394,7 @@ class DatasetInitApi(Resource):
         return response
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/indexing-estimate")
 class DocumentIndexingEstimateApi(DocumentResource):
-    @api.doc("estimate_document_indexing")
-    @api.doc(description="Estimate document indexing cost")
-    @api.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
-    @api.response(200, "Indexing estimate calculated successfully")
-    @api.response(404, "Document not found")
-    @api.response(400, "Document already finished")
     @setup_required
     @login_required
     @account_initialization_required
@@ -501,7 +457,6 @@ class DocumentIndexingEstimateApi(DocumentResource):
         return response, 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-estimate")
 class DocumentBatchIndexingEstimateApi(DocumentResource):
     @setup_required
     @login_required
@@ -544,7 +499,6 @@ 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"],
@@ -595,7 +549,6 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 raise IndexingEstimateError(str(e))
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-status")
 class DocumentBatchIndexingStatusApi(DocumentResource):
     @setup_required
     @login_required
@@ -640,13 +593,7 @@ class DocumentBatchIndexingStatusApi(DocumentResource):
         return data
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/indexing-status")
 class DocumentIndexingStatusApi(DocumentResource):
-    @api.doc("get_document_indexing_status")
-    @api.doc(description="Get document indexing status")
-    @api.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
-    @api.response(200, "Indexing status retrieved successfully")
-    @api.response(404, "Document not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -688,21 +635,9 @@ class DocumentIndexingStatusApi(DocumentResource):
         return marshal(document_dict, document_status_fields)
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 class DocumentApi(DocumentResource):
     METADATA_CHOICES = {"all", "only", "without"}
 
-    @api.doc("get_document")
-    @api.doc(description="Get document details")
-    @api.doc(
-        params={
-            "dataset_id": "Dataset ID",
-            "document_id": "Document ID",
-            "metadata": "Metadata inclusion (all/only/without)",
-        }
-    )
-    @api.response(200, "Document retrieved successfully")
-    @api.response(404, "Document not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -719,7 +654,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() if document.dataset_process_rule else {}
+            document_process_rules = document.dataset_process_rule.to_dict()
             data_source_info = document.data_source_detail_dict
             response = {
                 "id": document.id,
@@ -811,16 +746,7 @@ class DocumentApi(DocumentResource):
         return {"result": "success"}, 204
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/<string:action>")
 class DocumentProcessingApi(DocumentResource):
-    @api.doc("update_document_processing")
-    @api.doc(description="Update document processing status (pause/resume)")
-    @api.doc(
-        params={"dataset_id": "Dataset ID", "document_id": "Document ID", "action": "Action to perform (pause/resume)"}
-    )
-    @api.response(200, "Processing status updated successfully")
-    @api.response(404, "Document not found")
-    @api.response(400, "Invalid action")
     @setup_required
     @login_required
     @account_initialization_required
@@ -855,23 +781,7 @@ class DocumentProcessingApi(DocumentResource):
         return {"result": "success"}, 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/metadata")
 class DocumentMetadataApi(DocumentResource):
-    @api.doc("update_document_metadata")
-    @api.doc(description="Update document metadata")
-    @api.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
-    @api.expect(
-        api.model(
-            "UpdateDocumentMetadataRequest",
-            {
-                "doc_type": fields.String(description="Document type"),
-                "doc_metadata": fields.Raw(description="Document metadata"),
-            },
-        )
-    )
-    @api.response(200, "Document metadata updated successfully")
-    @api.response(404, "Document not found")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -915,7 +825,6 @@ class DocumentMetadataApi(DocumentResource):
         return {"result": "success", "message": "Document metadata updated."}, 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/status/<string:action>/batch")
 class DocumentStatusApi(DocumentResource):
     @setup_required
     @login_required
@@ -952,7 +861,6 @@ class DocumentStatusApi(DocumentResource):
         return {"result": "success"}, 200
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/pause")
 class DocumentPauseApi(DocumentResource):
     @setup_required
     @login_required
@@ -986,7 +894,6 @@ class DocumentPauseApi(DocumentResource):
         return {"result": "success"}, 204
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/resume")
 class DocumentRecoverApi(DocumentResource):
     @setup_required
     @login_required
@@ -1017,7 +924,6 @@ class DocumentRecoverApi(DocumentResource):
         return {"result": "success"}, 204
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/retry")
 class DocumentRetryApi(DocumentResource):
     @setup_required
     @login_required
@@ -1061,7 +967,6 @@ class DocumentRetryApi(DocumentResource):
         return {"result": "success"}, 204
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/rename")
 class DocumentRenameApi(DocumentResource):
     @setup_required
     @login_required
@@ -1085,7 +990,6 @@ class DocumentRenameApi(DocumentResource):
         return document
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/website-sync")
 class WebsiteDocumentSyncApi(DocumentResource):
     @setup_required
     @login_required
@@ -1113,41 +1017,6 @@ 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")
@@ -1169,6 +1038,3 @@ 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,9 +71,3 @@ 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/external.py

@@ -1,10 +1,10 @@
 from flask import request
 from flask_login import current_user
-from flask_restx import Resource, fields, marshal, reqparse
+from flask_restx import Resource, marshal, reqparse
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.datasets.error import DatasetNameDuplicateError
 from controllers.console.wraps import account_initialization_required, setup_required
 from fields.dataset_fields import dataset_detail_fields
@@ -21,18 +21,7 @@ def _validate_name(name):
     return name
 
 
-@console_ns.route("/datasets/external-knowledge-api")
 class ExternalApiTemplateListApi(Resource):
-    @api.doc("get_external_api_templates")
-    @api.doc(description="Get external knowledge API templates")
-    @api.doc(
-        params={
-            "page": "Page number (default: 1)",
-            "limit": "Number of items per page (default: 20)",
-            "keyword": "Search keyword",
-        }
-    )
-    @api.response(200, "External API templates retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -90,13 +79,7 @@ class ExternalApiTemplateListApi(Resource):
         return external_knowledge_api.to_dict(), 201
 
 
-@console_ns.route("/datasets/external-knowledge-api/<uuid:external_knowledge_api_id>")
 class ExternalApiTemplateApi(Resource):
-    @api.doc("get_external_api_template")
-    @api.doc(description="Get external knowledge API template details")
-    @api.doc(params={"external_knowledge_api_id": "External knowledge API ID"})
-    @api.response(200, "External API template retrieved successfully")
-    @api.response(404, "Template not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -148,19 +131,14 @@ class ExternalApiTemplateApi(Resource):
         external_knowledge_api_id = str(external_knowledge_api_id)
 
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not (current_user.is_editor or current_user.is_dataset_operator):
+        if not current_user.is_editor or current_user.is_dataset_operator:
             raise Forbidden()
 
         ExternalDatasetService.delete_external_knowledge_api(current_user.current_tenant_id, external_knowledge_api_id)
         return {"result": "success"}, 204
 
 
-@console_ns.route("/datasets/external-knowledge-api/<uuid:external_knowledge_api_id>/use-check")
 class ExternalApiUseCheckApi(Resource):
-    @api.doc("check_external_api_usage")
-    @api.doc(description="Check if external knowledge API is being used")
-    @api.doc(params={"external_knowledge_api_id": "External knowledge API ID"})
-    @api.response(200, "Usage check completed successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -173,24 +151,7 @@ class ExternalApiUseCheckApi(Resource):
         return {"is_using": external_knowledge_api_is_using, "count": count}, 200
 
 
-@console_ns.route("/datasets/external")
 class ExternalDatasetCreateApi(Resource):
-    @api.doc("create_external_dataset")
-    @api.doc(description="Create external knowledge dataset")
-    @api.expect(
-        api.model(
-            "CreateExternalDatasetRequest",
-            {
-                "external_knowledge_api_id": fields.String(required=True, description="External knowledge API ID"),
-                "external_knowledge_id": fields.String(required=True, description="External knowledge ID"),
-                "name": fields.String(required=True, description="Dataset name"),
-                "description": fields.String(description="Dataset description"),
-            },
-        )
-    )
-    @api.response(201, "External dataset created successfully", dataset_detail_fields)
-    @api.response(400, "Invalid parameters")
-    @api.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
@@ -230,24 +191,7 @@ class ExternalDatasetCreateApi(Resource):
         return marshal(dataset, dataset_detail_fields), 201
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/external-hit-testing")
 class ExternalKnowledgeHitTestingApi(Resource):
-    @api.doc("test_external_knowledge_retrieval")
-    @api.doc(description="Test external knowledge retrieval for dataset")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.expect(
-        api.model(
-            "ExternalHitTestingRequest",
-            {
-                "query": fields.String(required=True, description="Query text for testing"),
-                "retrieval_model": fields.Raw(description="Retrieval model configuration"),
-                "external_retrieval_model": fields.Raw(description="External retrieval model configuration"),
-            },
-        )
-    )
-    @api.response(200, "External hit testing completed successfully")
-    @api.response(404, "Dataset not found")
-    @api.response(400, "Invalid parameters")
     @setup_required
     @login_required
     @account_initialization_required
@@ -284,22 +228,8 @@ class ExternalKnowledgeHitTestingApi(Resource):
             raise InternalServerError(str(e))
 
 
-@console_ns.route("/test/retrieval")
 class BedrockRetrievalApi(Resource):
     # this api is only for internal testing
-    @api.doc("bedrock_retrieval_test")
-    @api.doc(description="Bedrock retrieval test (internal use only)")
-    @api.expect(
-        api.model(
-            "BedrockRetrievalTestRequest",
-            {
-                "retrieval_setting": fields.Raw(required=True, description="Retrieval settings"),
-                "query": fields.String(required=True, description="Query text"),
-                "knowledge_id": fields.String(required=True, description="Knowledge ID"),
-            },
-        )
-    )
-    @api.response(200, "Bedrock retrieval test completed")
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("retrieval_setting", nullable=False, required=True, type=dict, location="json")
@@ -317,3 +247,12 @@ class BedrockRetrievalApi(Resource):
             args["retrieval_setting"], args["query"], args["knowledge_id"]
         )
         return result, 200
+
+
+api.add_resource(ExternalKnowledgeHitTestingApi, "/datasets/<uuid:dataset_id>/external-hit-testing")
+api.add_resource(ExternalDatasetCreateApi, "/datasets/external")
+api.add_resource(ExternalApiTemplateListApi, "/datasets/external-knowledge-api")
+api.add_resource(ExternalApiTemplateApi, "/datasets/external-knowledge-api/<uuid:external_knowledge_api_id>")
+api.add_resource(ExternalApiUseCheckApi, "/datasets/external-knowledge-api/<uuid:external_knowledge_api_id>/use-check")
+# this api is only for internal test
+api.add_resource(BedrockRetrievalApi, "/test/retrieval")

api/controllers/console/datasets/hit_testing.py

@@ -1,6 +1,6 @@
-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.datasets.hit_testing_base import DatasetsHitTestingBase
 from controllers.console.wraps import (
     account_initialization_required,
@@ -10,25 +10,7 @@ from controllers.console.wraps import (
 from libs.login import login_required
 
 
-@console_ns.route("/datasets/<uuid:dataset_id>/hit-testing")
 class HitTestingApi(Resource, DatasetsHitTestingBase):
-    @api.doc("test_dataset_retrieval")
-    @api.doc(description="Test dataset knowledge retrieval")
-    @api.doc(params={"dataset_id": "Dataset ID"})
-    @api.expect(
-        api.model(
-            "HitTestingRequest",
-            {
-                "query": fields.String(required=True, description="Query text for testing"),
-                "retrieval_model": fields.Raw(description="Retrieval model configuration"),
-                "top_k": fields.Integer(description="Number of top results to return"),
-                "score_threshold": fields.Float(description="Score threshold for filtering results"),
-            },
-        )
-    )
-    @api.response(200, "Hit testing completed successfully")
-    @api.response(404, "Dataset not found")
-    @api.response(400, "Invalid parameters")
     @setup_required
     @login_required
     @account_initialization_required
@@ -41,3 +23,6 @@ class HitTestingApi(Resource, DatasetsHitTestingBase):
         self.hit_testing_args_check(args)
 
         return self.perform_hit_testing(dataset, args)
+
+
+api.add_resource(HitTestingApi, "/datasets/<uuid:dataset_id>/hit-testing")

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

@@ -1,362 +0,0 @@
-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

@@ -1,57 +0,0 @@
-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

@@ -1,164 +0,0 @@
-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).where(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

@@ -1,114 +0,0 @@
-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

@@ -1,389 +0,0 @@
-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

@@ -1,147 +0,0 @@
-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/website.py

@@ -1,32 +1,13 @@
-from flask_restx import Resource, fields, reqparse
+from flask_restx import Resource, reqparse
 
-from controllers.console import api, console_ns
+from controllers.console import api
 from controllers.console.datasets.error import WebsiteCrawlError
 from controllers.console.wraps import account_initialization_required, setup_required
 from libs.login import login_required
 from services.website_service import WebsiteCrawlApiRequest, WebsiteCrawlStatusApiRequest, WebsiteService
 
 
-@console_ns.route("/website/crawl")
 class WebsiteCrawlApi(Resource):
-    @api.doc("crawl_website")
-    @api.doc(description="Crawl website content")
-    @api.expect(
-        api.model(
-            "WebsiteCrawlRequest",
-            {
-                "provider": fields.String(
-                    required=True,
-                    description="Crawl provider (firecrawl/watercrawl/jinareader)",
-                    enum=["firecrawl", "watercrawl", "jinareader"],
-                ),
-                "url": fields.String(required=True, description="URL to crawl"),
-                "options": fields.Raw(required=True, description="Crawl options"),
-            },
-        )
-    )
-    @api.response(200, "Website crawl initiated successfully")
-    @api.response(400, "Invalid crawl parameters")
     @setup_required
     @login_required
     @account_initialization_required
@@ -58,14 +39,7 @@ class WebsiteCrawlApi(Resource):
         return result, 200
 
 
-@console_ns.route("/website/crawl/status/<string:job_id>")
 class WebsiteCrawlStatusApi(Resource):
-    @api.doc("get_crawl_status")
-    @api.doc(description="Get website crawl status")
-    @api.doc(params={"job_id": "Crawl job ID", "provider": "Crawl provider (firecrawl/watercrawl/jinareader)"})
-    @api.response(200, "Crawl status retrieved successfully")
-    @api.response(404, "Crawl job not found")
-    @api.response(400, "Invalid provider")
     @setup_required
     @login_required
     @account_initialization_required
@@ -88,3 +62,7 @@ class WebsiteCrawlStatusApi(Resource):
         except Exception as e:
             raise WebsiteCrawlError(str(e))
         return result, 200
+
+
+api.add_resource(WebsiteCrawlApi, "/website/crawl")
+api.add_resource(WebsiteCrawlStatusApi, "/website/crawl/status/<string:job_id>")

api/controllers/console/datasets/wraps.py

@@ -1,46 +0,0 @@
-from collections.abc import Callable
-from functools import wraps
-
-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: Callable | None = 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)
-                .where(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/workflow.py

@@ -20,7 +20,6 @@ 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
@@ -83,11 +82,6 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
             raise NotWorkflowAppError()
         assert current_user is not None
 
-        # 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)
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.EXPLORE, current_user.id)
 
         return {"result": "success"}

api/controllers/console/explore/wraps.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import Concatenate, ParamSpec, TypeVar
+from typing import Concatenate, Optional, ParamSpec, TypeVar
 
 from flask_login import current_user
 from flask_restx import Resource
@@ -20,7 +20,7 @@ R = TypeVar("R")
 T = TypeVar("T")
 
 
-def installed_app_required(view: Callable[Concatenate[InstalledApp, P], R] | None = None):
+def installed_app_required(view: Optional[Callable[Concatenate[InstalledApp, P], R]] = None):
     def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
         def decorated(installed_app_id: str, *args: P.args, **kwargs: P.kwargs):
@@ -50,7 +50,7 @@ def installed_app_required(view: Callable[Concatenate[InstalledApp, P], R] | Non
     return decorator
 
 
-def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] | None = None):
+def user_allowed_to_access_app(view: Optional[Callable[Concatenate[InstalledApp, P], R]] = None):
     def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
         def decorated(installed_app: InstalledApp, *args: P.args, **kwargs: P.kwargs):

api/controllers/console/files.py

@@ -20,7 +20,6 @@ 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
@@ -69,11 +68,10 @@ class FileApi(Resource):
         if source not in ("datasets", None):
             source = None
 
-        if not isinstance(current_user, Account):
-            raise ValueError("Invalid user account")
-
         try:
-            upload_file = FileService(db.engine).upload_file(
+            if not isinstance(current_user, Account):
+                raise ValueError("Invalid user account")
+            upload_file = FileService.upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,
@@ -94,7 +92,7 @@ class FilePreviewApi(Resource):
     @account_initialization_required
     def get(self, file_id):
         file_id = str(file_id)
-        text = FileService(db.engine).get_file_preview(file_id)
+        text = FileService.get_file_preview(file_id)
         return {"content": text}
 
 

api/controllers/console/remote_files.py

@@ -14,7 +14,6 @@ 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
@@ -62,7 +61,7 @@ class RemoteFileUploadApi(Resource):
 
         try:
             user = cast(Account, current_user)
-            upload_file = FileService(db.engine).upload_file(
+            upload_file = FileService.upload_file(
                 filename=file_info.filename,
                 content=content,
                 mimetype=file_info.mimetype,

api/controllers/console/spec.py

@@ -1,35 +0,0 @@
-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/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

api/controllers/console/workspace/workspace.py

@@ -227,7 +227,7 @@ class WebappLogoWorkspaceApi(Resource):
             raise UnsupportedFileTypeError()
 
         try:
-            upload_file = FileService(db.engine).upload_file(
+            upload_file = FileService.upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,

api/controllers/console/wraps.py

@@ -279,14 +279,3 @@ 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/image_preview.py

@@ -7,7 +7,6 @@ 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
 
@@ -29,7 +28,7 @@ class ImagePreviewApi(Resource):
             return {"content": "Invalid request."}, 400
 
         try:
-            generator, mimetype = FileService(db.engine).get_image_preview(
+            generator, mimetype = FileService.get_image_preview(
                 file_id=file_id,
                 timestamp=timestamp,
                 nonce=nonce,
@@ -58,7 +57,7 @@ class FilePreviewApi(Resource):
             return {"content": "Invalid request."}, 400
 
         try:
-            generator, upload_file = FileService(db.engine).get_file_generator_by_file_id(
+            generator, upload_file = FileService.get_file_generator_by_file_id(
                 file_id=file_id,
                 timestamp=args["timestamp"],
                 nonce=args["nonce"],
@@ -109,7 +108,7 @@ class WorkspaceWebappLogoApi(Resource):
             raise NotFound("webapp logo is not found")
 
         try:
-            generator, mimetype = FileService(db.engine).get_public_image_preview(
+            generator, mimetype = FileService.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 extensions.ext_database import db as global_db
+from models import db as global_db
 
 
 @files_ns.route("/tools/<uuid:file_id>.<string:extension>")

api/controllers/files/upload.py

@@ -1,4 +1,5 @@
 from mimetypes import guess_extension
+from typing import Optional
 
 from flask_restx import Resource, reqparse
 from flask_restx.api import HTTPStatus
@@ -72,11 +73,11 @@ class PluginUploadFileApi(Resource):
         nonce: str = args["nonce"]
         sign: str = args["sign"]
         tenant_id: str = args["tenant_id"]
-        user_id: str | None = args.get("user_id")
+        user_id: Optional[str] = args.get("user_id")
         user = get_user(tenant_id, user_id)
 
-        filename: str | None = file.filename
-        mimetype: str | None = file.mimetype
+        filename: Optional[str] = file.filename
+        mimetype: Optional[str] = file.mimetype
 
         if not filename or not mimetype:
             raise Forbidden("Invalid request.")

api/controllers/inner_api/plugin/plugin.py

@@ -420,12 +420,7 @@ class PluginUploadFileRequestApi(Resource):
     )
     def post(self, user_model: Account | EndUser, tenant_model: Tenant, payload: RequestRequestUploadFile):
         # generate signed url
-        url = get_signed_file_url_for_plugin(
-            filename=payload.filename,
-            mimetype=payload.mimetype,
-            tenant_id=tenant_model.id,
-            user_id=user_model.id,
-        )
+        url = get_signed_file_url_for_plugin(payload.filename, payload.mimetype, tenant_model.id, user_model.id)
         return BaseBackwardsInvocationResponse(data={"url": url}).model_dump()
 
 

api/controllers/inner_api/plugin/wraps.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import ParamSpec, TypeVar, cast
+from typing import Optional, ParamSpec, TypeVar, cast
 
 from flask import current_app, request
 from flask_login import user_logged_in
@@ -32,20 +32,11 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
             user_model = (
                 session.query(EndUser)
                 .where(
-                    EndUser.id == user_id,
+                    EndUser.session_id == user_id,
                     EndUser.tenant_id == tenant_id,
                 )
                 .first()
             )
-            if not user_model:
-                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,
@@ -63,7 +54,7 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
     return user_model
 
 
-def get_user_tenant(view: Callable[P, R] | None = None):
+def get_user_tenant(view: Optional[Callable[P, R]] = None):
     def decorator(view_func: Callable[P, R]):
         @wraps(view_func)
         def decorated_view(*args: P.args, **kwargs: P.kwargs):
@@ -115,7 +106,7 @@ def get_user_tenant(view: Callable[P, R] | None = None):
         return decorator(view)
 
 
-def plugin_data(view: Callable[P, R] | None = None, *, payload_type: type[BaseModel]):
+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):
             try: