fix: allow web app relogin when token expired

.devcontainer/post_create_command.sh

@@ -5,7 +5,7 @@ cd web && pnpm install
 pipx install uv
 
 echo 'alias start-api="cd /workspaces/dify/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug"' >> ~/.bashrc
-echo 'alias start-worker="cd /workspaces/dify/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage"' >> ~/.bashrc
+echo 'alias start-worker="cd /workspaces/dify/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion"' >> ~/.bashrc
 echo 'alias start-web="cd /workspaces/dify/web && pnpm dev"' >> ~/.bashrc
 echo 'alias start-web-prod="cd /workspaces/dify/web && pnpm build && pnpm start"' >> ~/.bashrc
 echo 'alias start-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d"' >> ~/.bashrc

.github/actions/setup-uv/action.yml

@@ -8,7 +8,7 @@ inputs:
   uv-version:
     description: UV version to set up
     required: true
-    default: '0.8.9'
+    default: '~=0.7.11'
   uv-lockfile:
     description: Path to the UV lockfile to restore cache from
     required: true
@@ -26,7 +26,7 @@ runs:
         python-version: ${{ inputs.python-version }}
 
     - name: Install uv
-      uses: astral-sh/setup-uv@v6
+      uses: astral-sh/setup-uv@v5
       with:
         version: ${{ inputs.uv-version }}
         python-version: ${{ inputs.python-version }}

.github/workflows/build-push.yml

@@ -8,7 +8,6 @@ on:
       - "deploy/enterprise"
       - "build/**"
       - "release/e-*"
-      - "deploy/rag-dev"
     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/[email protected]
         with:
-          host: ${{ secrets.RAG_SSH_HOST }}
+          host: ${{ secrets.SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.github/workflows/style.yml

@@ -82,7 +82,7 @@ jobs:
       - name: Install pnpm
         uses: pnpm/action-setup@v4
         with:
-          package_json_file: web/package.json
+          version: 10
           run_install: false
 
       - name: Setup NodeJS
@@ -95,12 +95,10 @@ jobs:
 
       - name: Web dependencies
         if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
       - name: Web style check
         if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: ./web
         run: pnpm run lint
 
   docker-compose-template:

.github/workflows/translate-i18n-base-on-english.yml

@@ -46,7 +46,7 @@ jobs:
       - name: Install pnpm
         uses: pnpm/action-setup@v4
         with:
-          package_json_file: web/package.json
+          version: 10
           run_install: false
 
       - name: Set up Node.js
@@ -59,12 +59,10 @@ jobs:
 
       - name: Install dependencies
         if: env.FILES_CHANGED == 'true'
-        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
       - name: Generate i18n translations
         if: env.FILES_CHANGED == 'true'
-        working-directory: ./web
         run: pnpm run auto-gen-i18n ${{ env.FILE_ARGS }}
 
       - name: Create Pull Request

.github/workflows/web-tests.yml

@@ -35,7 +35,7 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         uses: pnpm/action-setup@v4
         with:
-          package_json_file: web/package.json
+          version: 10
           run_install: false
 
       - name: Setup Node.js
@@ -48,10 +48,8 @@ jobs:
 
       - name: Install dependencies
         if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
       - name: Run tests
         if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: ./web
         run: pnpm test

CLAUDE.md

@@ -1,83 +0,0 @@
-# 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 --project api mypy .                 # 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/`
-2. Add service logic in `/api/services/`
-3. Update routes in controller's `__init__.py`
-4. Write tests in `/api/tests/`
-
-## Project-Specific Conventions
-
-- All async tasks use Celery with Redis as broker

README.md

@@ -225,8 +225,7 @@ Deploy Dify to AWS with [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS
 
-- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Using Alibaba Cloud Computing Nest
 

README_AR.md

@@ -208,8 +208,7 @@ docker compose up -d
 
 ##### AWS
 
-- [AWS CDK بواسطة @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK بواسطة @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK بواسطة @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### استخدام Alibaba Cloud للنشر
  [بسرعة نشر Dify إلى سحابة علي بابا مع عش الحوسبة السحابية علي بابا](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

README_BN.md

@@ -225,8 +225,7 @@ GitHub-এ ডিফাইকে স্টার দিয়ে রাখুন
 
 ##### AWS
 
-- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud ব্যবহার করে ডিপ্লয়
 

README_CN.md

@@ -223,8 +223,7 @@ docker compose up -d
 使用 [CDK](https://aws.amazon.com/cdk/) 将 Dify 部署到 AWS
 
 ##### AWS 
-- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### 使用 阿里云计算巢 部署
 

README_DE.md

@@ -220,8 +220,7 @@ Stellen Sie Dify mit nur einem Klick mithilfe von [terraform](https://www.terraf
 Bereitstellung von Dify auf AWS mit [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud 
 

README_ES.md

@@ -220,8 +220,7 @@ Despliega Dify en una plataforma en la nube con un solo clic utilizando [terrafo
 Despliegue Dify en AWS usando [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK por @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK por @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK por @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud
 

README_FR.md

@@ -218,8 +218,7 @@ Déployez Dify sur une plateforme cloud en un clic en utilisant [terraform](http
 Déployez Dify sur AWS en utilisant [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK par @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK par @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK par @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud
 

README_JA.md

@@ -219,8 +219,7 @@ docker compose up -d
 [CDK](https://aws.amazon.com/cdk/) を使用して、DifyをAWSにデプロイします
 
 ##### AWS
-- [@KevinZhaoによるAWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [@tmokmssによるAWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [@KevinZhaoによるAWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud
 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

README_KL.md

@@ -218,8 +218,7 @@ wa'logh nIqHom neH ghun deployment toy'wI' [terraform](https://www.terraform.io/
 wa'logh nIqHom neH ghun deployment toy'wI' [CDK](https://aws.amazon.com/cdk/) lo'laH.
 
 ##### AWS 
-- [AWS CDK qachlot @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK qachlot @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK qachlot @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud
 

README_KR.md

@@ -212,8 +212,7 @@ Dify를 Kubernetes에 배포하고 프리미엄 스케일링 설정을 구성했
 [CDK](https://aws.amazon.com/cdk/)를 사용하여 AWS에 Dify 배포
 
 ##### AWS 
-- [KevinZhao의 AWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [tmokmss의 AWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [KevinZhao의 AWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud
 

README_PT.md

@@ -217,8 +217,7 @@ Implante o Dify na Plataforma Cloud com um único clique usando [terraform](http
 Implante o Dify na AWS usando [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK por @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK por @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK por @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud
 

README_SI.md

@@ -218,8 +218,7 @@ namestite Dify v Cloud Platform z enim klikom z uporabo [terraform](https://www.
 Uvedite Dify v AWS z uporabo [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud
 

README_TR.md

@@ -211,8 +211,7 @@ Dify'ı bulut platformuna tek tıklamayla dağıtın [terraform](https://www.ter
 [CDK](https://aws.amazon.com/cdk/) kullanarak Dify'ı AWS'ye dağıtın
 
 ##### AWS 
-- [AWS CDK tarafından @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK tarafından @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK tarafından @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### Alibaba Cloud
 

README_TW.md

@@ -223,8 +223,7 @@ Dify 的所有功能都提供相應的 API，因此您可以輕鬆地將 Dify
 
 ### AWS
 
-- [由 @KevinZhao 提供的 AWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [由 @tmokmss 提供的 AWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [由 @KevinZhao 提供的 AWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 #### 使用 阿里云计算巢進行部署
 

README_VI.md

@@ -213,8 +213,7 @@ Triển khai Dify lên nền tảng đám mây với một cú nhấp chuột b
 Triển khai Dify trên AWS bằng [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK bởi @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
-- [AWS CDK bởi @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
+- [AWS CDK bởi @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
 
 
 #### Alibaba Cloud

api/Dockerfile

@@ -4,7 +4,7 @@ FROM python:3.12-slim-bookworm AS base
 WORKDIR /app/api
 
 # Install uv
-ENV UV_VERSION=0.8.9
+ENV UV_VERSION=0.7.11
 
 RUN pip install --no-cache-dir uv==${UV_VERSION}
 

api/README.md

@@ -74,7 +74,7 @@
 10. If you need to handle and debug the async tasks (e.g. dataset importing and documents indexing), please start the worker service.
 
    ```bash
-   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage
+   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin
    ```
 
    Addition, if you want to debug the celery scheduled tasks, you can use the following command in another terminal:

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/app_factory.py

@@ -51,7 +51,6 @@ def initialize_extensions(app: DifyApp):
         ext_login,
         ext_mail,
         ext_migrate,
-        ext_orjson,
         ext_otel,
         ext_proxy_fix,
         ext_redis,
@@ -68,7 +67,6 @@ def initialize_extensions(app: DifyApp):
         ext_logging,
         ext_warnings,
         ext_import_modules,
-        ext_orjson,
         ext_set_secretkey,
         ext_compress,
         ext_code_based_extension,

api/commands.py

@@ -13,14 +13,11 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from constants.languages import languages
-from core.helper import encrypter
-from core.plugin.entities.plugin import DatasourceProviderID, PluginInstallationSource, ToolProviderID
-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
@@ -33,15 +30,12 @@ from models import Tenant
 from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, DatasetMetadataBinding, DocumentSegment
 from models.dataset import Document as DatasetDocument
 from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation
-from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
 from models.provider import Provider, ProviderModel
-from models.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 tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
 
 @click.command("reset-password", help="Reset the account password.")
@@ -1208,363 +1202,3 @@ def setup_system_tool_oauth_client(provider, client_params):
     db.session.add(oauth_client)
     db.session.commit()
     click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
-
-
-def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
-    """
-    Find draft variables that reference non-existent apps.
-
-    Args:
-        batch_size: Maximum number of orphaned app IDs to return
-
-    Returns:
-        List of app IDs that have draft variables but don't exist in the apps table
-    """
-    query = """
-        SELECT DISTINCT wdv.app_id
-        FROM workflow_draft_variables AS wdv
-        WHERE NOT EXISTS(
-            SELECT 1 FROM apps WHERE apps.id = wdv.app_id
-        )
-        LIMIT :batch_size
-    """
-
-    with db.engine.connect() as conn:
-        result = conn.execute(sa.text(query), {"batch_size": batch_size})
-        return [row[0] for row in result]
-
-
-def _count_orphaned_draft_variables() -> dict[str, Any]:
-    """
-    Count orphaned draft variables by app.
-
-    Returns:
-        Dictionary with statistics about orphaned variables
-    """
-    query = """
-        SELECT
-            wdv.app_id,
-            COUNT(*) as variable_count
-        FROM workflow_draft_variables AS wdv
-        WHERE NOT EXISTS(
-            SELECT 1 FROM apps WHERE apps.id = wdv.app_id
-        )
-        GROUP BY wdv.app_id
-        ORDER BY variable_count DESC
-    """
-
-    with db.engine.connect() as conn:
-        result = conn.execute(sa.text(query))
-        orphaned_by_app = {row[0]: row[1] for row in result}
-
-        total_orphaned = sum(orphaned_by_app.values())
-        app_count = len(orphaned_by_app)
-
-        return {
-            "total_orphaned_variables": total_orphaned,
-            "orphaned_app_count": app_count,
-            "orphaned_by_app": orphaned_by_app,
-        }
-
-
-@click.command()
-@click.option("--dry-run", is_flag=True, help="Show what would be deleted without actually deleting")
-@click.option("--batch-size", default=1000, help="Number of records to process per batch (default 1000)")
-@click.option("--max-apps", default=None, type=int, help="Maximum number of apps to process (default: no limit)")
-@click.option("-f", "--force", is_flag=True, help="Skip user confirmation and force the command to execute.")
-def cleanup_orphaned_draft_variables(
-    dry_run: bool,
-    batch_size: int,
-    max_apps: int | None,
-    force: bool = False,
-):
-    """
-    Clean up orphaned draft variables from the database.
-
-    This script finds and removes draft variables that belong to apps
-    that no longer exist in the database.
-    """
-    logger = logging.getLogger(__name__)
-
-    # Get statistics
-    stats = _count_orphaned_draft_variables()
-
-    logger.info("Found %s orphaned draft variables", stats["total_orphaned_variables"])
-    logger.info("Across %s non-existent apps", stats["orphaned_app_count"])
-
-    if stats["total_orphaned_variables"] == 0:
-        logger.info("No orphaned draft variables found. Exiting.")
-        return
-
-    if dry_run:
-        logger.info("DRY RUN: Would delete the following:")
-        for app_id, count in sorted(stats["orphaned_by_app"].items(), key=lambda x: x[1], reverse=True)[
-            :10
-        ]:  # Show top 10
-            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
-
-    # Confirm deletion
-    if not force:
-        click.confirm(
-            f"Are you sure you want to delete {stats['total_orphaned_variables']} "
-            f"orphaned draft variables from {stats['orphaned_app_count']} apps?",
-            abort=True,
-        )
-
-    total_deleted = 0
-    processed_apps = 0
-
-    while True:
-        if max_apps and processed_apps >= max_apps:
-            logger.info("Reached maximum app limit (%s). Stopping.", max_apps)
-            break
-
-        orphaned_app_ids = _find_orphaned_draft_variables(batch_size=10)
-        if not orphaned_app_ids:
-            logger.info("No more orphaned draft variables found.")
-            break
-
-        for app_id in orphaned_app_ids:
-            if max_apps and processed_apps >= max_apps:
-                break
-
-            try:
-                deleted_count = delete_draft_variables_batch(app_id, batch_size)
-                total_deleted += deleted_count
-                processed_apps += 1
-
-                logger.info("Deleted %s variables for app %s", deleted_count, app_id)
-
-            except Exception:
-                logger.exception("Error processing app %s", app_id)
-                continue
-
-    logger.info("Cleanup completed. Total deleted: %s variables across %s apps", total_deleted, processed_apps)
-
-
-@click.command("setup-datasource-oauth-client", help="Setup datasource oauth client.")
-@click.option("--provider", prompt=True, help="Provider name")
-@click.option("--client-params", prompt=True, help="Client Params")
-def setup_datasource_oauth_client(provider, client_params):
-    """
-    Setup datasource oauth client
-    """
-    provider_id = DatasourceProviderID(provider)
-    provider_name = provider_id.provider_name
-    plugin_id = provider_id.plugin_id
-
-    try:
-        # json validate
-        click.echo(click.style(f"Validating client params: {client_params}", fg="yellow"))
-        client_params_dict = TypeAdapter(dict[str, Any]).validate_json(client_params)
-        click.echo(click.style("Client params validated successfully.", fg="green"))
-    except Exception as e:
-        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
-        return
-
-    click.echo(click.style(f"Ready to delete existing oauth client params: {provider_name}", fg="yellow"))
-    deleted_count = (
-        db.session.query(DatasourceOauthParamConfig)
-        .filter_by(
-            provider=provider_name,
-            plugin_id=plugin_id,
-        )
-        .delete()
-    )
-    if deleted_count > 0:
-        click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
-
-    click.echo(click.style(f"Ready to setup datasource oauth client: {provider_name}", fg="yellow"))
-    oauth_client = DatasourceOauthParamConfig(
-        provider=provider_name,
-        plugin_id=plugin_id,
-        system_credentials=client_params_dict,
-    )
-    db.session.add(oauth_client)
-    db.session.commit()
-    click.echo(click.style(f"provider: {provider_name}", fg="green"))
-    click.echo(click.style(f"plugin_id: {plugin_id}", fg="green"))
-    click.echo(click.style(f"params: {json.dumps(client_params_dict, indent=2, ensure_ascii=False)}", fg="green"))
-    click.echo(click.style(f"Datasource oauth client setup successfully. id: {oauth_client.id}", fg="green"))
-
-
-@click.command("transform-datasource-credentials", help="Transform datasource credentials.")
-def transform_datasource_credentials():
-    """
-    Transform datasource credentials
-    """
-    try:
-        installer_manager = PluginInstaller()
-        plugin_migration = PluginMigration()
-
-        notion_plugin_id = "langgenius/notion_datasource"
-        firecrawl_plugin_id = "langgenius/firecrawl_datasource"
-        jina_plugin_id = "langgenius/jina_datasource"
-        notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)
-        firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)
-        jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)
-        oauth_credential_type = CredentialType.OAUTH2
-        api_key_credential_type = CredentialType.API_KEY
-
-        # deal notion credentials
-        deal_notion_count = 0
-        notion_credentials = db.session.query(DataSourceOauthBinding).filter_by(provider="notion").all()
-        notion_credentials_tenant_mapping: dict[str, list[DataSourceOauthBinding]] = {}
-        for credential in notion_credentials:
-            tenant_id = credential.tenant_id
-            if tenant_id not in notion_credentials_tenant_mapping:
-                notion_credentials_tenant_mapping[tenant_id] = []
-            notion_credentials_tenant_mapping[tenant_id].append(credential)
-        for tenant_id, credentials in notion_credentials_tenant_mapping.items():
-            # check notion plugin is installed
-            installed_plugins = installer_manager.list_plugins(tenant_id)
-            installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
-            if notion_plugin_id not in installed_plugins_ids:
-                if notion_plugin_unique_identifier:
-                    # install notion plugin
-                    installer_manager.install_from_identifiers(
-                        tenant_id,
-                        [notion_plugin_unique_identifier],
-                        PluginInstallationSource.Marketplace,
-                        metas=[
-                            {
-                                "plugin_unique_identifier": notion_plugin_unique_identifier,
-                            }
-                        ],
-                    )
-            auth_count = 0
-            for credential in credentials:
-                auth_count += 1
-                # get credential oauth params
-                access_token = credential.access_token
-                # notion info
-                notion_info = credential.source_info
-                workspace_id = notion_info.get("workspace_id")
-                workspace_name = notion_info.get("workspace_name")
-                workspace_icon = notion_info.get("workspace_icon")
-                new_credentials = {
-                    "integration_secret": encrypter.encrypt_token(tenant_id, access_token),
-                    "workspace_id": workspace_id,
-                    "workspace_name": workspace_name,
-                    "workspace_icon": workspace_icon,
-                }
-                datasource_provider = DatasourceProvider(
-                    provider="notion",
-                    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()
-        firecrawl_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
-        for credential in firecrawl_credentials:
-            tenant_id = credential.tenant_id
-            if tenant_id not in firecrawl_credentials_tenant_mapping:
-                firecrawl_credentials_tenant_mapping[tenant_id] = []
-            firecrawl_credentials_tenant_mapping[tenant_id].append(credential)
-        for tenant_id, credentials in firecrawl_credentials_tenant_mapping.items():
-            # check firecrawl plugin is installed
-            installed_plugins = installer_manager.list_plugins(tenant_id)
-            installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
-            if firecrawl_plugin_id not in installed_plugins_ids:
-                if firecrawl_plugin_unique_identifier:
-                    # install firecrawl plugin
-                    installer_manager.install_from_identifiers(
-                        tenant_id,
-                        [firecrawl_plugin_unique_identifier],
-                        PluginInstallationSource.Marketplace,
-                        metas=[
-                            {
-                                "plugin_unique_identifier": firecrawl_plugin_unique_identifier,
-                            }
-                        ],
-                    )
-            auth_count = 0
-            for credential in credentials:
-                auth_count += 1
-                # get credential api key
-                api_key = credential.credentials.get("config", {}).get("api_key")
-                base_url = credential.credentials.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="jina").all()
-        jina_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
-        for credential in jina_credentials:
-            tenant_id = credential.tenant_id
-            if tenant_id not in jina_credentials_tenant_mapping:
-                jina_credentials_tenant_mapping[tenant_id] = []
-            jina_credentials_tenant_mapping[tenant_id].append(credential)
-        for tenant_id, credentials in jina_credentials_tenant_mapping.items():
-            # check jina plugin is installed
-            installed_plugins = installer_manager.list_plugins(tenant_id)
-            installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
-            if jina_plugin_id not in installed_plugins_ids:
-                if jina_plugin_unique_identifier:
-                    # install jina plugin
-                    installer_manager.install_from_identifiers(
-                        tenant_id,
-                        [jina_plugin_unique_identifier],
-                        PluginInstallationSource.Marketplace,
-                        metas=[
-                            {
-                                "plugin_unique_identifier": jina_plugin_unique_identifier,
-                            }
-                        ],
-                    )
-            auth_count = 0
-            for credential in credentials:
-                auth_count += 1
-                # get credential api key
-                api_key = credential.credentials.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"))

api/configs/feature/__init__.py

@@ -552,18 +552,12 @@ class RepositoryConfig(BaseSettings):
     """
 
     CORE_WORKFLOW_EXECUTION_REPOSITORY: str = Field(
-        description="Repository implementation for WorkflowExecution. Options: "
-        "'core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository' (default), "
-        "'core.repositories.celery_workflow_execution_repository.CeleryWorkflowExecutionRepository'",
+        description="Repository implementation for WorkflowExecution. Specify as a module path",
         default="core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository",
     )
 
     CORE_WORKFLOW_NODE_EXECUTION_REPOSITORY: str = Field(
-        description="Repository implementation for WorkflowNodeExecution. Options: "
-        "'core.repositories.sqlalchemy_workflow_node_execution_repository."
-        "SQLAlchemyWorkflowNodeExecutionRepository' (default), "
-        "'core.repositories.celery_workflow_node_execution_repository."
-        "CeleryWorkflowNodeExecutionRepository'",
+        description="Repository implementation for WorkflowNodeExecution. Specify as a module path",
         default="core.repositories.sqlalchemy_workflow_node_execution_repository.SQLAlchemyWorkflowNodeExecutionRepository",
     )
 

api/configs/feature/hosted_service/__init__.py

@@ -222,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="database",
-    )
-
-    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/contexts/__init__.py

@@ -3,7 +3,6 @@ from threading import Lock
 from typing import TYPE_CHECKING
 
 from contexts.wrapper import RecyclableContextVar
-from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
 
 if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
@@ -34,11 +33,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/common/errors.py

@@ -1,7 +1,5 @@
 from werkzeug.exceptions import HTTPException
 
-from libs.exception import BaseHTTPException
-
 
 class FilenameNotExistsError(HTTPException):
     code = 400
@@ -11,27 +9,3 @@ class FilenameNotExistsError(HTTPException):
 class RemoteFileUploadError(HTTPException):
     code = 400
     description = "Error uploading remote file."
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400

api/controllers/console/__init__.py

@@ -87,15 +87,6 @@ from .datasets import (
     upload_file,
     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 (

api/controllers/console/app/annotation.py

@@ -3,8 +3,9 @@ from flask_login import current_user
 from flask_restful import Resource, marshal, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden
 
-from controllers.common.errors import NoFileUploadedError, TooManyFilesError
 from controllers.console import api
+from controllers.console.app.error import NoFileUploadedError
+from controllers.console.datasets.error import TooManyFilesError
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,

api/controllers/console/app/completion.py

@@ -1,7 +1,6 @@
 import logging
 
 import flask_login
-from flask import request
 from flask_restful import Resource, reqparse
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -25,7 +24,6 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
@@ -117,10 +115,6 @@ class ChatMessageApi(Resource):
         streaming = args["response_mode"] != "blocking"
         args["auto_generate_name"] = False
 
-        external_trace_id = get_external_trace_id(request)
-        if external_trace_id:
-            args["external_trace_id"] = external_trace_id
-
         account = flask_login.current_user
 
         try:

api/controllers/console/app/error.py

@@ -79,6 +79,18 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
+class NoFileUploadedError(BaseHTTPException):
+    error_code = "no_file_uploaded"
+    description = "Please upload your file."
+    code = 400
+
+
+class TooManyFilesError(BaseHTTPException):
+    error_code = "too_many_files"
+    description = "Only one file is allowed."
+    code = 400
+
+
 class DraftWorkflowNotExist(BaseHTTPException):
     error_code = "draft_workflow_not_exist"
     description = "Draft workflow need to be initialized."

api/controllers/console/app/message.py

@@ -27,7 +27,7 @@ from fields.conversation_fields import annotation_fields, message_detail_fields
 from libs.helper import uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import login_required
-from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
+from models.model import AppMode, Conversation, Message, MessageAnnotation
 from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
@@ -124,33 +124,16 @@ class MessageFeedbackApi(Resource):
         parser.add_argument("rating", type=str, choices=["like", "dislike", None], location="json")
         args = parser.parse_args()
 
-        message_id = str(args["message_id"])
-
-        message = db.session.query(Message).filter(Message.id == message_id, Message.app_id == app_model.id).first()
-
-        if not message:
-            raise NotFound("Message Not Exists.")
-
-        feedback = message.admin_feedback
-
-        if not args["rating"] and feedback:
-            db.session.delete(feedback)
-        elif args["rating"] and feedback:
-            feedback.rating = args["rating"]
-        elif not args["rating"] and not feedback:
-            raise ValueError("rating cannot be None when feedback not exists")
-        else:
-            feedback = MessageFeedback(
-                app_id=app_model.id,
-                conversation_id=message.conversation_id,
-                message_id=message.id,
-                rating=args["rating"],
-                from_source="admin",
-                from_account_id=current_user.id,
+        try:
+            MessageService.create_feedback(
+                app_model=app_model,
+                message_id=str(args["message_id"]),
+                user=current_user,
+                rating=args.get("rating"),
+                content=None,
             )
-            db.session.add(feedback)
-
-        db.session.commit()
+        except MessageNotExistsError:
+            raise NotFound("Message Not Exists.")
 
         return {"result": "success"}
 

api/controllers/console/app/workflow.py

@@ -23,7 +23,6 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 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 extensions.ext_database import db
 from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -186,10 +185,6 @@ class AdvancedChatDraftWorkflowRunApi(Resource):
 
         args = parser.parse_args()
 
-        external_trace_id = get_external_trace_id(request)
-        if external_trace_id:
-            args["external_trace_id"] = external_trace_id
-
         try:
             response = AppGenerateService.generate(
                 app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=True
@@ -378,10 +373,6 @@ class DraftWorkflowRunApi(Resource):
         parser.add_argument("files", type=list, required=False, location="json")
         args = parser.parse_args()
 
-        external_trace_id = get_external_trace_id(request)
-        if external_trace_id:
-            args["external_trace_id"] = external_trace_id
-
         try:
             response = AppGenerateService.generate(
                 app_model=app_model,

api/controllers/console/app/workflow_draft_variable.py

@@ -163,11 +163,11 @@ class WorkflowVariableCollectionApi(Resource):
             draft_var_srv = WorkflowDraftVariableService(
                 session=session,
             )
-            workflow_vars = draft_var_srv.list_variables_without_values(
-                app_id=app_model.id,
-                page=args.page,
-                limit=args.limit,
-            )
+        workflow_vars = draft_var_srv.list_variables_without_values(
+            app_id=app_model.id,
+            page=args.page,
+            limit=args.limit,
+        )
 
         return workflow_vars
 

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.extract_setting import ExtractSetting
 from core.rag.extractor.notion_extractor import NotionExtractor
@@ -22,7 +18,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
 
 
@@ -117,18 +112,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
@@ -152,49 +135,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):
@@ -202,25 +167,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,
         )
 
@@ -245,12 +212,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="notion_import",
                     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

@@ -279,15 +279,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()
 
@@ -438,12 +429,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="notion_import",
                         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_document.py

@@ -1,4 +1,3 @@
-import json
 import logging
 from argparse import ArgumentTypeError
 from typing import cast
@@ -52,7 +51,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
 
@@ -510,7 +508,6 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 extract_setting = ExtractSetting(
                     datasource_type="notion_import",
                     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"],
@@ -664,7 +661,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,
@@ -1029,41 +1026,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")
@@ -1085,6 +1047,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

@@ -1,6 +1,30 @@
 from libs.exception import BaseHTTPException
 
 
+class NoFileUploadedError(BaseHTTPException):
+    error_code = "no_file_uploaded"
+    description = "Please upload your file."
+    code = 400
+
+
+class TooManyFilesError(BaseHTTPException):
+    error_code = "too_many_files"
+    description = "Only one file is allowed."
+    code = 400
+
+
+class FileTooLargeError(BaseHTTPException):
+    error_code = "file_too_large"
+    description = "File size exceeded. {message}"
+    code = 413
+
+
+class UnsupportedFileTypeError(BaseHTTPException):
+    error_code = "unsupported_file_type"
+    description = "File type not allowed."
+    code = 415
+
+
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."
@@ -71,9 +95,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/rag_pipeline/datasource_auth.py

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

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

@@ -1,171 +0,0 @@
-from flask_login import current_user  # type: ignore  # type: ignore
-from flask_restful import Resource, marshal, reqparse  # type: ignore
-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 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 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(
-            "name",
-            nullable=False,
-            required=True,
-            help="type is required. 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,
-            nullable=True,
-            required=False,
-            default={},
-        )
-
-        parser.add_argument(
-            "permission",
-            type=str,
-            choices=(DatasetPermissionEnum.ONLY_ME, DatasetPermissionEnum.ALL_TEAM, DatasetPermissionEnum.PARTIAL_TEAM),
-            nullable=True,
-            required=False,
-            default=DatasetPermissionEnum.ONLY_ME,
-        )
-
-        parser.add_argument(
-            "partial_member_list",
-            type=list,
-            nullable=True,
-            required=False,
-            default=[],
-        )
-
-        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(**args)
-        try:
-            import_info = RagPipelineDslService.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()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument(
-            "name",
-            nullable=False,
-            required=True,
-            help="type is required. 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,
-            nullable=True,
-            required=False,
-            default={},
-        )
-
-        parser.add_argument(
-            "permission",
-            type=str,
-            choices=(DatasetPermissionEnum.ONLY_ME, DatasetPermissionEnum.ALL_TEAM, DatasetPermissionEnum.PARTIAL_TEAM),
-            nullable=True,
-            required=False,
-            default=DatasetPermissionEnum.ONLY_ME,
-        )
-
-        parser.add_argument(
-            "partial_member_list",
-            type=list,
-            nullable=True,
-            required=False,
-            default=[],
-        )
-
-        args = parser.parse_args()
-        dataset = DatasetService.create_empty_rag_pipeline_dataset(
-            tenant_id=current_user.current_tenant_id,
-            rag_pipeline_dataset_create_entity=RagPipelineDatasetCreateEntity(**args),
-        )
-        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,416 +0,0 @@
-import logging
-from typing import Any, NoReturn
-
-from flask import Response
-from flask_restful 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.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 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 db
-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
-
-
-_WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
-    "id": fields.String,
-    "type": fields.String(attribute=lambda model: model.get_variable_type()),
-    "name": fields.String,
-    "description": fields.String,
-    "selector": fields.List(fields.String, attribute=lambda model: model.get_selector()),
-    "value_type": fields.String,
-    "edited": fields.Boolean(attribute=lambda model: model.edited),
-    "visible": fields.Boolean,
-}
-
-_WORKFLOW_DRAFT_VARIABLE_FIELDS = dict(
-    _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,
-    value=fields.Raw(attribute=_serialize_var_value),
-)
-
-_WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS = {
-    "id": fields.String,
-    "type": fields.String(attribute=lambda _: "env"),
-    "name": fields.String,
-    "description": fields.String,
-    "selector": fields.List(fields.String, attribute=lambda model: model.get_selector()),
-    "value_type": fields.String,
-    "edited": fields.Boolean(attribute=lambda model: model.edited),
-    "visible": fields.Boolean,
-}
-
-_WORKFLOW_DRAFT_ENV_VARIABLE_LIST_FIELDS = {
-    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS)),
-}
-
-
-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 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_restful import Resource, marshal_with, reqparse  # type: ignore
-from sqlalchemy.orm import Session
-from werkzeug.exceptions import Forbidden
-
-from controllers.console import api
-from controllers.console.datasets.wraps import get_rag_pipeline
-from controllers.console.wraps import (
-    account_initialization_required,
-    setup_required,
-)
-from extensions.ext_database import db
-from fields.rag_pipeline_fields import pipeline_import_check_dependencies_fields, pipeline_import_fields
-from libs.login import login_required
-from models import Account
-from models.dataset import Pipeline
-from services.app_dsl_service import ImportStatus
-from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
-
-
-class RagPipelineImportApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @marshal_with(pipeline_import_fields)
-    def post(self):
-        # Check user role first
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("mode", type=str, required=True, location="json")
-        parser.add_argument("yaml_content", type=str, location="json")
-        parser.add_argument("yaml_url", type=str, location="json")
-        parser.add_argument("name", type=str, location="json")
-        parser.add_argument("description", type=str, location="json")
-        parser.add_argument("icon_type", type=str, location="json")
-        parser.add_argument("icon", type=str, location="json")
-        parser.add_argument("icon_background", type=str, location="json")
-        parser.add_argument("pipeline_id", type=str, location="json")
-        args = parser.parse_args()
-
-        # Create service with session
-        with Session(db.engine) as session:
-            import_service = RagPipelineDslService(session)
-            # Import app
-            account = cast(Account, current_user)
-            result = import_service.import_rag_pipeline(
-                account=account,
-                import_mode=args["mode"],
-                yaml_content=args.get("yaml_content"),
-                yaml_url=args.get("yaml_url"),
-                pipeline_id=args.get("pipeline_id"),
-                dataset_name=args.get("name"),
-            )
-            session.commit()
-
-        # Return appropriate status code based on result
-        status = result.status
-        if status == ImportStatus.FAILED.value:
-            return result.model_dump(mode="json"), 400
-        elif status == ImportStatus.PENDING.value:
-            return result.model_dump(mode="json"), 202
-        return result.model_dump(mode="json"), 200
-
-
-class RagPipelineImportConfirmApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @marshal_with(pipeline_import_fields)
-    def post(self, import_id):
-        # Check user role first
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        # Create service with session
-        with Session(db.engine) as session:
-            import_service = RagPipelineDslService(session)
-            # Confirm import
-            account = cast(Account, current_user)
-            result = import_service.confirm_import(import_id=import_id, account=account)
-            session.commit()
-
-        # Return appropriate status code based on result
-        if result.status == ImportStatus.FAILED.value:
-            return result.model_dump(mode="json"), 400
-        return result.model_dump(mode="json"), 200
-
-
-class RagPipelineImportCheckDependenciesApi(Resource):
-    @setup_required
-    @login_required
-    @get_rag_pipeline
-    @account_initialization_required
-    @marshal_with(pipeline_import_check_dependencies_fields)
-    def get(self, pipeline: Pipeline):
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        with Session(db.engine) as session:
-            import_service = RagPipelineDslService(session)
-            result = import_service.check_dependencies(pipeline=pipeline)
-
-        return result.model_dump(mode="json"), 200
-
-
-class RagPipelineExportApi(Resource):
-    @setup_required
-    @login_required
-    @get_rag_pipeline
-    @account_initialization_required
-    def get(self, pipeline: Pipeline):
-        if not current_user.is_editor:
-            raise Forbidden()
-
-            # Add include_secret params
-        parser = reqparse.RequestParser()
-        parser.add_argument("include_secret", type=bool, default=False, location="args")
-        args = parser.parse_args()
-
-        with Session(db.engine) as session:
-            export_service = RagPipelineDslService(session)
-            result = export_service.export_rag_pipeline_dsl(pipeline=pipeline, include_secret=args["include_secret"])
-
-        return {"data": result}, 200
-
-
-# Import Rag Pipeline
-api.add_resource(
-    RagPipelineImportApi,
-    "/rag/pipelines/imports",
-)
-api.add_resource(
-    RagPipelineImportConfirmApi,
-    "/rag/pipelines/imports/<string:import_id>/confirm",
-)
-api.add_resource(
-    RagPipelineImportCheckDependenciesApi,
-    "/rag/pipelines/imports/<string:pipeline_id>/check-dependencies",
-)
-api.add_resource(
-    RagPipelineExportApi,
-    "/rag/pipelines/<string:pipeline_id>/exports",
-)

api/controllers/console/datasets/wraps.py

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

api/controllers/console/error.py

@@ -76,6 +76,30 @@ class EmailSendIpLimitError(BaseHTTPException):
     code = 429
 
 
+class FileTooLargeError(BaseHTTPException):
+    error_code = "file_too_large"
+    description = "File size exceeded. {message}"
+    code = 413
+
+
+class UnsupportedFileTypeError(BaseHTTPException):
+    error_code = "unsupported_file_type"
+    description = "File type not allowed."
+    code = 415
+
+
+class TooManyFilesError(BaseHTTPException):
+    error_code = "too_many_files"
+    description = "Only one file is allowed."
+    code = 400
+
+
+class NoFileUploadedError(BaseHTTPException):
+    error_code = "no_file_uploaded"
+    description = "Please upload your file."
+    code = 400
+
+
 class UnauthorizedAndForceLogout(BaseHTTPException):
     error_code = "unauthorized_and_force_logout"
     description = "Unauthorized and force logout."

api/controllers/console/files.py

@@ -8,13 +8,7 @@ from werkzeug.exceptions import Forbidden
 import services
 from configs import dify_config
 from constants import DOCUMENT_EXTENSIONS
-from controllers.common.errors import (
-    FilenameNotExistsError,
-    FileTooLargeError,
-    NoFileUploadedError,
-    TooManyFilesError,
-    UnsupportedFileTypeError,
-)
+from controllers.common.errors import FilenameNotExistsError
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,
@@ -24,6 +18,13 @@ from fields.file_fields import file_fields, upload_config_fields
 from libs.login import login_required
 from services.file_service import FileService
 
+from .error import (
+    FileTooLargeError,
+    NoFileUploadedError,
+    TooManyFilesError,
+    UnsupportedFileTypeError,
+)
+
 PREVIEW_WORDS_LIMIT = 3000
 
 

api/controllers/console/remote_files.py

@@ -7,17 +7,18 @@ from flask_restful import Resource, marshal_with, reqparse
 
 import services
 from controllers.common import helpers
-from controllers.common.errors import (
-    FileTooLargeError,
-    RemoteFileUploadError,
-    UnsupportedFileTypeError,
-)
+from controllers.common.errors import RemoteFileUploadError
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 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
 
+from .error import (
+    FileTooLargeError,
+    UnsupportedFileTypeError,
+)
+
 
 class RemoteFileInfoApi(Resource):
     @marshal_with(remote_file_info_fields)

api/controllers/console/version.py

@@ -32,7 +32,7 @@ class VersionApi(Resource):
             return result
 
         try:
-            response = requests.get(check_update_url, {"current_version": args.get("current_version")}, timeout=(3, 10))
+            response = requests.get(check_update_url, {"current_version": args.get("current_version")})
         except Exception as error:
             logging.warning("Check update version error: %s.", str(error))
             result["version"] = args.get("current_version")

api/controllers/console/workspace/tool_providers.py

@@ -862,10 +862,6 @@ class ToolProviderMCPApi(Resource):
         parser.add_argument("icon_type", type=str, required=True, nullable=False, location="json")
         parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json", default="")
         parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("timeout", type=float, required=False, nullable=False, location="json", default=30)
-        parser.add_argument(
-            "sse_read_timeout", type=float, required=False, nullable=False, location="json", default=300
-        )
         args = parser.parse_args()
         user = current_user
         if not is_valid_url(args["server_url"]):
@@ -880,8 +876,6 @@ class ToolProviderMCPApi(Resource):
                 icon_background=args["icon_background"],
                 user_id=user.id,
                 server_identifier=args["server_identifier"],
-                timeout=args["timeout"],
-                sse_read_timeout=args["sse_read_timeout"],
             )
         )
 
@@ -897,8 +891,6 @@ class ToolProviderMCPApi(Resource):
         parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json")
         parser.add_argument("provider_id", type=str, required=True, nullable=False, location="json")
         parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("timeout", type=float, required=False, nullable=True, location="json")
-        parser.add_argument("sse_read_timeout", type=float, required=False, nullable=True, location="json")
         args = parser.parse_args()
         if not is_valid_url(args["server_url"]):
             if "[__HIDDEN__]" in args["server_url"]:
@@ -914,8 +906,6 @@ class ToolProviderMCPApi(Resource):
             icon_type=args["icon_type"],
             icon_background=args["icon_background"],
             server_identifier=args["server_identifier"],
-            timeout=args.get("timeout"),
-            sse_read_timeout=args.get("sse_read_timeout"),
         )
         return {"result": "success"}
 

api/controllers/console/workspace/workspace.py

@@ -7,15 +7,15 @@ from sqlalchemy import select
 from werkzeug.exceptions import Unauthorized
 
 import services
-from controllers.common.errors import (
-    FilenameNotExistsError,
+from controllers.common.errors import FilenameNotExistsError
+from controllers.console import api
+from controllers.console.admin import admin_required
+from controllers.console.datasets.error import (
     FileTooLargeError,
     NoFileUploadedError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
-from controllers.console import api
-from controllers.console.admin import admin_required
 from controllers.console.error import AccountNotLinkTenantError
 from controllers.console.wraps import (
     account_initialization_required,

api/controllers/console/wraps.py

@@ -261,14 +261,3 @@ def is_allow_transfer_owner(view):
         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/error.py

@@ -0,0 +1,7 @@
+from libs.exception import BaseHTTPException
+
+
+class UnsupportedFileTypeError(BaseHTTPException):
+    error_code = "unsupported_file_type"
+    description = "File type not allowed."
+    code = 415

api/controllers/files/image_preview.py

@@ -5,8 +5,8 @@ from flask_restful import Resource, reqparse
 from werkzeug.exceptions import NotFound
 
 import services
-from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import api
+from controllers.files.error import UnsupportedFileTypeError
 from services.account_service import TenantService
 from services.file_service import FileService
 

api/controllers/files/tool_files.py

@@ -4,8 +4,8 @@ from flask import Response
 from flask_restful import Resource, reqparse
 from werkzeug.exceptions import Forbidden, NotFound
 
-from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import api
+from controllers.files.error import UnsupportedFileTypeError
 from core.tools.signature import verify_tool_file_signature
 from core.tools.tool_file_manager import ToolFileManager
 from models import db as global_db

api/controllers/files/upload.py

@@ -5,13 +5,11 @@ from flask_restful import Resource, marshal_with
 from werkzeug.exceptions import Forbidden
 
 import services
-from controllers.common.errors import (
-    FileTooLargeError,
-    UnsupportedFileTypeError,
-)
 from controllers.console.wraps import setup_required
 from controllers.files import api
+from controllers.files.error import UnsupportedFileTypeError
 from controllers.inner_api.plugin.wraps import get_user
+from controllers.service_api.app.error import FileTooLargeError
 from core.file.helpers import verify_plugin_file_signature
 from core.tools.tool_file_manager import ToolFileManager
 from fields.file_fields import file_fields

api/controllers/service_api/app/conversation.py

@@ -1,3 +1,5 @@
+import json
+
 from flask_restful import Resource, marshal_with, reqparse
 from flask_restful.inputs import int_range
 from sqlalchemy.orm import Session
@@ -134,15 +136,12 @@ class ConversationVariableDetailApi(Resource):
         variable_id = str(variable_id)
 
         parser = reqparse.RequestParser()
-        # using lambda is for passing the already-typed value without modification
-        # if no lambda, it will be converted to string
-        # the string cannot be converted using json.loads
-        parser.add_argument("value", required=True, location="json", type=lambda x: x)
+        parser.add_argument("value", required=True, location="json")
         args = parser.parse_args()
 
         try:
             return ConversationService.update_conversation_variable(
-                app_model, conversation_id, variable_id, end_user, args["value"]
+                app_model, conversation_id, variable_id, end_user, json.loads(args["value"])
             )
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")

api/controllers/service_api/app/error.py

@@ -85,6 +85,30 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
+class NoFileUploadedError(BaseHTTPException):
+    error_code = "no_file_uploaded"
+    description = "Please upload your file."
+    code = 400
+
+
+class TooManyFilesError(BaseHTTPException):
+    error_code = "too_many_files"
+    description = "Only one file is allowed."
+    code = 400
+
+
+class FileTooLargeError(BaseHTTPException):
+    error_code = "file_too_large"
+    description = "File size exceeded. {message}"
+    code = 413
+
+
+class UnsupportedFileTypeError(BaseHTTPException):
+    error_code = "unsupported_file_type"
+    description = "File type not allowed."
+    code = 415
+
+
 class FileNotFoundError(BaseHTTPException):
     error_code = "file_not_found"
     description = "The requested file was not found."

api/controllers/service_api/app/file.py

@@ -2,14 +2,14 @@ from flask import request
 from flask_restful import Resource, marshal_with
 
 import services
-from controllers.common.errors import (
-    FilenameNotExistsError,
+from controllers.common.errors import FilenameNotExistsError
+from controllers.service_api import api
+from controllers.service_api.app.error import (
     FileTooLargeError,
     NoFileUploadedError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
-from controllers.service_api import api
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from fields.file_fields import file_fields
 from models.model import App, EndUser

api/controllers/service_api/dataset/document.py

@@ -6,15 +6,15 @@ from sqlalchemy import desc, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
-from controllers.common.errors import (
-    FilenameNotExistsError,
+from controllers.common.errors import FilenameNotExistsError
+from controllers.service_api import api
+from controllers.service_api.app.error import (
     FileTooLargeError,
     NoFileUploadedError,
+    ProviderNotInitializeError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
-from controllers.service_api import api
-from controllers.service_api.app.error import ProviderNotInitializeError
 from controllers.service_api.dataset.error import (
     ArchivedDocumentImmutableError,
     DocumentIndexingError,

api/controllers/service_api/dataset/error.py

@@ -1,6 +1,30 @@
 from libs.exception import BaseHTTPException
 
 
+class NoFileUploadedError(BaseHTTPException):
+    error_code = "no_file_uploaded"
+    description = "Please upload your file."
+    code = 400
+
+
+class TooManyFilesError(BaseHTTPException):
+    error_code = "too_many_files"
+    description = "Only one file is allowed."
+    code = 400
+
+
+class FileTooLargeError(BaseHTTPException):
+    error_code = "file_too_large"
+    description = "File size exceeded. {message}"
+    code = 413
+
+
+class UnsupportedFileTypeError(BaseHTTPException):
+    error_code = "unsupported_file_type"
+    description = "File type not allowed."
+    code = 415
+
+
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."

api/controllers/web/error.py

@@ -97,6 +97,30 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
+class NoFileUploadedError(BaseHTTPException):
+    error_code = "no_file_uploaded"
+    description = "Please upload your file."
+    code = 400
+
+
+class TooManyFilesError(BaseHTTPException):
+    error_code = "too_many_files"
+    description = "Only one file is allowed."
+    code = 400
+
+
+class FileTooLargeError(BaseHTTPException):
+    error_code = "file_too_large"
+    description = "File size exceeded. {message}"
+    code = 413
+
+
+class UnsupportedFileTypeError(BaseHTTPException):
+    error_code = "unsupported_file_type"
+    description = "File type not allowed."
+    code = 415
+
+
 class WebAppAuthRequiredError(BaseHTTPException):
     error_code = "web_sso_auth_required"
     description = "Web app authentication required."

api/controllers/web/files.py

@@ -2,13 +2,8 @@ from flask import request
 from flask_restful import marshal_with
 
 import services
-from controllers.common.errors import (
-    FilenameNotExistsError,
-    FileTooLargeError,
-    NoFileUploadedError,
-    TooManyFilesError,
-    UnsupportedFileTypeError,
-)
+from controllers.common.errors import FilenameNotExistsError
+from controllers.web.error import FileTooLargeError, NoFileUploadedError, TooManyFilesError, UnsupportedFileTypeError
 from controllers.web.wraps import WebApiResource
 from fields.file_fields import file_fields
 from services.file_service import FileService

api/controllers/web/remote_files.py

@@ -5,17 +5,15 @@ from flask_restful import marshal_with, reqparse
 
 import services
 from controllers.common import helpers
-from controllers.common.errors import (
-    FileTooLargeError,
-    RemoteFileUploadError,
-    UnsupportedFileTypeError,
-)
+from controllers.common.errors import RemoteFileUploadError
 from controllers.web.wraps import WebApiResource
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields
 from services.file_service import FileService
 
+from .error import FileTooLargeError, UnsupportedFileTypeError
+
 
 class RemoteFileInfoApi(WebApiResource):
     @marshal_with(remote_file_info_fields)

api/core/app/app_config/entities.py

@@ -97,7 +97,6 @@ class VariableEntityType(StrEnum):
     EXTERNAL_DATA_TOOL = "external_data_tool"
     FILE = "file"
     FILE_LIST = "file-list"
-    CHECKBOX = "checkbox"
 
 
 class VariableEntity(BaseModel):
@@ -114,9 +113,9 @@ class VariableEntity(BaseModel):
     hide: bool = False
     max_length: Optional[int] = None
     options: Sequence[str] = Field(default_factory=list)
-    allowed_file_types: Optional[Sequence[FileType]] = Field(default_factory=list)
-    allowed_file_extensions: Optional[Sequence[str]] = Field(default_factory=list)
-    allowed_file_upload_methods: Optional[Sequence[FileTransferMethod]] = Field(default_factory=list)
+    allowed_file_types: Sequence[FileType] = Field(default_factory=list)
+    allowed_file_extensions: Sequence[str] = Field(default_factory=list)
+    allowed_file_upload_methods: Sequence[FileTransferMethod] = Field(default_factory=list)
 
     @field_validator("description", mode="before")
     @classmethod
@@ -129,16 +128,6 @@ class VariableEntity(BaseModel):
         return v or []
 
 
-class RagPipelineVariableEntity(VariableEntity):
-    """
-    Rag Pipeline Variable Entity.
-    """
-
-    tooltips: Optional[str] = None
-    placeholder: Optional[str] = None
-    belong_to_node_id: str
-
-
 class ExternalDataVariableEntity(BaseModel):
     """
     External Data Variable Entity.
@@ -298,7 +287,7 @@ class AppConfig(BaseModel):
     tenant_id: str
     app_id: str
     app_mode: AppMode
-    additional_features: Optional[AppAdditionalFeatures] = None
+    additional_features: AppAdditionalFeatures
     variables: list[VariableEntity] = []
     sensitive_word_avoidance: Optional[SensitiveWordAvoidanceEntity] = None
 

api/core/app/app_config/workflow_ui_based_app/variables/manager.py

@@ -1,6 +1,4 @@
-import re
-
-from core.app.app_config.entities import RagPipelineVariableEntity, VariableEntity
+from core.app.app_config.entities import VariableEntity
 from models.workflow import Workflow
 
 
@@ -22,44 +20,3 @@ class WorkflowVariablesConfigManager:
             variables.append(VariableEntity.model_validate(variable))
 
         return variables
-
-    @classmethod
-    def convert_rag_pipeline_variable(cls, workflow: Workflow, start_node_id: str) -> list[RagPipelineVariableEntity]:
-        """
-        Convert workflow start variables to variables
-
-        :param workflow: workflow instance
-        """
-        variables = []
-
-        # get second step node
-        rag_pipeline_variables = workflow.rag_pipeline_variables
-        if not rag_pipeline_variables:
-            return []
-        variables_map = {item["variable"]: item for item in rag_pipeline_variables}
-
-        # get datasource node data
-        datasource_node_data = None
-        datasource_nodes = workflow.graph_dict.get("nodes", [])
-        for datasource_node in datasource_nodes:
-            if datasource_node.get("id") == start_node_id:
-                datasource_node_data = datasource_node.get("data", {})
-                break
-        if datasource_node_data:
-            datasource_parameters = datasource_node_data.get("datasource_parameters", {})
-
-            for key, value in datasource_parameters.items():
-                if value.get("value") and isinstance(value.get("value"), str):
-                    pattern = r"\{\{#([a-zA-Z0-9_]{1,50}(?:\.[a-zA-Z0-9_][a-zA-Z0-9_]{0,29}){1,10})#\}\}"
-                    match = re.match(pattern, value["value"])
-                    if match:
-                        full_path = match.group(1)
-                        last_part = full_path.split(".")[-1]
-                        variables_map.pop(last_part)
-        all_second_step_variables = list(variables_map.values())
-
-        for item in all_second_step_variables:
-            if item.get("belong_to_node_id") == start_node_id or item.get("belong_to_node_id") == "shared":
-                variables.append(RagPipelineVariableEntity.model_validate(item))
-
-        return variables

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -74,7 +74,6 @@ from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
 from events.message_event import message_was_created
 from extensions.ext_database import db
-from libs.datetime_utils import naive_utc_now
 from models import Conversation, EndUser, Message, MessageFile
 from models.account import Account
 from models.enums import CreatorUserRole
@@ -569,7 +568,7 @@ class AdvancedChatAppGenerateTaskPipeline:
             )
 
         yield workflow_finish_resp
-        self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
+        self._base_task_pipeline._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
 
     def _handle_workflow_partial_success_event(
         self,
@@ -601,7 +600,7 @@ class AdvancedChatAppGenerateTaskPipeline:
             )
 
         yield workflow_finish_resp
-        self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
+        self._base_task_pipeline._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
 
     def _handle_workflow_failed_event(
         self,
@@ -846,7 +845,7 @@ class AdvancedChatAppGenerateTaskPipeline:
         # Initialize graph runtime state
         graph_runtime_state: Optional[GraphRuntimeState] = None
 
-        for queue_message in self._base_task_pipeline.queue_manager.listen():
+        for queue_message in self._base_task_pipeline._queue_manager.listen():
             event = queue_message.event
 
             match event:
@@ -897,7 +896,6 @@ class AdvancedChatAppGenerateTaskPipeline:
     def _save_message(self, *, session: Session, graph_runtime_state: Optional[GraphRuntimeState] = None) -> None:
         message = self._get_message(session=session)
         message.answer = self._task_state.answer
-        message.updated_at = naive_utc_now()
         message.provider_response_latency = time.perf_counter() - self._base_task_pipeline._start_at
         message.message_metadata = self._task_state.metadata.model_dump_json()
         message_files = [
@@ -961,11 +959,11 @@ class AdvancedChatAppGenerateTaskPipeline:
         if self._base_task_pipeline._output_moderation_handler:
             if self._base_task_pipeline._output_moderation_handler.should_direct_output():
                 self._task_state.answer = self._base_task_pipeline._output_moderation_handler.get_final_output()
-                self._base_task_pipeline.queue_manager.publish(
+                self._base_task_pipeline._queue_manager.publish(
                     QueueTextChunkEvent(text=self._task_state.answer), PublishFrom.TASK_PIPELINE
                 )
 
-                self._base_task_pipeline.queue_manager.publish(
+                self._base_task_pipeline._queue_manager.publish(
                     QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE
                 )
                 return True

api/core/app/apps/chat/app_generator.py

@@ -140,9 +140,7 @@ class ChatAppGenerator(MessageBasedAppGenerator):
         )
 
         # get tracing instance
-        trace_manager = TraceQueueManager(
-            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
-        )
+        trace_manager = TraceQueueManager(app_id=app_model.id)
 
         # init application generate entity
         application_generate_entity = ChatAppGenerateEntity(

api/core/app/apps/common/workflow_response_converter.py

@@ -43,13 +43,11 @@ from core.app.entities.task_entities import (
     WorkflowStartStreamResponse,
 )
 from core.file import FILE_MODEL_IDENTITY, File
-from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.tool_manager import ToolManager
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.entities.workflow_execution import WorkflowExecution
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecution, WorkflowNodeExecutionStatus
 from core.workflow.nodes import NodeType
-from core.workflow.nodes.datasource.entities import DatasourceNodeData
 from core.workflow.nodes.tool.entities import ToolNodeData
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from models import (
@@ -185,14 +183,6 @@ class WorkflowResponseConverter:
                 provider_type=node_data.provider_type,
                 provider_id=node_data.provider_id,
             )
-        elif event.node_type == NodeType.DATASOURCE:
-            node_data = cast(DatasourceNodeData, event.node_data)
-            manager = PluginDatasourceManager()
-            provider_entity = manager.fetch_datasource_provider(
-                self._application_generate_entity.app_config.tenant_id,
-                f"{node_data.plugin_id}/{node_data.provider_name}",
-            )
-            response.data.extras["icon"] = provider_entity.declaration.identity.icon
 
         return response
 

api/core/app/apps/completion/app_generator.py

@@ -124,9 +124,7 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         )
 
         # get tracing instance
-        trace_manager = TraceQueueManager(
-            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
-        )
+        trace_manager = TraceQueueManager(app_model.id)
 
         # init application generate entity
         application_generate_entity = CompletionAppGenerateEntity(

api/core/app/apps/message_based_app_queue_manager.py

@@ -6,6 +6,7 @@ from core.app.entities.queue_entities import (
     MessageQueueMessage,
     QueueAdvancedChatMessageEndEvent,
     QueueErrorEvent,
+    QueueMessage,
     QueueMessageEndEvent,
     QueueStopEvent,
 )
@@ -21,6 +22,15 @@ class MessageBasedAppQueueManager(AppQueueManager):
         self._app_mode = app_mode
         self._message_id = str(message_id)
 
+    def construct_queue_message(self, event: AppQueueEvent) -> QueueMessage:
+        return MessageQueueMessage(
+            task_id=self._task_id,
+            message_id=self._message_id,
+            conversation_id=self._conversation_id,
+            app_mode=self._app_mode,
+            event=event,
+        )
+
     def _publish(self, event: AppQueueEvent, pub_from: PublishFrom) -> None:
         """
         Publish event to queue

api/core/app/apps/pipeline/generate_response_converter.py

@@ -1,95 +0,0 @@
-from collections.abc import Generator
-from typing import cast
-
-from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
-from core.app.entities.task_entities import (
-    AppStreamResponse,
-    ErrorStreamResponse,
-    NodeFinishStreamResponse,
-    NodeStartStreamResponse,
-    PingStreamResponse,
-    WorkflowAppBlockingResponse,
-    WorkflowAppStreamResponse,
-)
-
-
-class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
-    _blocking_response_type = WorkflowAppBlockingResponse
-
-    @classmethod
-    def convert_blocking_full_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict:  # type: ignore[override]
-        """
-        Convert blocking full response.
-        :param blocking_response: blocking response
-        :return:
-        """
-        return dict(blocking_response.to_dict())
-
-    @classmethod
-    def convert_blocking_simple_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict:  # type: ignore[override]
-        """
-        Convert blocking simple response.
-        :param blocking_response: blocking response
-        :return:
-        """
-        return cls.convert_blocking_full_response(blocking_response)
-
-    @classmethod
-    def convert_stream_full_response(
-        cls, stream_response: Generator[AppStreamResponse, None, None]
-    ) -> Generator[dict | str, None, None]:
-        """
-        Convert stream full response.
-        :param stream_response: stream response
-        :return:
-        """
-        for chunk in stream_response:
-            chunk = cast(WorkflowAppStreamResponse, chunk)
-            sub_stream_response = chunk.stream_response
-
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
-
-            response_chunk = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(data)
-            else:
-                response_chunk.update(sub_stream_response.to_dict())
-            yield response_chunk
-
-    @classmethod
-    def convert_stream_simple_response(
-        cls, stream_response: Generator[AppStreamResponse, None, None]
-    ) -> Generator[dict | str, None, None]:
-        """
-        Convert stream simple response.
-        :param stream_response: stream response
-        :return:
-        """
-        for chunk in stream_response:
-            chunk = cast(WorkflowAppStreamResponse, chunk)
-            sub_stream_response = chunk.stream_response
-
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
-
-            response_chunk = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(data)
-            elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
-                response_chunk.update(sub_stream_response.to_ignore_detail_dict())
-            else:
-                response_chunk.update(sub_stream_response.to_dict())
-            yield response_chunk

api/core/app/apps/pipeline/pipeline_config_manager.py

@@ -1,66 +0,0 @@
-from core.app.app_config.base_app_config_manager import BaseAppConfigManager
-from core.app.app_config.common.sensitive_word_avoidance.manager import SensitiveWordAvoidanceConfigManager
-from core.app.app_config.entities import RagPipelineVariableEntity, WorkflowUIBasedAppConfig
-from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
-from core.app.app_config.features.text_to_speech.manager import TextToSpeechConfigManager
-from core.app.app_config.workflow_ui_based_app.variables.manager import WorkflowVariablesConfigManager
-from models.dataset import Pipeline
-from models.model import AppMode
-from models.workflow import Workflow
-
-
-class PipelineConfig(WorkflowUIBasedAppConfig):
-    """
-    Pipeline Config Entity.
-    """
-
-    rag_pipeline_variables: list[RagPipelineVariableEntity] = []
-    pass
-
-
-class PipelineConfigManager(BaseAppConfigManager):
-    @classmethod
-    def get_pipeline_config(cls, pipeline: Pipeline, workflow: Workflow, start_node_id: str) -> PipelineConfig:
-        pipeline_config = PipelineConfig(
-            tenant_id=pipeline.tenant_id,
-            app_id=pipeline.id,
-            app_mode=AppMode.RAG_PIPELINE,
-            workflow_id=workflow.id,
-            rag_pipeline_variables=WorkflowVariablesConfigManager.convert_rag_pipeline_variable(
-                workflow=workflow, start_node_id=start_node_id
-            ),
-        )
-
-        return pipeline_config
-
-    @classmethod
-    def config_validate(cls, tenant_id: str, config: dict, only_structure_validate: bool = False) -> dict:
-        """
-        Validate for pipeline config
-
-        :param tenant_id: tenant id
-        :param config: app model config args
-        :param only_structure_validate: only validate the structure of the config
-        """
-        related_config_keys = []
-
-        # file upload validation
-        config, current_related_config_keys = FileUploadConfigManager.validate_and_set_defaults(config=config)
-        related_config_keys.extend(current_related_config_keys)
-
-        # text_to_speech
-        config, current_related_config_keys = TextToSpeechConfigManager.validate_and_set_defaults(config)
-        related_config_keys.extend(current_related_config_keys)
-
-        # moderation validation
-        config, current_related_config_keys = SensitiveWordAvoidanceConfigManager.validate_and_set_defaults(
-            tenant_id=tenant_id, config=config, only_structure_validate=only_structure_validate
-        )
-        related_config_keys.extend(current_related_config_keys)
-
-        related_config_keys = list(set(related_config_keys))
-
-        # Filter out extra parameters
-        filtered_config = {key: config.get(key) for key in related_config_keys}
-
-        return filtered_config

api/core/app/apps/pipeline/pipeline_generator.py

@@ -1,812 +0,0 @@
-import contextvars
-import datetime
-import json
-import logging
-import secrets
-import threading
-import time
-import uuid
-from collections.abc import Generator, Mapping
-from typing import Any, Literal, Optional, Union, cast, overload
-
-from flask import Flask, current_app
-from pydantic import ValidationError
-from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
-
-import contexts
-from configs import dify_config
-from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.exc import GenerateTaskStoppedError
-from core.app.apps.pipeline.pipeline_config_manager import PipelineConfigManager
-from core.app.apps.pipeline.pipeline_queue_manager import PipelineQueueManager
-from core.app.apps.pipeline.pipeline_runner import PipelineRunner
-from core.app.apps.workflow.generate_response_converter import WorkflowAppGenerateResponseConverter
-from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline
-from core.app.entities.app_invoke_entities import InvokeFrom, RagPipelineGenerateEntity
-from core.app.entities.task_entities import WorkflowAppBlockingResponse, WorkflowAppStreamResponse
-from core.datasource.entities.datasource_entities import (
-    DatasourceProviderType,
-    OnlineDriveBrowseFilesRequest,
-)
-from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasourcePlugin
-from core.entities.knowledge_entities import PipelineDataset, PipelineDocument
-from core.model_runtime.errors.invoke import InvokeAuthorizationError
-from core.rag.index_processor.constant.built_in_field import BuiltInField
-from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
-from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
-from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
-from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
-from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
-from extensions.ext_database import db
-from libs.flask_utils import preserve_flask_contexts
-from models import Account, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
-from models.dataset import Document, DocumentPipelineExecutionLog, Pipeline
-from models.enums import WorkflowRunTriggeredFrom
-from models.model import AppMode
-from services.dataset_service import DocumentService
-from services.datasource_provider_service import DatasourceProviderService
-from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
-
-logger = logging.getLogger(__name__)
-
-
-class PipelineGenerator(BaseAppGenerator):
-    @overload
-    def generate(
-        self,
-        *,
-        pipeline: Pipeline,
-        workflow: Workflow,
-        user: Union[Account, EndUser],
-        args: Mapping[str, Any],
-        invoke_from: InvokeFrom,
-        streaming: Literal[True],
-        call_depth: int,
-        workflow_thread_pool_id: Optional[str],
-    ) -> Mapping[str, Any] | Generator[Mapping | str, None, None] | None: ...
-
-    @overload
-    def generate(
-        self,
-        *,
-        pipeline: Pipeline,
-        workflow: Workflow,
-        user: Union[Account, EndUser],
-        args: Mapping[str, Any],
-        invoke_from: InvokeFrom,
-        streaming: Literal[False],
-        call_depth: int,
-        workflow_thread_pool_id: Optional[str],
-    ) -> Mapping[str, Any]: ...
-
-    @overload
-    def generate(
-        self,
-        *,
-        pipeline: Pipeline,
-        workflow: Workflow,
-        user: Union[Account, EndUser],
-        args: Mapping[str, Any],
-        invoke_from: InvokeFrom,
-        streaming: bool,
-        call_depth: int,
-        workflow_thread_pool_id: Optional[str],
-    ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ...
-
-    def generate(
-        self,
-        *,
-        pipeline: Pipeline,
-        workflow: Workflow,
-        user: Union[Account, EndUser],
-        args: Mapping[str, Any],
-        invoke_from: InvokeFrom,
-        streaming: bool = True,
-        call_depth: int = 0,
-        workflow_thread_pool_id: Optional[str] = None,
-    ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None], None]:
-        # Add null check for dataset
-        dataset = pipeline.dataset
-        if not dataset:
-            raise ValueError("Pipeline dataset is required")
-        inputs: Mapping[str, Any] = args["inputs"]
-        start_node_id: str = args["start_node_id"]
-        datasource_type: str = args["datasource_type"]
-        datasource_info_list: list[Mapping[str, Any]] = self._format_datasource_info_list(
-            datasource_type, args["datasource_info_list"], pipeline, workflow, start_node_id, user
-        )
-        batch = time.strftime("%Y%m%d%H%M%S") + str(secrets.randbelow(900000) + 100000)
-        # convert to app config
-        pipeline_config = PipelineConfigManager.get_pipeline_config(
-            pipeline=pipeline, workflow=workflow, start_node_id=start_node_id
-        )
-        documents = []
-        if invoke_from == InvokeFrom.PUBLISHED:
-            for datasource_info in datasource_info_list:
-                position = DocumentService.get_documents_position(dataset.id)
-                document = self._build_document(
-                    tenant_id=pipeline.tenant_id,
-                    dataset_id=dataset.id,
-                    built_in_field_enabled=dataset.built_in_field_enabled,
-                    datasource_type=datasource_type,
-                    datasource_info=datasource_info,
-                    created_from="rag-pipeline",
-                    position=position,
-                    account=user,
-                    batch=batch,
-                    document_form=dataset.chunk_structure,
-                )
-                db.session.add(document)
-                documents.append(document)
-            db.session.commit()
-
-        # run in child thread
-        for i, datasource_info in enumerate(datasource_info_list):
-            workflow_run_id = str(uuid.uuid4())
-            document_id = None
-            if invoke_from == InvokeFrom.PUBLISHED:
-                document_id = documents[i].id
-                document_pipeline_execution_log = DocumentPipelineExecutionLog(
-                    document_id=document_id,
-                    datasource_type=datasource_type,
-                    datasource_info=json.dumps(datasource_info),
-                    datasource_node_id=start_node_id,
-                    input_data=inputs,
-                    pipeline_id=pipeline.id,
-                    created_by=user.id,
-                )
-                db.session.add(document_pipeline_execution_log)
-                db.session.commit()
-            application_generate_entity = RagPipelineGenerateEntity(
-                task_id=str(uuid.uuid4()),
-                app_config=pipeline_config,
-                pipeline_config=pipeline_config,
-                datasource_type=datasource_type,
-                datasource_info=datasource_info,
-                dataset_id=dataset.id,
-                start_node_id=start_node_id,
-                batch=batch,
-                document_id=document_id,
-                inputs=self._prepare_user_inputs(
-                    user_inputs=inputs,
-                    variables=pipeline_config.rag_pipeline_variables,
-                    tenant_id=pipeline.tenant_id,
-                    strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
-                ),
-                files=[],
-                user_id=user.id,
-                stream=streaming,
-                invoke_from=invoke_from,
-                call_depth=call_depth,
-                workflow_execution_id=workflow_run_id,
-            )
-
-            contexts.plugin_tool_providers.set({})
-            contexts.plugin_tool_providers_lock.set(threading.Lock())
-            if invoke_from == InvokeFrom.DEBUGGER:
-                workflow_triggered_from = WorkflowRunTriggeredFrom.RAG_PIPELINE_DEBUGGING
-            else:
-                workflow_triggered_from = WorkflowRunTriggeredFrom.RAG_PIPELINE_RUN
-            # Create workflow node execution repository
-            session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
-            workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
-                session_factory=session_factory,
-                user=user,
-                app_id=application_generate_entity.app_config.app_id,
-                triggered_from=workflow_triggered_from,
-            )
-
-            workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
-                session_factory=session_factory,
-                user=user,
-                app_id=application_generate_entity.app_config.app_id,
-                triggered_from=WorkflowNodeExecutionTriggeredFrom.RAG_PIPELINE_RUN,
-            )
-            if invoke_from == InvokeFrom.DEBUGGER:
-                return self._generate(
-                    flask_app=current_app._get_current_object(),  # type: ignore
-                    context=contextvars.copy_context(),
-                    pipeline=pipeline,
-                    workflow_id=workflow.id,
-                    user=user,
-                    application_generate_entity=application_generate_entity,
-                    invoke_from=invoke_from,
-                    workflow_execution_repository=workflow_execution_repository,
-                    workflow_node_execution_repository=workflow_node_execution_repository,
-                    streaming=streaming,
-                    workflow_thread_pool_id=workflow_thread_pool_id,
-                )
-            else:
-                # run in child thread
-                context = contextvars.copy_context()
-
-                worker_thread = threading.Thread(
-                    target=self._generate,
-                    kwargs={
-                        "flask_app": current_app._get_current_object(),  # type: ignore
-                        "context": context,
-                        "pipeline": pipeline,
-                        "workflow_id": workflow.id,
-                        "user": user,
-                        "application_generate_entity": application_generate_entity,
-                        "invoke_from": invoke_from,
-                        "workflow_execution_repository": workflow_execution_repository,
-                        "workflow_node_execution_repository": workflow_node_execution_repository,
-                        "streaming": streaming,
-                        "workflow_thread_pool_id": workflow_thread_pool_id,
-                    },
-                )
-
-                worker_thread.start()
-        # return batch, dataset, documents
-        return {
-            "batch": batch,
-            "dataset": PipelineDataset(
-                id=dataset.id,
-                name=dataset.name,
-                description=dataset.description,
-                chunk_structure=dataset.chunk_structure,
-            ).model_dump(),
-            "documents": [
-                PipelineDocument(
-                    id=document.id,
-                    position=document.position,
-                    data_source_type=document.data_source_type,
-                    data_source_info=json.loads(document.data_source_info) if document.data_source_info else None,
-                    name=document.name,
-                    indexing_status=document.indexing_status,
-                    error=document.error,
-                    enabled=document.enabled,
-                ).model_dump()
-                for document in documents
-            ],
-        }
-
-    def _generate(
-        self,
-        *,
-        flask_app: Flask,
-        context: contextvars.Context,
-        pipeline: Pipeline,
-        workflow_id: str,
-        user: Union[Account, EndUser],
-        application_generate_entity: RagPipelineGenerateEntity,
-        invoke_from: InvokeFrom,
-        workflow_execution_repository: WorkflowExecutionRepository,
-        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
-        streaming: bool = True,
-        variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
-        workflow_thread_pool_id: Optional[str] = None,
-    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
-        """
-        Generate App response.
-
-        :param pipeline: Pipeline
-        :param workflow: Workflow
-        :param user: account or end user
-        :param application_generate_entity: application generate entity
-        :param invoke_from: invoke from source
-        :param workflow_execution_repository: repository for workflow execution
-        :param workflow_node_execution_repository: repository for workflow node execution
-        :param streaming: is stream
-        :param workflow_thread_pool_id: workflow thread pool id
-        """
-        with preserve_flask_contexts(flask_app, context_vars=context):
-            # init queue manager
-            workflow = db.session.query(Workflow).filter(Workflow.id == workflow_id).first()
-            if not workflow:
-                raise ValueError(f"Workflow not found: {workflow_id}")
-            queue_manager = PipelineQueueManager(
-                task_id=application_generate_entity.task_id,
-                user_id=application_generate_entity.user_id,
-                invoke_from=application_generate_entity.invoke_from,
-                app_mode=AppMode.RAG_PIPELINE,
-            )
-            context = contextvars.copy_context()
-
-            # new thread
-            worker_thread = threading.Thread(
-                target=self._generate_worker,
-                kwargs={
-                    "flask_app": current_app._get_current_object(),  # type: ignore
-                    "context": context,
-                    "queue_manager": queue_manager,
-                    "application_generate_entity": application_generate_entity,
-                    "workflow_thread_pool_id": workflow_thread_pool_id,
-                    "variable_loader": variable_loader,
-                },
-            )
-
-            worker_thread.start()
-
-            draft_var_saver_factory = self._get_draft_var_saver_factory(
-                invoke_from,
-            )
-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                workflow=workflow,
-                queue_manager=queue_manager,
-                user=user,
-                workflow_execution_repository=workflow_execution_repository,
-                workflow_node_execution_repository=workflow_node_execution_repository,
-                stream=streaming,
-                draft_var_saver_factory=draft_var_saver_factory,
-            )
-
-            return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
-
-    def single_iteration_generate(
-        self,
-        pipeline: Pipeline,
-        workflow: Workflow,
-        node_id: str,
-        user: Account | EndUser,
-        args: Mapping[str, Any],
-        streaming: bool = True,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
-        """
-        Generate App response.
-
-        :param app_model: App
-        :param workflow: Workflow
-        :param node_id: the node id
-        :param user: account or end user
-        :param args: request args
-        :param streaming: is streamed
-        """
-        if not node_id:
-            raise ValueError("node_id is required")
-
-        if args.get("inputs") is None:
-            raise ValueError("inputs is required")
-
-        # convert to app config
-        pipeline_config = PipelineConfigManager.get_pipeline_config(
-            pipeline=pipeline, workflow=workflow, start_node_id=args.get("start_node_id", "shared")
-        )
-
-        dataset = pipeline.dataset
-        if not dataset:
-            raise ValueError("Pipeline dataset is required")
-
-        # init application generate entity - use RagPipelineGenerateEntity instead
-        application_generate_entity = RagPipelineGenerateEntity(
-            task_id=str(uuid.uuid4()),
-            app_config=pipeline_config,
-            pipeline_config=pipeline_config,
-            datasource_type=args.get("datasource_type", ""),
-            datasource_info=args.get("datasource_info", {}),
-            dataset_id=dataset.id,
-            batch=args.get("batch", ""),
-            document_id=args.get("document_id"),
-            inputs={},
-            files=[],
-            user_id=user.id,
-            stream=streaming,
-            invoke_from=InvokeFrom.DEBUGGER,
-            call_depth=0,
-            workflow_execution_id=str(uuid.uuid4()),
-        )
-        contexts.plugin_tool_providers.set({})
-        contexts.plugin_tool_providers_lock.set(threading.Lock())
-        # Create workflow node execution repository
-        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
-
-        workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
-            session_factory=session_factory,
-            user=user,
-            app_id=application_generate_entity.app_config.app_id,
-            triggered_from=WorkflowRunTriggeredFrom.RAG_PIPELINE_DEBUGGING,
-        )
-
-        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
-            session_factory=session_factory,
-            user=user,
-            app_id=application_generate_entity.app_config.app_id,
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
-        )
-        draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(workflow)
-        var_loader = DraftVarLoader(
-            engine=db.engine,
-            app_id=application_generate_entity.app_config.app_id,
-            tenant_id=application_generate_entity.app_config.tenant_id,
-        )
-
-        return self._generate(
-            flask_app=current_app._get_current_object(),  # type: ignore
-            pipeline=pipeline,
-            workflow_id=workflow.id,
-            user=user,
-            invoke_from=InvokeFrom.DEBUGGER,
-            application_generate_entity=application_generate_entity,
-            workflow_execution_repository=workflow_execution_repository,
-            workflow_node_execution_repository=workflow_node_execution_repository,
-            streaming=streaming,
-            variable_loader=var_loader,
-        )
-
-    def single_loop_generate(
-        self,
-        pipeline: Pipeline,
-        workflow: Workflow,
-        node_id: str,
-        user: Account | EndUser,
-        args: Mapping[str, Any],
-        streaming: bool = True,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
-        """
-        Generate App response.
-
-        :param app_model: App
-        :param workflow: Workflow
-        :param node_id: the node id
-        :param user: account or end user
-        :param args: request args
-        :param streaming: is streamed
-        """
-        if not node_id:
-            raise ValueError("node_id is required")
-
-        if args.get("inputs") is None:
-            raise ValueError("inputs is required")
-
-        dataset = pipeline.dataset
-        if not dataset:
-            raise ValueError("Pipeline dataset is required")
-
-        # convert to app config
-        pipeline_config = PipelineConfigManager.get_pipeline_config(
-            pipeline=pipeline, workflow=workflow, start_node_id=args.get("start_node_id", "shared")
-        )
-
-        # init application generate entity
-        application_generate_entity = RagPipelineGenerateEntity(
-            task_id=str(uuid.uuid4()),
-            app_config=pipeline_config,
-            pipeline_config=pipeline_config,
-            datasource_type=args.get("datasource_type", ""),
-            datasource_info=args.get("datasource_info", {}),
-            batch=args.get("batch", ""),
-            document_id=args.get("document_id"),
-            dataset_id=dataset.id,
-            inputs={},
-            files=[],
-            user_id=user.id,
-            stream=streaming,
-            invoke_from=InvokeFrom.DEBUGGER,
-            extras={"auto_generate_conversation_name": False},
-            single_loop_run=RagPipelineGenerateEntity.SingleLoopRunEntity(node_id=node_id, inputs=args["inputs"]),
-            workflow_execution_id=str(uuid.uuid4()),
-        )
-        contexts.plugin_tool_providers.set({})
-        contexts.plugin_tool_providers_lock.set(threading.Lock())
-
-        # Create workflow node execution repository
-        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
-
-        workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
-            session_factory=session_factory,
-            user=user,
-            app_id=application_generate_entity.app_config.app_id,
-            triggered_from=WorkflowRunTriggeredFrom.RAG_PIPELINE_DEBUGGING,
-        )
-
-        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
-            session_factory=session_factory,
-            user=user,
-            app_id=application_generate_entity.app_config.app_id,
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
-        )
-        draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(workflow)
-        var_loader = DraftVarLoader(
-            engine=db.engine,
-            app_id=application_generate_entity.app_config.app_id,
-            tenant_id=application_generate_entity.app_config.tenant_id,
-        )
-
-        return self._generate(
-            flask_app=current_app._get_current_object(),  # type: ignore
-            pipeline=pipeline,
-            workflow_id=workflow.id,
-            user=user,
-            invoke_from=InvokeFrom.DEBUGGER,
-            application_generate_entity=application_generate_entity,
-            workflow_execution_repository=workflow_execution_repository,
-            workflow_node_execution_repository=workflow_node_execution_repository,
-            streaming=streaming,
-            variable_loader=var_loader,
-        )
-
-    def _generate_worker(
-        self,
-        flask_app: Flask,
-        application_generate_entity: RagPipelineGenerateEntity,
-        queue_manager: AppQueueManager,
-        context: contextvars.Context,
-        variable_loader: VariableLoader,
-        workflow_thread_pool_id: Optional[str] = None,
-    ) -> None:
-        """
-        Generate worker in a new thread.
-        :param flask_app: Flask app
-        :param application_generate_entity: application generate entity
-        :param queue_manager: queue manager
-        :param workflow_thread_pool_id: workflow thread pool id
-        :return:
-        """
-
-        with preserve_flask_contexts(flask_app, context_vars=context):
-            try:
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow = session.scalar(
-                        select(Workflow).where(
-                            Workflow.tenant_id == application_generate_entity.app_config.tenant_id,
-                            Workflow.app_id == application_generate_entity.app_config.app_id,
-                            Workflow.id == application_generate_entity.app_config.workflow_id,
-                        )
-                    )
-                    if workflow is None:
-                        raise ValueError("Workflow not found")
-
-                    # Determine system_user_id based on invocation source
-                    is_external_api_call = application_generate_entity.invoke_from in {
-                        InvokeFrom.WEB_APP,
-                        InvokeFrom.SERVICE_API,
-                    }
-
-                    if is_external_api_call:
-                        # For external API calls, use end user's session ID
-                        end_user = session.scalar(
-                            select(EndUser).where(EndUser.id == application_generate_entity.user_id)
-                        )
-                        system_user_id = end_user.session_id if end_user else ""
-                    else:
-                        # For internal calls, use the original user ID
-                        system_user_id = application_generate_entity.user_id
-                    # workflow app
-                    runner = PipelineRunner(
-                        application_generate_entity=application_generate_entity,
-                        queue_manager=queue_manager,
-                        workflow_thread_pool_id=workflow_thread_pool_id,
-                        variable_loader=variable_loader,
-                        workflow=workflow,
-                        system_user_id=system_user_id,
-                    )
-
-                    runner.run()
-            except GenerateTaskStoppedError:
-                pass
-            except InvokeAuthorizationError:
-                queue_manager.publish_error(
-                    InvokeAuthorizationError("Incorrect API key provided"), PublishFrom.APPLICATION_MANAGER
-                )
-            except ValidationError as e:
-                logger.exception("Validation Error when generating")
-                queue_manager.publish_error(e, PublishFrom.APPLICATION_MANAGER)
-            except ValueError as e:
-                if dify_config.DEBUG:
-                    logger.exception("Error when generating")
-                queue_manager.publish_error(e, PublishFrom.APPLICATION_MANAGER)
-            except Exception as e:
-                logger.exception("Unknown Error when generating")
-                queue_manager.publish_error(e, PublishFrom.APPLICATION_MANAGER)
-            finally:
-                db.session.close()
-
-    def _handle_response(
-        self,
-        application_generate_entity: RagPipelineGenerateEntity,
-        workflow: Workflow,
-        queue_manager: AppQueueManager,
-        user: Union[Account, EndUser],
-        workflow_execution_repository: WorkflowExecutionRepository,
-        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
-        draft_var_saver_factory: DraftVariableSaverFactory,
-        stream: bool = False,
-    ) -> Union[WorkflowAppBlockingResponse, Generator[WorkflowAppStreamResponse, None, None]]:
-        """
-        Handle response.
-        :param application_generate_entity: application generate entity
-        :param workflow: workflow
-        :param queue_manager: queue manager
-        :param user: account or end user
-        :param stream: is stream
-        :param workflow_node_execution_repository: optional repository for workflow node execution
-        :return:
-        """
-        # init generate task pipeline
-        generate_task_pipeline = WorkflowAppGenerateTaskPipeline(
-            application_generate_entity=application_generate_entity,
-            workflow=workflow,
-            queue_manager=queue_manager,
-            user=user,
-            stream=stream,
-            workflow_node_execution_repository=workflow_node_execution_repository,
-            workflow_execution_repository=workflow_execution_repository,
-            draft_var_saver_factory=draft_var_saver_factory,
-        )
-
-        try:
-            return generate_task_pipeline.process()
-        except ValueError as e:
-            if len(e.args) > 0 and e.args[0] == "I/O operation on closed file.":  # ignore this error
-                raise GenerateTaskStoppedError()
-            else:
-                logger.exception(
-                    "Fails to process generate task pipeline, task_id: %r",
-                    application_generate_entity.task_id,
-                )
-                raise e
-
-    def _build_document(
-        self,
-        tenant_id: str,
-        dataset_id: str,
-        built_in_field_enabled: bool,
-        datasource_type: str,
-        datasource_info: Mapping[str, Any],
-        created_from: str,
-        position: int,
-        account: Union[Account, EndUser],
-        batch: str,
-        document_form: str,
-    ):
-        if datasource_type == "local_file":
-            name = datasource_info["name"]
-        elif datasource_type == "online_document":
-            name = datasource_info["page"]["page_name"]
-        elif datasource_type == "website_crawl":
-            name = datasource_info["title"]
-        elif datasource_type == "online_drive":
-            name = datasource_info["key"]
-        else:
-            raise ValueError(f"Unsupported datasource type: {datasource_type}")
-
-        document = Document(
-            tenant_id=tenant_id,
-            dataset_id=dataset_id,
-            position=position,
-            data_source_type=datasource_type,
-            data_source_info=json.dumps(datasource_info),
-            batch=batch,
-            name=name,
-            created_from=created_from,
-            created_by=account.id,
-            doc_form=document_form,
-        )
-        doc_metadata = {}
-        if built_in_field_enabled:
-            doc_metadata = {
-                BuiltInField.document_name: name,
-                BuiltInField.uploader: account.name,
-                BuiltInField.upload_date: datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%d %H:%M:%S"),
-                BuiltInField.last_update_date: datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%d %H:%M:%S"),
-                BuiltInField.source: datasource_type,
-            }
-        if doc_metadata:
-            document.doc_metadata = doc_metadata
-        return document
-
-    def _format_datasource_info_list(
-        self,
-        datasource_type: str,
-        datasource_info_list: list[Mapping[str, Any]],
-        pipeline: Pipeline,
-        workflow: Workflow,
-        start_node_id: str,
-        user: Union[Account, EndUser],
-    ) -> list[Mapping[str, Any]]:
-        """
-        Format datasource info list.
-        """
-        if datasource_type == "online_drive":
-            all_files = []
-            datasource_node_data = None
-            datasource_nodes = workflow.graph_dict.get("nodes", [])
-            for datasource_node in datasource_nodes:
-                if datasource_node.get("id") == start_node_id:
-                    datasource_node_data = datasource_node.get("data", {})
-                    break
-            if not datasource_node_data:
-                raise ValueError("Datasource node data not found")
-
-            from core.datasource.datasource_manager import DatasourceManager
-
-            datasource_runtime = DatasourceManager.get_datasource_runtime(
-                provider_id=f"{datasource_node_data.get('plugin_id')}/{datasource_node_data.get('provider_name')}",
-                datasource_name=datasource_node_data.get("datasource_name"),
-                tenant_id=pipeline.tenant_id,
-                datasource_type=DatasourceProviderType(datasource_type),
-            )
-            datasource_provider_service = DatasourceProviderService()
-            credentials = datasource_provider_service.get_datasource_credentials(
-                tenant_id=pipeline.tenant_id,
-                provider=datasource_node_data.get("provider_name"),
-                plugin_id=datasource_node_data.get("plugin_id"),
-                credential_id=datasource_node_data.get("credential_id"),
-            )
-            if credentials:
-                datasource_runtime.runtime.credentials = credentials
-            datasource_runtime = cast(OnlineDriveDatasourcePlugin, datasource_runtime)
-
-            for datasource_info in datasource_info_list:
-                if datasource_info.get("id") and datasource_info.get("type") == "folder":
-                    # get all files in the folder
-                    self._get_files_in_folder(
-                        datasource_runtime,
-                        datasource_info.get("id", ""),
-                        datasource_info.get("bucket", None),
-                        user.id,
-                        all_files,
-                        datasource_info,
-                        None,
-                    )
-                else:
-                    all_files.append(
-                        {
-                            "id": datasource_info.get("id", ""),
-                            "bucket": datasource_info.get("bucket", None),
-                        }
-                    )
-            return all_files
-        else:
-            return datasource_info_list
-
-    def _get_files_in_folder(
-        self,
-        datasource_runtime: OnlineDriveDatasourcePlugin,
-        prefix: str,
-        bucket: Optional[str],
-        user_id: str,
-        all_files: list,
-        datasource_info: Mapping[str, Any],
-        next_page_parameters: Optional[dict] = None,
-    ):
-        """
-        Get files in a folder.
-        """
-        result_generator = datasource_runtime.online_drive_browse_files(
-            user_id=user_id,
-            request=OnlineDriveBrowseFilesRequest(
-                bucket=bucket,
-                prefix=prefix,
-                max_keys=20,
-                next_page_parameters=next_page_parameters,
-            ),
-            provider_type=datasource_runtime.datasource_provider_type(),
-        )
-        is_truncated = False
-        last_file_key = None
-        for result in result_generator:
-            for files in result.result:
-                for file in files.files:
-                    if file.type == "folder":
-                        self._get_files_in_folder(
-                            datasource_runtime,
-                            file.id,
-                            bucket,
-                            user_id,
-                            all_files,
-                            datasource_info,
-                            None,
-                        )
-                    else:
-                        all_files.append(
-                            {
-                                "id": file.id,
-                                "bucket": bucket,
-                            }
-                        )
-                is_truncated = files.is_truncated
-                next_page_parameters = files.next_page_parameters
-
-        if is_truncated:
-            self._get_files_in_folder(
-                datasource_runtime, prefix, bucket, user_id, all_files, datasource_info, next_page_parameters
-            )

api/core/app/apps/pipeline/pipeline_queue_manager.py

@@ -1,45 +0,0 @@
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.exc import GenerateTaskStoppedError
-from core.app.entities.app_invoke_entities import InvokeFrom
-from core.app.entities.queue_entities import (
-    AppQueueEvent,
-    QueueErrorEvent,
-    QueueMessageEndEvent,
-    QueueStopEvent,
-    QueueWorkflowFailedEvent,
-    QueueWorkflowPartialSuccessEvent,
-    QueueWorkflowSucceededEvent,
-    WorkflowQueueMessage,
-)
-
-
-class PipelineQueueManager(AppQueueManager):
-    def __init__(self, task_id: str, user_id: str, invoke_from: InvokeFrom, app_mode: str) -> None:
-        super().__init__(task_id, user_id, invoke_from)
-
-        self._app_mode = app_mode
-
-    def _publish(self, event: AppQueueEvent, pub_from: PublishFrom) -> None:
-        """
-        Publish event to queue
-        :param event:
-        :param pub_from:
-        :return:
-        """
-        message = WorkflowQueueMessage(task_id=self._task_id, app_mode=self._app_mode, event=event)
-
-        self._q.put(message)
-
-        if isinstance(
-            event,
-            QueueStopEvent
-            | QueueErrorEvent
-            | QueueMessageEndEvent
-            | QueueWorkflowSucceededEvent
-            | QueueWorkflowFailedEvent
-            | QueueWorkflowPartialSuccessEvent,
-        ):
-            self.stop_listen()
-
-        if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
-            raise GenerateTaskStoppedError()

api/core/app/apps/pipeline/pipeline_runner.py

@@ -1,252 +0,0 @@
-import logging
-from collections.abc import Mapping
-from typing import Any, Optional, cast
-
-from configs import dify_config
-from core.app.apps.base_app_queue_manager import AppQueueManager
-from core.app.apps.pipeline.pipeline_config_manager import PipelineConfig
-from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
-from core.app.entities.app_invoke_entities import (
-    InvokeFrom,
-    RagPipelineGenerateEntity,
-)
-from core.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
-from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
-from core.workflow.entities.variable_pool import VariablePool
-from core.workflow.graph_engine.entities.event import GraphEngineEvent, GraphRunFailedEvent
-from core.workflow.graph_engine.entities.graph import Graph
-from core.workflow.system_variable import SystemVariable
-from core.workflow.variable_loader import VariableLoader
-from core.workflow.workflow_entry import WorkflowEntry
-from extensions.ext_database import db
-from models.dataset import Document, Pipeline
-from models.enums import UserFrom
-from models.model import EndUser
-from models.workflow import Workflow, WorkflowType
-
-logger = logging.getLogger(__name__)
-
-
-class PipelineRunner(WorkflowBasedAppRunner):
-    """
-    Pipeline Application Runner
-    """
-
-    def __init__(
-        self,
-        application_generate_entity: RagPipelineGenerateEntity,
-        queue_manager: AppQueueManager,
-        variable_loader: VariableLoader,
-        workflow: Workflow,
-        system_user_id: str,
-        workflow_thread_pool_id: Optional[str] = None,
-    ) -> None:
-        """
-        :param application_generate_entity: application generate entity
-        :param queue_manager: application queue manager
-        :param workflow_thread_pool_id: workflow thread pool id
-        """
-        super().__init__(
-            queue_manager=queue_manager,
-            variable_loader=variable_loader,
-            app_id=application_generate_entity.app_config.app_id,
-        )
-        self.application_generate_entity = application_generate_entity
-        self.workflow_thread_pool_id = workflow_thread_pool_id
-        self._workflow = workflow
-        self._sys_user_id = system_user_id
-
-    def _get_app_id(self) -> str:
-        return self.application_generate_entity.app_config.app_id
-
-    def run(self) -> None:
-        """
-        Run application
-        """
-        app_config = self.application_generate_entity.app_config
-        app_config = cast(PipelineConfig, app_config)
-
-        user_id = None
-        if self.application_generate_entity.invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
-            end_user = db.session.query(EndUser).filter(EndUser.id == self.application_generate_entity.user_id).first()
-            if end_user:
-                user_id = end_user.session_id
-        else:
-            user_id = self.application_generate_entity.user_id
-
-        pipeline = db.session.query(Pipeline).filter(Pipeline.id == app_config.app_id).first()
-        if not pipeline:
-            raise ValueError("Pipeline not found")
-
-        workflow = self.get_workflow(pipeline=pipeline, workflow_id=app_config.workflow_id)
-        if not workflow:
-            raise ValueError("Workflow not initialized")
-
-        db.session.close()
-
-        workflow_callbacks: list[WorkflowCallback] = []
-        if dify_config.DEBUG:
-            workflow_callbacks.append(WorkflowLoggingCallback())
-
-        # if only single iteration run is requested
-        if self.application_generate_entity.single_iteration_run:
-            # if only single iteration run is requested
-            graph, variable_pool = self._get_graph_and_variable_pool_of_single_iteration(
-                workflow=workflow,
-                node_id=self.application_generate_entity.single_iteration_run.node_id,
-                user_inputs=self.application_generate_entity.single_iteration_run.inputs,
-            )
-        elif self.application_generate_entity.single_loop_run:
-            # if only single loop run is requested
-            graph, variable_pool = self._get_graph_and_variable_pool_of_single_loop(
-                workflow=workflow,
-                node_id=self.application_generate_entity.single_loop_run.node_id,
-                user_inputs=self.application_generate_entity.single_loop_run.inputs,
-            )
-        else:
-            inputs = self.application_generate_entity.inputs
-            files = self.application_generate_entity.files
-
-            # Create a variable pool.
-            system_inputs = SystemVariable(
-                files=files,
-                user_id=user_id,
-                app_id=app_config.app_id,
-                workflow_id=app_config.workflow_id,
-                workflow_execution_id=self.application_generate_entity.workflow_execution_id,
-                document_id=self.application_generate_entity.document_id,
-                batch=self.application_generate_entity.batch,
-                dataset_id=self.application_generate_entity.dataset_id,
-                datasource_type=self.application_generate_entity.datasource_type,
-                datasource_info=self.application_generate_entity.datasource_info,
-                invoke_from=self.application_generate_entity.invoke_from.value,
-            )
-            rag_pipeline_variables = []
-            if workflow.rag_pipeline_variables:
-                for v in workflow.rag_pipeline_variables:
-                    rag_pipeline_variable = RAGPipelineVariable(**v)
-                    if (
-                        rag_pipeline_variable.belong_to_node_id
-                        in (self.application_generate_entity.start_node_id, "shared")
-                    ) and rag_pipeline_variable.variable in inputs:
-                        rag_pipeline_variables.append(
-                            RAGPipelineVariableInput(
-                                variable=rag_pipeline_variable,
-                                value=inputs[rag_pipeline_variable.variable],
-                            )
-                        )
-
-            variable_pool = VariablePool(
-                system_variables=system_inputs,
-                user_inputs=inputs,
-                environment_variables=workflow.environment_variables,
-                conversation_variables=[],
-                rag_pipeline_variables=rag_pipeline_variables,
-            )
-
-            # init graph
-            graph = self._init_rag_pipeline_graph(
-                graph_config=workflow.graph_dict,
-                start_node_id=self.application_generate_entity.start_node_id,
-            )
-
-        # RUN WORKFLOW
-        workflow_entry = WorkflowEntry(
-            tenant_id=workflow.tenant_id,
-            app_id=workflow.app_id,
-            workflow_id=workflow.id,
-            workflow_type=WorkflowType.value_of(workflow.type),
-            graph=graph,
-            graph_config=workflow.graph_dict,
-            user_id=self.application_generate_entity.user_id,
-            user_from=(
-                UserFrom.ACCOUNT
-                if self.application_generate_entity.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                else UserFrom.END_USER
-            ),
-            invoke_from=self.application_generate_entity.invoke_from,
-            call_depth=self.application_generate_entity.call_depth,
-            variable_pool=variable_pool,
-            thread_pool_id=self.workflow_thread_pool_id,
-        )
-
-        generator = workflow_entry.run(callbacks=workflow_callbacks)
-
-        for event in generator:
-            self._update_document_status(
-                event, self.application_generate_entity.document_id, self.application_generate_entity.dataset_id
-            )
-            self._handle_event(workflow_entry, event)
-
-    def get_workflow(self, pipeline: Pipeline, workflow_id: str) -> Optional[Workflow]:
-        """
-        Get workflow
-        """
-        # fetch workflow by workflow_id
-        workflow = (
-            db.session.query(Workflow)
-            .filter(
-                Workflow.tenant_id == pipeline.tenant_id, Workflow.app_id == pipeline.id, Workflow.id == workflow_id
-            )
-            .first()
-        )
-
-        # return workflow
-        return workflow
-
-    def _init_rag_pipeline_graph(self, graph_config: Mapping[str, Any], start_node_id: Optional[str] = None) -> Graph:
-        """
-        Init pipeline graph
-        """
-        if "nodes" not in graph_config or "edges" not in graph_config:
-            raise ValueError("nodes or edges not found in workflow graph")
-
-        if not isinstance(graph_config.get("nodes"), list):
-            raise ValueError("nodes in workflow graph must be a list")
-
-        if not isinstance(graph_config.get("edges"), list):
-            raise ValueError("edges in workflow graph must be a list")
-        nodes = graph_config.get("nodes", [])
-        edges = graph_config.get("edges", [])
-        real_run_nodes = []
-        real_edges = []
-        exclude_node_ids = []
-        for node in nodes:
-            node_id = node.get("id")
-            node_type = node.get("data", {}).get("type", "")
-            if node_type == "datasource":
-                if start_node_id != node_id:
-                    exclude_node_ids.append(node_id)
-                    continue
-            real_run_nodes.append(node)
-        for edge in edges:
-            if edge.get("source") in exclude_node_ids:
-                continue
-            real_edges.append(edge)
-        graph_config = dict(graph_config)
-        graph_config["nodes"] = real_run_nodes
-        graph_config["edges"] = real_edges
-        # init graph
-        graph = Graph.init(graph_config=graph_config)
-
-        if not graph:
-            raise ValueError("graph not found in workflow")
-
-        return graph
-
-    def _update_document_status(self, event: GraphEngineEvent, document_id: str | None, dataset_id: str | None) -> None:
-        """
-        Update document status
-        """
-        if isinstance(event, GraphRunFailedEvent):
-            if document_id and dataset_id:
-                document = (
-                    db.session.query(Document)
-                    .filter(Document.id == document_id, Document.dataset_id == dataset_id)
-                    .first()
-                )
-                if document:
-                    document.indexing_status = "error"
-                    document.error = event.error or "Unknown error"
-                    db.session.add(document)
-                    db.session.commit()

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -711,7 +711,7 @@ class WorkflowAppGenerateTaskPipeline:
         # Initialize graph runtime state
         graph_runtime_state = None
 
-        for queue_message in self._base_task_pipeline.queue_manager.listen():
+        for queue_message in self._base_task_pipeline._queue_manager.listen():
             event = queue_message.event
 
             match event:

api/core/app/entities/app_invoke_entities.py

@@ -35,7 +35,6 @@ class InvokeFrom(Enum):
     # DEBUGGER indicates that this invocation is from
     # the workflow (or chatflow) edit page.
     DEBUGGER = "debugger"
-    PUBLISHED = "published"
 
     @classmethod
     def value_of(cls, value: str):
@@ -241,38 +240,3 @@ class WorkflowAppGenerateEntity(AppGenerateEntity):
         inputs: dict
 
     single_loop_run: Optional[SingleLoopRunEntity] = None
-
-
-class RagPipelineGenerateEntity(WorkflowAppGenerateEntity):
-    """
-    RAG Pipeline Application Generate Entity.
-    """
-
-    # pipeline config
-    pipeline_config: WorkflowUIBasedAppConfig
-    datasource_type: str
-    datasource_info: Mapping[str, Any]
-    dataset_id: str
-    batch: str
-    document_id: Optional[str] = None
-    start_node_id: Optional[str] = None
-
-    class SingleIterationRunEntity(BaseModel):
-        """
-        Single Iteration Run Entity.
-        """
-
-        node_id: str
-        inputs: dict
-
-    single_iteration_run: Optional[SingleIterationRunEntity] = None
-
-    class SingleLoopRunEntity(BaseModel):
-        """
-        Single Loop Run Entity.
-        """
-
-        node_id: str
-        inputs: dict
-
-    single_loop_run: Optional[SingleLoopRunEntity] = None

api/core/app/task_pipeline/based_generate_task_pipeline.py

@@ -37,7 +37,7 @@ class BasedGenerateTaskPipeline:
         stream: bool,
     ) -> None:
         self._application_generate_entity = application_generate_entity
-        self.queue_manager = queue_manager
+        self._queue_manager = queue_manager
         self._start_at = time.perf_counter()
         self._output_moderation_handler = self._init_output_moderation()
         self._stream = stream
@@ -113,7 +113,7 @@ class BasedGenerateTaskPipeline:
                 tenant_id=app_config.tenant_id,
                 app_id=app_config.app_id,
                 rule=ModerationRule(type=sensitive_word_avoidance.type, config=sensitive_word_avoidance.config),
-                queue_manager=self.queue_manager,
+                queue_manager=self._queue_manager,
             )
         return None
 

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -57,7 +57,6 @@ from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from events.message_event import message_was_created
 from extensions.ext_database import db
-from libs.datetime_utils import naive_utc_now
 from models.model import AppMode, Conversation, Message, MessageAgentThought
 
 logger = logging.getLogger(__name__)
@@ -258,7 +257,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         Process stream response.
         :return:
         """
-        for message in self.queue_manager.listen():
+        for message in self._queue_manager.listen():
             if publisher:
                 publisher.publish(message)
             event = message.event
@@ -390,7 +389,6 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
             if llm_result.message.content
             else ""
         )
-        message.updated_at = naive_utc_now()
         message.answer_tokens = usage.completion_tokens
         message.answer_unit_price = usage.completion_unit_price
         message.answer_price_unit = usage.completion_price_unit
@@ -501,7 +499,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
             if self._output_moderation_handler.should_direct_output():
                 # stop subscribe new token when output moderation should direct output
                 self._task_state.llm_result.message.content = self._output_moderation_handler.get_final_output()
-                self.queue_manager.publish(
+                self._queue_manager.publish(
                     QueueLLMChunkEvent(
                         chunk=LLMResultChunk(
                             model=self._task_state.llm_result.model,
@@ -515,7 +513,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                     PublishFrom.TASK_PIPELINE,
                 )
 
-                self.queue_manager.publish(
+                self._queue_manager.publish(
                     QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE
                 )
                 return True

api/core/callback_handler/agent_tool_callback_handler.py

@@ -105,14 +105,6 @@ class DifyAgentCallbackHandler(BaseModel):
 
         self.current_loop += 1
 
-    def on_datasource_start(self, datasource_name: str, datasource_inputs: Mapping[str, Any]) -> None:
-        """Run on datasource start."""
-        if dify_config.DEBUG:
-            print_text(
-                "\n[on_datasource_start] DatasourceCall:" + datasource_name + "\n" + str(datasource_inputs) + "\n",
-                color=self.color,
-            )
-
     @property
     def ignore_agent(self) -> bool:
         """Whether to ignore agent callbacks."""

api/core/datasource/__base/datasource_plugin.py

@@ -1,33 +0,0 @@
-from abc import ABC, abstractmethod
-
-from core.datasource.__base.datasource_runtime import DatasourceRuntime
-from core.datasource.entities.datasource_entities import (
-    DatasourceEntity,
-    DatasourceProviderType,
-)
-
-
-class DatasourcePlugin(ABC):
-    entity: DatasourceEntity
-    runtime: DatasourceRuntime
-
-    def __init__(
-        self,
-        entity: DatasourceEntity,
-        runtime: DatasourceRuntime,
-    ) -> None:
-        self.entity = entity
-        self.runtime = runtime
-
-    @abstractmethod
-    def datasource_provider_type(self) -> str:
-        """
-        returns the type of the datasource provider
-        """
-        return DatasourceProviderType.LOCAL_FILE
-
-    def fork_datasource_runtime(self, runtime: DatasourceRuntime) -> "DatasourcePlugin":
-        return self.__class__(
-            entity=self.entity.model_copy(),
-            runtime=runtime,
-        )

api/core/datasource/__base/datasource_provider.py

@@ -1,118 +0,0 @@
-from abc import ABC, abstractmethod
-from typing import Any
-
-from core.datasource.__base.datasource_plugin import DatasourcePlugin
-from core.datasource.entities.datasource_entities import DatasourceProviderEntityWithPlugin, DatasourceProviderType
-from core.entities.provider_entities import ProviderConfig
-from core.plugin.impl.tool import PluginToolManager
-from core.tools.errors import ToolProviderCredentialValidationError
-
-
-class DatasourcePluginProviderController(ABC):
-    entity: DatasourceProviderEntityWithPlugin
-    tenant_id: str
-
-    def __init__(self, entity: DatasourceProviderEntityWithPlugin, tenant_id: str) -> None:
-        self.entity = entity
-        self.tenant_id = tenant_id
-
-    @property
-    def need_credentials(self) -> bool:
-        """
-        returns whether the provider needs credentials
-
-        :return: whether the provider needs credentials
-        """
-        return self.entity.credentials_schema is not None and len(self.entity.credentials_schema) != 0
-
-    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]) -> None:
-        """
-        validate the credentials of the provider
-        """
-        manager = PluginToolManager()
-        if not manager.validate_datasource_credentials(
-            tenant_id=self.tenant_id,
-            user_id=user_id,
-            provider=self.entity.identity.name,
-            credentials=credentials,
-        ):
-            raise ToolProviderCredentialValidationError("Invalid credentials")
-
-    @property
-    def provider_type(self) -> DatasourceProviderType:
-        """
-        returns the type of the provider
-        """
-        return DatasourceProviderType.LOCAL_FILE
-
-    @abstractmethod
-    def get_datasource(self, datasource_name: str) -> DatasourcePlugin:
-        """
-        return datasource with given name
-        """
-        pass
-
-    def validate_credentials_format(self, credentials: dict[str, Any]) -> None:
-        """
-        validate the format of the credentials of the provider and set the default value if needed
-
-        :param credentials: the credentials of the tool
-        """
-        credentials_schema = dict[str, ProviderConfig]()
-        if credentials_schema is None:
-            return
-
-        for credential in self.entity.credentials_schema:
-            credentials_schema[credential.name] = credential
-
-        credentials_need_to_validate: dict[str, ProviderConfig] = {}
-        for credential_name in credentials_schema:
-            credentials_need_to_validate[credential_name] = credentials_schema[credential_name]
-
-        for credential_name in credentials:
-            if credential_name not in credentials_need_to_validate:
-                raise ToolProviderCredentialValidationError(
-                    f"credential {credential_name} not found in provider {self.entity.identity.name}"
-                )
-
-            # check type
-            credential_schema = credentials_need_to_validate[credential_name]
-            if not credential_schema.required and credentials[credential_name] is None:
-                continue
-
-            if credential_schema.type in {ProviderConfig.Type.SECRET_INPUT, ProviderConfig.Type.TEXT_INPUT}:
-                if not isinstance(credentials[credential_name], str):
-                    raise ToolProviderCredentialValidationError(f"credential {credential_name} should be string")
-
-            elif credential_schema.type == ProviderConfig.Type.SELECT:
-                if not isinstance(credentials[credential_name], str):
-                    raise ToolProviderCredentialValidationError(f"credential {credential_name} should be string")
-
-                options = credential_schema.options
-                if not isinstance(options, list):
-                    raise ToolProviderCredentialValidationError(f"credential {credential_name} options should be list")
-
-                if credentials[credential_name] not in [x.value for x in options]:
-                    raise ToolProviderCredentialValidationError(
-                        f"credential {credential_name} should be one of {options}"
-                    )
-
-            credentials_need_to_validate.pop(credential_name)
-
-        for credential_name in credentials_need_to_validate:
-            credential_schema = credentials_need_to_validate[credential_name]
-            if credential_schema.required:
-                raise ToolProviderCredentialValidationError(f"credential {credential_name} is required")
-
-            # the credential is not set currently, set the default value if needed
-            if credential_schema.default is not None:
-                default_value = credential_schema.default
-                # parse default value into the correct type
-                if credential_schema.type in {
-                    ProviderConfig.Type.SECRET_INPUT,
-                    ProviderConfig.Type.TEXT_INPUT,
-                    ProviderConfig.Type.SELECT,
-                }:
-                    default_value = str(default_value)
-
-                credentials[credential_name] = default_value

api/core/datasource/__base/datasource_runtime.py

@@ -1,36 +0,0 @@
-from typing import Any, Optional
-
-from openai import BaseModel
-from pydantic import Field
-
-from core.app.entities.app_invoke_entities import InvokeFrom
-from core.datasource.entities.datasource_entities import DatasourceInvokeFrom
-
-
-class DatasourceRuntime(BaseModel):
-    """
-    Meta data of a datasource call processing
-    """
-
-    tenant_id: str
-    datasource_id: Optional[str] = None
-    invoke_from: Optional[InvokeFrom] = None
-    datasource_invoke_from: Optional[DatasourceInvokeFrom] = None
-    credentials: dict[str, Any] = Field(default_factory=dict)
-    runtime_parameters: dict[str, Any] = Field(default_factory=dict)
-
-
-class FakeDatasourceRuntime(DatasourceRuntime):
-    """
-    Fake datasource runtime for testing
-    """
-
-    def __init__(self):
-        super().__init__(
-            tenant_id="fake_tenant_id",
-            datasource_id="fake_datasource_id",
-            invoke_from=InvokeFrom.DEBUGGER,
-            datasource_invoke_from=DatasourceInvokeFrom.RAG_PIPELINE,
-            credentials={},
-            runtime_parameters={},
-        )

api/core/datasource/datasource_file_manager.py

@@ -1,247 +0,0 @@
-import base64
-import hashlib
-import hmac
-import logging
-import os
-import time
-from datetime import datetime
-from mimetypes import guess_extension, guess_type
-from typing import Optional, Union
-from uuid import uuid4
-
-import httpx
-
-from configs import dify_config
-from core.helper import ssrf_proxy
-from extensions.ext_database import db
-from extensions.ext_storage import storage
-from models.enums import CreatorUserRole
-from models.model import MessageFile, UploadFile
-from models.tools import ToolFile
-
-logger = logging.getLogger(__name__)
-
-
-class DatasourceFileManager:
-    @staticmethod
-    def sign_file(datasource_file_id: str, extension: str) -> str:
-        """
-        sign file to get a temporary url
-        """
-        base_url = dify_config.FILES_URL
-        file_preview_url = f"{base_url}/files/datasources/{datasource_file_id}{extension}"
-
-        timestamp = str(int(time.time()))
-        nonce = os.urandom(16).hex()
-        data_to_sign = f"file-preview|{datasource_file_id}|{timestamp}|{nonce}"
-        secret_key = dify_config.SECRET_KEY.encode() if dify_config.SECRET_KEY else b""
-        sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
-        encoded_sign = base64.urlsafe_b64encode(sign).decode()
-
-        return f"{file_preview_url}?timestamp={timestamp}&nonce={nonce}&sign={encoded_sign}"
-
-    @staticmethod
-    def verify_file(datasource_file_id: str, timestamp: str, nonce: str, sign: str) -> bool:
-        """
-        verify signature
-        """
-        data_to_sign = f"file-preview|{datasource_file_id}|{timestamp}|{nonce}"
-        secret_key = dify_config.SECRET_KEY.encode() if dify_config.SECRET_KEY else b""
-        recalculated_sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
-        recalculated_encoded_sign = base64.urlsafe_b64encode(recalculated_sign).decode()
-
-        # verify signature
-        if sign != recalculated_encoded_sign:
-            return False
-
-        current_time = int(time.time())
-        return current_time - int(timestamp) <= dify_config.FILES_ACCESS_TIMEOUT
-
-    @staticmethod
-    def create_file_by_raw(
-        *,
-        user_id: str,
-        tenant_id: str,
-        conversation_id: Optional[str],
-        file_binary: bytes,
-        mimetype: str,
-        filename: Optional[str] = None,
-    ) -> UploadFile:
-        extension = guess_extension(mimetype) or ".bin"
-        unique_name = uuid4().hex
-        unique_filename = f"{unique_name}{extension}"
-        # default just as before
-        present_filename = unique_filename
-        if filename is not None:
-            has_extension = len(filename.split(".")) > 1
-            # Add extension flexibly
-            present_filename = filename if has_extension else f"{filename}{extension}"
-        filepath = f"datasources/{tenant_id}/{unique_filename}"
-        storage.save(filepath, file_binary)
-
-        upload_file = UploadFile(
-            tenant_id=tenant_id,
-            storage_type=dify_config.STORAGE_TYPE,
-            key=filepath,
-            name=present_filename,
-            size=len(file_binary),
-            extension=extension,
-            mime_type=mimetype,
-            created_by_role=CreatorUserRole.ACCOUNT,
-            created_by=user_id,
-            used=False,
-            hash=hashlib.sha3_256(file_binary).hexdigest(),
-            source_url="",
-            created_at=datetime.now(),
-        )
-
-        db.session.add(upload_file)
-        db.session.commit()
-        db.session.refresh(upload_file)
-
-        return upload_file
-
-    @staticmethod
-    def create_file_by_url(
-        user_id: str,
-        tenant_id: str,
-        file_url: str,
-        conversation_id: Optional[str] = None,
-    ) -> UploadFile:
-        # try to download image
-        try:
-            response = ssrf_proxy.get(file_url)
-            response.raise_for_status()
-            blob = response.content
-        except httpx.TimeoutException:
-            raise ValueError(f"timeout when downloading file from {file_url}")
-
-        mimetype = (
-            guess_type(file_url)[0]
-            or response.headers.get("Content-Type", "").split(";")[0].strip()
-            or "application/octet-stream"
-        )
-        extension = guess_extension(mimetype) or ".bin"
-        unique_name = uuid4().hex
-        filename = f"{unique_name}{extension}"
-        filepath = f"tools/{tenant_id}/{filename}"
-        storage.save(filepath, blob)
-
-        upload_file = UploadFile(
-            tenant_id=tenant_id,
-            storage_type=dify_config.STORAGE_TYPE,
-            key=filepath,
-            name=filename,
-            size=len(blob),
-            extension=extension,
-            mime_type=mimetype,
-            created_by_role=CreatorUserRole.ACCOUNT,
-            created_by=user_id,
-            used=False,
-            hash=hashlib.sha3_256(blob).hexdigest(),
-            source_url=file_url,
-            created_at=datetime.now(),
-        )
-
-        db.session.add(upload_file)
-        db.session.commit()
-
-        return upload_file
-
-    @staticmethod
-    def get_file_binary(id: str) -> Union[tuple[bytes, str], None]:
-        """
-        get file binary
-
-        :param id: the id of the file
-
-        :return: the binary of the file, mime type
-        """
-        upload_file: UploadFile | None = (
-            db.session.query(UploadFile)
-            .filter(
-                UploadFile.id == id,
-            )
-            .first()
-        )
-
-        if not upload_file:
-            return None
-
-        blob = storage.load_once(upload_file.key)
-
-        return blob, upload_file.mime_type
-
-    @staticmethod
-    def get_file_binary_by_message_file_id(id: str) -> Union[tuple[bytes, str], None]:
-        """
-        get file binary
-
-        :param id: the id of the file
-
-        :return: the binary of the file, mime type
-        """
-        message_file: MessageFile | None = (
-            db.session.query(MessageFile)
-            .filter(
-                MessageFile.id == id,
-            )
-            .first()
-        )
-
-        # Check if message_file is not None
-        if message_file is not None:
-            # get tool file id
-            if message_file.url is not None:
-                tool_file_id = message_file.url.split("/")[-1]
-                # trim extension
-                tool_file_id = tool_file_id.split(".")[0]
-            else:
-                tool_file_id = None
-        else:
-            tool_file_id = None
-
-        tool_file: ToolFile | None = (
-            db.session.query(ToolFile)
-            .filter(
-                ToolFile.id == tool_file_id,
-            )
-            .first()
-        )
-
-        if not tool_file:
-            return None
-
-        blob = storage.load_once(tool_file.file_key)
-
-        return blob, tool_file.mimetype
-
-    @staticmethod
-    def get_file_generator_by_upload_file_id(upload_file_id: str):
-        """
-        get file binary
-
-        :param tool_file_id: the id of the tool file
-
-        :return: the binary of the file, mime type
-        """
-        upload_file: UploadFile | None = (
-            db.session.query(UploadFile)
-            .filter(
-                UploadFile.id == upload_file_id,
-            )
-            .first()
-        )
-
-        if not upload_file:
-            return None, None
-
-        stream = storage.load_stream(upload_file.key)
-
-        return stream, upload_file.mime_type
-
-
-# init tool_file_parser
-# from core.file.datasource_file_parser import datasource_file_manager
-#
-# datasource_file_manager["manager"] = DatasourceFileManager

api/core/datasource/datasource_manager.py

@@ -1,108 +0,0 @@
-import logging
-from threading import Lock
-from typing import Union
-
-import contexts
-from core.datasource.__base.datasource_plugin import DatasourcePlugin
-from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
-from core.datasource.entities.common_entities import I18nObject
-from core.datasource.entities.datasource_entities import DatasourceProviderType
-from core.datasource.errors import DatasourceProviderNotFoundError
-from core.datasource.local_file.local_file_provider import LocalFileDatasourcePluginProviderController
-from core.datasource.online_document.online_document_provider import OnlineDocumentDatasourcePluginProviderController
-from core.datasource.online_drive.online_drive_provider import OnlineDriveDatasourcePluginProviderController
-from core.datasource.website_crawl.website_crawl_provider import WebsiteCrawlDatasourcePluginProviderController
-from core.plugin.impl.datasource import PluginDatasourceManager
-
-logger = logging.getLogger(__name__)
-
-
-class DatasourceManager:
-    _builtin_provider_lock = Lock()
-    _hardcoded_providers: dict[str, DatasourcePluginProviderController] = {}
-    _builtin_providers_loaded = False
-    _builtin_tools_labels: dict[str, Union[I18nObject, None]] = {}
-
-    @classmethod
-    def get_datasource_plugin_provider(
-        cls, provider_id: str, tenant_id: str, datasource_type: DatasourceProviderType
-    ) -> DatasourcePluginProviderController:
-        """
-        get the datasource plugin provider
-        """
-        # check if context is set
-        try:
-            contexts.datasource_plugin_providers.get()
-        except LookupError:
-            contexts.datasource_plugin_providers.set({})
-            contexts.datasource_plugin_providers_lock.set(Lock())
-
-        with contexts.datasource_plugin_providers_lock.get():
-            datasource_plugin_providers = contexts.datasource_plugin_providers.get()
-            if provider_id in datasource_plugin_providers:
-                return datasource_plugin_providers[provider_id]
-
-            manager = PluginDatasourceManager()
-            provider_entity = manager.fetch_datasource_provider(tenant_id, provider_id)
-            if not provider_entity:
-                raise DatasourceProviderNotFoundError(f"plugin provider {provider_id} not found")
-
-            match datasource_type:
-                case DatasourceProviderType.ONLINE_DOCUMENT:
-                    controller = OnlineDocumentDatasourcePluginProviderController(
-                        entity=provider_entity.declaration,
-                        plugin_id=provider_entity.plugin_id,
-                        plugin_unique_identifier=provider_entity.plugin_unique_identifier,
-                        tenant_id=tenant_id,
-                    )
-                case DatasourceProviderType.ONLINE_DRIVE:
-                    controller = OnlineDriveDatasourcePluginProviderController(
-                        entity=provider_entity.declaration,
-                        plugin_id=provider_entity.plugin_id,
-                        plugin_unique_identifier=provider_entity.plugin_unique_identifier,
-                        tenant_id=tenant_id,
-                    )
-                case DatasourceProviderType.WEBSITE_CRAWL:
-                    controller = WebsiteCrawlDatasourcePluginProviderController(
-                        entity=provider_entity.declaration,
-                        plugin_id=provider_entity.plugin_id,
-                        plugin_unique_identifier=provider_entity.plugin_unique_identifier,
-                        tenant_id=tenant_id,
-                    )
-                case DatasourceProviderType.LOCAL_FILE:
-                    controller = LocalFileDatasourcePluginProviderController(
-                        entity=provider_entity.declaration,
-                        plugin_id=provider_entity.plugin_id,
-                        plugin_unique_identifier=provider_entity.plugin_unique_identifier,
-                        tenant_id=tenant_id,
-                    )
-                case _:
-                    raise ValueError(f"Unsupported datasource type: {datasource_type}")
-
-            datasource_plugin_providers[provider_id] = controller
-
-        return controller
-
-    @classmethod
-    def get_datasource_runtime(
-        cls,
-        provider_id: str,
-        datasource_name: str,
-        tenant_id: str,
-        datasource_type: DatasourceProviderType,
-    ) -> DatasourcePlugin:
-        """
-        get the datasource runtime
-
-        :param provider_type: the type of the provider
-        :param provider_id: the id of the provider
-        :param datasource_name: the name of the datasource
-        :param tenant_id: the tenant id
-
-        :return: the datasource plugin
-        """
-        return cls.get_datasource_plugin_provider(
-            provider_id,
-            tenant_id,
-            datasource_type,
-        ).get_datasource(datasource_name)

api/core/datasource/entities/api_entities.py

@@ -1,71 +0,0 @@
-from typing import Literal, Optional
-
-from pydantic import BaseModel, Field, field_validator
-
-from core.datasource.entities.datasource_entities import DatasourceParameter
-from core.model_runtime.utils.encoders import jsonable_encoder
-from core.tools.entities.common_entities import I18nObject
-
-
-class DatasourceApiEntity(BaseModel):
-    author: str
-    name: str  # identifier
-    label: I18nObject  # label
-    description: I18nObject
-    parameters: Optional[list[DatasourceParameter]] = None
-    labels: list[str] = Field(default_factory=list)
-    output_schema: Optional[dict] = None
-
-
-ToolProviderTypeApiLiteral = Optional[Literal["builtin", "api", "workflow"]]
-
-
-class DatasourceProviderApiEntity(BaseModel):
-    id: str
-    author: str
-    name: str  # identifier
-    description: I18nObject
-    icon: str | dict
-    label: I18nObject  # label
-    type: str
-    masked_credentials: Optional[dict] = None
-    original_credentials: Optional[dict] = None
-    is_team_authorization: bool = False
-    allow_delete: bool = True
-    plugin_id: Optional[str] = Field(default="", description="The plugin id of the datasource")
-    plugin_unique_identifier: Optional[str] = Field(default="", description="The unique identifier of the datasource")
-    datasources: list[DatasourceApiEntity] = Field(default_factory=list)
-    labels: list[str] = Field(default_factory=list)
-
-    @field_validator("datasources", mode="before")
-    @classmethod
-    def convert_none_to_empty_list(cls, v):
-        return v if v is not None else []
-
-    def to_dict(self) -> dict:
-        # -------------
-        # overwrite datasource parameter types for temp fix
-        datasources = jsonable_encoder(self.datasources)
-        for datasource in datasources:
-            if datasource.get("parameters"):
-                for parameter in datasource.get("parameters"):
-                    if parameter.get("type") == DatasourceParameter.DatasourceParameterType.SYSTEM_FILES.value:
-                        parameter["type"] = "files"
-        # -------------
-
-        return {
-            "id": self.id,
-            "author": self.author,
-            "name": self.name,
-            "plugin_id": self.plugin_id,
-            "plugin_unique_identifier": self.plugin_unique_identifier,
-            "description": self.description.to_dict(),
-            "icon": self.icon,
-            "label": self.label.to_dict(),
-            "type": self.type.value,
-            "team_credentials": self.masked_credentials,
-            "is_team_authorization": self.is_team_authorization,
-            "allow_delete": self.allow_delete,
-            "datasources": datasources,
-            "labels": self.labels,
-        }

api/core/datasource/entities/common_entities.py

@@ -1,23 +0,0 @@
-from typing import Optional
-
-from pydantic import BaseModel, Field
-
-
-class I18nObject(BaseModel):
-    """
-    Model class for i18n object.
-    """
-
-    en_US: str
-    zh_Hans: Optional[str] = Field(default=None)
-    pt_BR: Optional[str] = Field(default=None)
-    ja_JP: Optional[str] = Field(default=None)
-
-    def __init__(self, **data):
-        super().__init__(**data)
-        self.zh_Hans = self.zh_Hans or self.en_US
-        self.pt_BR = self.pt_BR or self.en_US
-        self.ja_JP = self.ja_JP or self.en_US
-
-    def to_dict(self) -> dict:
-        return {"zh_Hans": self.zh_Hans, "en_US": self.en_US, "pt_BR": self.pt_BR, "ja_JP": self.ja_JP}

api/core/datasource/entities/datasource_entities.py

@@ -1,363 +0,0 @@
-import enum
-from enum import Enum
-from typing import Any, Optional
-
-from pydantic import BaseModel, Field, ValidationInfo, field_validator
-
-from core.entities.provider_entities import ProviderConfig
-from core.plugin.entities.oauth import OAuthSchema
-from core.plugin.entities.parameters import (
-    PluginParameter,
-    PluginParameterOption,
-    PluginParameterType,
-    as_normal_type,
-    cast_parameter_value,
-    init_frontend_parameter,
-)
-from core.tools.entities.common_entities import I18nObject
-from core.tools.entities.tool_entities import ToolInvokeMessage, ToolLabelEnum
-
-
-class DatasourceProviderType(enum.StrEnum):
-    """
-    Enum class for datasource provider
-    """
-
-    ONLINE_DOCUMENT = "online_document"
-    LOCAL_FILE = "local_file"
-    WEBSITE_CRAWL = "website_crawl"
-    ONLINE_DRIVE = "online_drive"
-
-    @classmethod
-    def value_of(cls, value: str) -> "DatasourceProviderType":
-        """
-        Get value of given mode.
-
-        :param value: mode value
-        :return: mode
-        """
-        for mode in cls:
-            if mode.value == value:
-                return mode
-        raise ValueError(f"invalid mode value {value}")
-
-
-class DatasourceParameter(PluginParameter):
-    """
-    Overrides type
-    """
-
-    class DatasourceParameterType(enum.StrEnum):
-        """
-        removes TOOLS_SELECTOR from PluginParameterType
-        """
-
-        STRING = PluginParameterType.STRING.value
-        NUMBER = PluginParameterType.NUMBER.value
-        BOOLEAN = PluginParameterType.BOOLEAN.value
-        SELECT = PluginParameterType.SELECT.value
-        SECRET_INPUT = PluginParameterType.SECRET_INPUT.value
-        FILE = PluginParameterType.FILE.value
-        FILES = PluginParameterType.FILES.value
-
-        # deprecated, should not use.
-        SYSTEM_FILES = PluginParameterType.SYSTEM_FILES.value
-
-        def as_normal_type(self):
-            return as_normal_type(self)
-
-        def cast_value(self, value: Any):
-            return cast_parameter_value(self, value)
-
-    type: DatasourceParameterType = Field(..., description="The type of the parameter")
-    description: I18nObject = Field(..., description="The description of the parameter")
-
-    @classmethod
-    def get_simple_instance(
-        cls,
-        name: str,
-        typ: DatasourceParameterType,
-        required: bool,
-        options: Optional[list[str]] = None,
-    ) -> "DatasourceParameter":
-        """
-        get a simple datasource parameter
-
-        :param name: the name of the parameter
-        :param llm_description: the description presented to the LLM
-        :param typ: the type of the parameter
-        :param required: if the parameter is required
-        :param options: the options of the parameter
-        """
-        # convert options to ToolParameterOption
-        # FIXME fix the type error
-        if options:
-            option_objs = [
-                PluginParameterOption(value=option, label=I18nObject(en_US=option, zh_Hans=option))
-                for option in options
-            ]
-        else:
-            option_objs = []
-
-        return cls(
-            name=name,
-            label=I18nObject(en_US="", zh_Hans=""),
-            placeholder=None,
-            type=typ,
-            required=required,
-            options=option_objs,
-            description=I18nObject(en_US="", zh_Hans=""),
-        )
-
-    def init_frontend_parameter(self, value: Any):
-        return init_frontend_parameter(self, self.type, value)
-
-
-class DatasourceIdentity(BaseModel):
-    author: str = Field(..., description="The author of the datasource")
-    name: str = Field(..., description="The name of the datasource")
-    label: I18nObject = Field(..., description="The label of the datasource")
-    provider: str = Field(..., description="The provider of the datasource")
-    icon: Optional[str] = None
-
-
-class DatasourceEntity(BaseModel):
-    identity: DatasourceIdentity
-    parameters: list[DatasourceParameter] = Field(default_factory=list)
-    description: I18nObject = Field(..., description="The label of the datasource")
-    output_schema: Optional[dict] = None
-
-    @field_validator("parameters", mode="before")
-    @classmethod
-    def set_parameters(cls, v, validation_info: ValidationInfo) -> list[DatasourceParameter]:
-        return v or []
-
-
-class DatasourceProviderIdentity(BaseModel):
-    author: str = Field(..., description="The author of the tool")
-    name: str = Field(..., description="The name of the tool")
-    description: I18nObject = Field(..., description="The description of the tool")
-    icon: str = Field(..., description="The icon of the tool")
-    label: I18nObject = Field(..., description="The label of the tool")
-    tags: Optional[list[ToolLabelEnum]] = Field(
-        default=[],
-        description="The tags of the tool",
-    )
-
-
-class DatasourceProviderEntity(BaseModel):
-    """
-    Datasource provider entity
-    """
-
-    identity: DatasourceProviderIdentity
-    credentials_schema: list[ProviderConfig] = Field(default_factory=list)
-    oauth_schema: Optional[OAuthSchema] = None
-    provider_type: DatasourceProviderType
-
-
-class DatasourceProviderEntityWithPlugin(DatasourceProviderEntity):
-    datasources: list[DatasourceEntity] = Field(default_factory=list)
-
-
-class DatasourceInvokeMeta(BaseModel):
-    """
-    Datasource invoke meta
-    """
-
-    time_cost: float = Field(..., description="The time cost of the tool invoke")
-    error: Optional[str] = None
-    tool_config: Optional[dict] = None
-
-    @classmethod
-    def empty(cls) -> "DatasourceInvokeMeta":
-        """
-        Get an empty instance of DatasourceInvokeMeta
-        """
-        return cls(time_cost=0.0, error=None, tool_config={})
-
-    @classmethod
-    def error_instance(cls, error: str) -> "DatasourceInvokeMeta":
-        """
-        Get an instance of DatasourceInvokeMeta with error
-        """
-        return cls(time_cost=0.0, error=error, tool_config={})
-
-    def to_dict(self) -> dict:
-        return {
-            "time_cost": self.time_cost,
-            "error": self.error,
-            "tool_config": self.tool_config,
-        }
-
-
-class DatasourceLabel(BaseModel):
-    """
-    Datasource label
-    """
-
-    name: str = Field(..., description="The name of the tool")
-    label: I18nObject = Field(..., description="The label of the tool")
-    icon: str = Field(..., description="The icon of the tool")
-
-
-class DatasourceInvokeFrom(Enum):
-    """
-    Enum class for datasource invoke
-    """
-
-    RAG_PIPELINE = "rag_pipeline"
-
-
-class OnlineDocumentPage(BaseModel):
-    """
-    Online document page
-    """
-
-    page_id: str = Field(..., description="The page id")
-    page_name: str = Field(..., description="The page title")
-    page_icon: Optional[dict] = Field(None, description="The page icon")
-    type: str = Field(..., description="The type of the page")
-    last_edited_time: str = Field(..., description="The last edited time")
-    parent_id: Optional[str] = Field(None, description="The parent page id")
-
-
-class OnlineDocumentInfo(BaseModel):
-    """
-    Online document info
-    """
-
-    workspace_id: Optional[str] = Field(None, description="The workspace id")
-    workspace_name: Optional[str] = Field(None, description="The workspace name")
-    workspace_icon: Optional[str] = Field(None, description="The workspace icon")
-    total: int = Field(..., description="The total number of documents")
-    pages: list[OnlineDocumentPage] = Field(..., description="The pages of the online document")
-
-
-class OnlineDocumentPagesMessage(BaseModel):
-    """
-    Get online document pages response
-    """
-
-    result: list[OnlineDocumentInfo]
-
-
-class GetOnlineDocumentPageContentRequest(BaseModel):
-    """
-    Get online document page content request
-    """
-
-    workspace_id: str = Field(..., description="The workspace id")
-    page_id: str = Field(..., description="The page id")
-    type: str = Field(..., description="The type of the page")
-
-
-class OnlineDocumentPageContent(BaseModel):
-    """
-    Online document page content
-    """
-
-    workspace_id: str = Field(..., description="The workspace id")
-    page_id: str = Field(..., description="The page id")
-    content: str = Field(..., description="The content of the page")
-
-
-class GetOnlineDocumentPageContentResponse(BaseModel):
-    """
-    Get online document page content response
-    """
-
-    result: OnlineDocumentPageContent
-
-
-class GetWebsiteCrawlRequest(BaseModel):
-    """
-    Get website crawl request
-    """
-
-    crawl_parameters: dict = Field(..., description="The crawl parameters")
-
-
-class WebSiteInfoDetail(BaseModel):
-    source_url: str = Field(..., description="The url of the website")
-    content: str = Field(..., description="The content of the website")
-    title: str = Field(..., description="The title of the website")
-    description: str = Field(..., description="The description of the website")
-
-
-class WebSiteInfo(BaseModel):
-    """
-    Website info
-    """
-
-    status: Optional[str] = Field(..., description="crawl job status")
-    web_info_list: Optional[list[WebSiteInfoDetail]] = []
-    total: Optional[int] = Field(default=0, description="The total number of websites")
-    completed: Optional[int] = Field(default=0, description="The number of completed websites")
-
-
-class WebsiteCrawlMessage(BaseModel):
-    """
-    Get website crawl response
-    """
-
-    result: WebSiteInfo = WebSiteInfo(status="", web_info_list=[], total=0, completed=0)
-
-
-class DatasourceMessage(ToolInvokeMessage):
-    pass
-
-
-#########################
-# Online drive file
-#########################
-
-
-class OnlineDriveFile(BaseModel):
-    """
-    Online drive file
-    """
-
-    id: str = Field(..., description="The file ID")
-    name: str = Field(..., description="The file name")
-    size: int = Field(..., description="The file size")
-    type: str = Field(..., description="The file type: folder or file")
-
-
-class OnlineDriveFileBucket(BaseModel):
-    """
-    Online drive file bucket
-    """
-
-    bucket: Optional[str] = Field(None, description="The file bucket")
-    files: list[OnlineDriveFile] = Field(..., description="The file list")
-    is_truncated: bool = Field(False, description="Whether the result is truncated")
-    next_page_parameters: Optional[dict] = Field(None, description="Parameters for fetching the next page")
-
-
-class OnlineDriveBrowseFilesRequest(BaseModel):
-    """
-    Get online drive file list request
-    """
-
-    bucket: Optional[str] = Field(None, description="The file bucket")
-    prefix: str = Field(..., description="The parent folder ID")
-    max_keys: int = Field(20, description="Page size for pagination")
-    next_page_parameters: Optional[dict] = Field(None, description="Parameters for fetching the next page")
-
-
-class OnlineDriveBrowseFilesResponse(BaseModel):
-    """
-    Get online drive file list response
-    """
-
-    result: list[OnlineDriveFileBucket] = Field(..., description="The list of file buckets")
-
-
-class OnlineDriveDownloadFileRequest(BaseModel):
-    """
-    Get online drive file
-    """
-
-    id: str = Field(..., description="The id of the file")
-    bucket: Optional[str] = Field(None, description="The name of the bucket")

api/core/datasource/errors.py

@@ -1,37 +0,0 @@
-from core.datasource.entities.datasource_entities import DatasourceInvokeMeta
-
-
-class DatasourceProviderNotFoundError(ValueError):
-    pass
-
-
-class DatasourceNotFoundError(ValueError):
-    pass
-
-
-class DatasourceParameterValidationError(ValueError):
-    pass
-
-
-class DatasourceProviderCredentialValidationError(ValueError):
-    pass
-
-
-class DatasourceNotSupportedError(ValueError):
-    pass
-
-
-class DatasourceInvokeError(ValueError):
-    pass
-
-
-class DatasourceApiSchemaError(ValueError):
-    pass
-
-
-class DatasourceEngineInvokeError(Exception):
-    meta: DatasourceInvokeMeta
-
-    def __init__(self, meta, **kwargs):
-        self.meta = meta
-        super().__init__(**kwargs)

api/core/datasource/local_file/local_file_plugin.py

@@ -1,28 +0,0 @@
-from core.datasource.__base.datasource_plugin import DatasourcePlugin
-from core.datasource.__base.datasource_runtime import DatasourceRuntime
-from core.datasource.entities.datasource_entities import (
-    DatasourceEntity,
-    DatasourceProviderType,
-)
-
-
-class LocalFileDatasourcePlugin(DatasourcePlugin):
-    tenant_id: str
-    icon: str
-    plugin_unique_identifier: str
-
-    def __init__(
-        self,
-        entity: DatasourceEntity,
-        runtime: DatasourceRuntime,
-        tenant_id: str,
-        icon: str,
-        plugin_unique_identifier: str,
-    ) -> None:
-        super().__init__(entity, runtime)
-        self.tenant_id = tenant_id
-        self.icon = icon
-        self.plugin_unique_identifier = plugin_unique_identifier
-
-    def datasource_provider_type(self) -> str:
-        return DatasourceProviderType.LOCAL_FILE

api/core/datasource/local_file/local_file_provider.py

@@ -1,56 +0,0 @@
-from typing import Any
-
-from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
-from core.datasource.__base.datasource_runtime import DatasourceRuntime
-from core.datasource.entities.datasource_entities import DatasourceProviderEntityWithPlugin, DatasourceProviderType
-from core.datasource.local_file.local_file_plugin import LocalFileDatasourcePlugin
-
-
-class LocalFileDatasourcePluginProviderController(DatasourcePluginProviderController):
-    entity: DatasourceProviderEntityWithPlugin
-    plugin_id: str
-    plugin_unique_identifier: str
-
-    def __init__(
-        self, entity: DatasourceProviderEntityWithPlugin, plugin_id: str, plugin_unique_identifier: str, tenant_id: str
-    ) -> None:
-        super().__init__(entity, tenant_id)
-        self.plugin_id = plugin_id
-        self.plugin_unique_identifier = plugin_unique_identifier
-
-    @property
-    def provider_type(self) -> DatasourceProviderType:
-        """
-        returns the type of the provider
-        """
-        return DatasourceProviderType.LOCAL_FILE
-
-    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]) -> None:
-        """
-        validate the credentials of the provider
-        """
-        pass
-
-    def get_datasource(self, datasource_name: str) -> LocalFileDatasourcePlugin:  # type: ignore
-        """
-        return datasource with given name
-        """
-        datasource_entity = next(
-            (
-                datasource_entity
-                for datasource_entity in self.entity.datasources
-                if datasource_entity.identity.name == datasource_name
-            ),
-            None,
-        )
-
-        if not datasource_entity:
-            raise ValueError(f"Datasource with name {datasource_name} not found")
-
-        return LocalFileDatasourcePlugin(
-            entity=datasource_entity,
-            runtime=DatasourceRuntime(tenant_id=self.tenant_id),
-            tenant_id=self.tenant_id,
-            icon=self.entity.identity.icon,
-            plugin_unique_identifier=self.plugin_unique_identifier,
-        )