fix: template

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

.github/workflows/api-tests.yml

@@ -62,7 +62,7 @@ jobs:
           compose-file: |
             docker/docker-compose.middleware.yaml
           services: |
-            db
+            db_postgres
             redis
             sandbox
             ssrf_proxy

.github/workflows/db-migration-test.yml

@@ -8,7 +8,7 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  db-migration-test:
+  db-migration-test-postgres:
     runs-on: ubuntu-latest
 
     steps:
@@ -45,7 +45,7 @@ jobs:
           compose-file: |
             docker/docker-compose.middleware.yaml
           services: |
-            db
+            db_postgres
             redis
 
       - name: Prepare configs
@@ -57,3 +57,54 @@ jobs:
         env:
           DEBUG: true
         run: uv run --directory api flask upgrade-db
+
+  db-migration-test-mysql:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Setup UV and Python
+        uses: astral-sh/setup-uv@v6
+        with:
+          enable-cache: true
+          python-version: "3.12"
+          cache-dependency-glob: api/uv.lock
+
+      - name: Install dependencies
+        run: uv sync --project api
+      - name: Ensure Offline migration are supported
+        run: |
+          # upgrade
+          uv run --directory api flask db upgrade 'base:head' --sql
+          # downgrade
+          uv run --directory api flask db downgrade 'head:base' --sql
+
+      - name: Prepare middleware env
+        run: |
+          cd docker
+          cp middleware.env.example middleware.env
+
+      - name: Set up Middlewares
+        uses: hoverkraft-tech/compose-action@v2.0.2
+        with:
+          compose-file: |
+            docker/docker-compose.middleware.yaml
+          services: |
+            db_mysql
+            redis
+
+      - name: Prepare configs for MySQL
+        run: |
+          cd api
+          cp .env.example .env
+          sed -i 's/DB_TYPE=postgresql/DB_TYPE=mysql/' .env
+
+      - name: Run DB Migration
+        env:
+          DEBUG: true
+        run: uv run --directory api flask upgrade-db

.gitignore

@@ -183,6 +183,7 @@ docker/volumes/couchbase/*
 docker/volumes/oceanbase/*
 docker/volumes/plugin_daemon/*
 docker/volumes/matrixone/*
+docker/volumes/mysql/data/*
 !docker/volumes/oceanbase/init.d
 
 docker/nginx/conf.d/default.conf

README.md

@@ -117,7 +117,7 @@ All of Dify's offerings come with corresponding APIs, so you could effortlessly
   Use our [documentation](https://docs.dify.ai) for further references and more in-depth instructions.
 
 - **Dify for enterprise / organizations<br/>**
-  We provide additional enterprise-centric features. [Log your questions for us through this chatbot](https://udify.app/chat/22L1zSxg6yW1cWQg) or [send us an email](mailto:business@dify.ai?subject=%5BGitHub%5DBusiness%20License%20Inquiry) to discuss enterprise needs. <br/>
+  We provide additional enterprise-centric features. [Send us an email](mailto:business@dify.ai?subject=%5BGitHub%5DBusiness%20License%20Inquiry) to discuss your enterprise needs. <br/>
 
   > For startups and small businesses using AWS, check out [Dify Premium on AWS Marketplace](https://aws.amazon.com/marketplace/pp/prodview-t22mebxzwjhu6) and deploy it to your own AWS VPC with one click. It's an affordable AMI offering with the option to create apps with custom logo and branding.
 

api/.env.example

@@ -69,12 +69,31 @@ REDIS_CLUSTERS_PASSWORD=
 # celery configuration
 CELERY_BROKER_URL=redis://:difyai123456@localhost:${REDIS_PORT}/1
 CELERY_BACKEND=redis
-# PostgreSQL database configuration
-DB_USERNAME=postgres
-DB_PASSWORD=difyai123456
-DB_HOST=localhost
-DB_PORT=5432
-DB_DATABASE=dify
+
+# Database configuration
+DB_TYPE=postgresql
+
+# PostgreSQL configuration
+POSTGRES_USER=postgres
+POSTGRES_PASSWORD=difyai123456
+POSTGRES_HOST=localhost
+POSTGRES_PORT=5432
+POSTGRES_DATABASE=dify
+
+# MySQL configuration
+MYSQL_USER=mysql
+MYSQL_PASSWORD=difyai123456
+MYSQL_HOST=localhost
+MYSQL_PORT=3306
+MYSQL_DATABASE=dify
+
+# OceanBase configuration
+OCEANBASE_USER=root@test
+OCEANBASE_PASSWORD=difyai123456
+OCEANBASE_HOST=localhost
+OCEANBASE_PORT=2881
+OCEANBASE_DATABASE=test
+
 SQLALCHEMY_POOL_PRE_PING=true
 SQLALCHEMY_POOL_TIMEOUT=30
 
@@ -172,6 +191,15 @@ WEAVIATE_API_KEY=WVF5YThaHlkYwhGUSmCRgsX3tD5ngdN8pkih
 WEAVIATE_GRPC_ENABLED=false
 WEAVIATE_BATCH_SIZE=100
 
+# OceanBase Vector configuration
+OCEANBASE_HOST=127.0.0.1
+OCEANBASE_PORT=2881
+OCEANBASE_USER=root@test
+OCEANBASE_PASSWORD=difyai123456
+OCEANBASE_DATABASE=test
+OCEANBASE_MEMORY_LIMIT=6G
+OCEANBASE_ENABLE_HYBRID_SEARCH=false
+
 # Qdrant configuration, use `http://localhost:6333` for local mode or `https://your-qdrant-cluster-url.qdrant.io` for remote mode
 QDRANT_URL=http://localhost:6333
 QDRANT_API_KEY=difyai123456
@@ -337,15 +365,6 @@ LINDORM_PASSWORD=admin
 LINDORM_USING_UGC=True
 LINDORM_QUERY_TIMEOUT=1
 
-# OceanBase Vector configuration
-OCEANBASE_VECTOR_HOST=127.0.0.1
-OCEANBASE_VECTOR_PORT=2881
-OCEANBASE_VECTOR_USER=root@test
-OCEANBASE_VECTOR_PASSWORD=difyai123456
-OCEANBASE_VECTOR_DATABASE=test
-OCEANBASE_MEMORY_LIMIT=6G
-OCEANBASE_ENABLE_HYBRID_SEARCH=false
-
 # AlibabaCloud MySQL Vector configuration
 ALIBABACLOUD_MYSQL_HOST=127.0.0.1
 ALIBABACLOUD_MYSQL_PORT=3306
@@ -371,6 +390,12 @@ UPLOAD_IMAGE_FILE_SIZE_LIMIT=10
 UPLOAD_VIDEO_FILE_SIZE_LIMIT=100
 UPLOAD_AUDIO_FILE_SIZE_LIMIT=50
 
+# Comma-separated list of file extensions blocked from upload for security reasons.
+# Extensions should be lowercase without dots (e.g., exe,bat,sh,dll).
+# Empty by default to allow all file types.
+# Recommended: exe,bat,cmd,com,scr,vbs,ps1,msi,dll
+UPLOAD_FILE_EXTENSION_BLACKLIST=
+
 # Model configuration
 MULTIMODAL_SEND_FORMAT=base64
 PROMPT_GENERATION_MAX_TOKENS=512
@@ -608,3 +633,9 @@ SWAGGER_UI_PATH=/swagger-ui.html
 # Whether to encrypt dataset IDs when exporting DSL files (default: true)
 # Set to false to export dataset IDs as plain text for easier cross-environment import
 DSL_EXPORT_ENCRYPT_DATASET_ID=true
+
+# Tenant isolated task queue configuration
+TENANT_ISOLATED_TASK_CONCURRENCY=1
+
+# Maximum number of segments for dataset segments API (0 for unlimited)
+DATASET_MAX_SEGMENTS_PER_REQUEST=0

api/Dockerfile

@@ -15,7 +15,11 @@ FROM base AS packages
 # RUN sed -i 's@deb.debian.org@mirrors.aliyun.com@g' /etc/apt/sources.list.d/debian.sources
 
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends gcc g++ libc-dev libffi-dev libgmp-dev libmpfr-dev libmpc-dev
+    && apt-get install -y --no-install-recommends \
+          # basic environment
+          g++ \
+          # for building gmpy2
+          libmpfr-dev libmpc-dev
 
 # Install Python dependencies
 COPY pyproject.toml uv.lock ./
@@ -49,7 +53,9 @@ RUN \
     # Install dependencies
     && apt-get install -y --no-install-recommends \
         # basic environment
-        curl nodejs libgmp-dev libmpfr-dev libmpc-dev \
+        curl nodejs \
+        # for gmpy2 \
+        libgmp-dev libmpfr-dev libmpc-dev \
         # For Security
         expat libldap-2.5-0 perl libsqlite3-0 zlib1g \
         # install fonts to support the use of tools like pypdfium2

api/README.md

@@ -15,8 +15,8 @@
    ```bash
    cd ../docker
    cp middleware.env.example middleware.env
-   # change the profile to other vector database if you are not using weaviate
-   docker compose -f docker-compose.middleware.yaml --profile weaviate -p dify up -d
+   # change the profile to mysql if you are not using postgres,change the profile to other vector database if you are not using weaviate
+   docker compose -f docker-compose.middleware.yaml --profile postgresql --profile weaviate -p dify up -d
    cd ../api
    ```
 

api/commands.py

@@ -1601,7 +1601,7 @@ def transform_datasource_credentials():
                             "integration_secret": api_key,
                         }
                         datasource_provider = DatasourceProvider(
-                            provider="jina",
+                            provider="jinareader",
                             tenant_id=tenant_id,
                             plugin_id=jina_plugin_id,
                             auth_type=api_key_credential_type.value,

api/configs/feature/__init__.py

@@ -331,6 +331,31 @@ class FileUploadConfig(BaseSettings):
         default=10,
     )
 
+    inner_UPLOAD_FILE_EXTENSION_BLACKLIST: str = Field(
+        description=(
+            "Comma-separated list of file extensions that are blocked from upload. "
+            "Extensions should be lowercase without dots (e.g., 'exe,bat,sh,dll'). "
+            "Empty by default to allow all file types."
+        ),
+        validation_alias=AliasChoices("UPLOAD_FILE_EXTENSION_BLACKLIST"),
+        default="",
+    )
+
+    @computed_field  # type: ignore[misc]
+    @property
+    def UPLOAD_FILE_EXTENSION_BLACKLIST(self) -> set[str]:
+        """
+        Parse and return the blacklist as a set of lowercase extensions.
+        Returns an empty set if no blacklist is configured.
+        """
+        if not self.inner_UPLOAD_FILE_EXTENSION_BLACKLIST:
+            return set()
+        return {
+            ext.strip().lower().strip(".")
+            for ext in self.inner_UPLOAD_FILE_EXTENSION_BLACKLIST.split(",")
+            if ext.strip()
+        }
+
 
 class HttpConfig(BaseSettings):
     """
@@ -920,6 +945,11 @@ class DataSetConfig(BaseSettings):
         default=True,
     )
 
+    DATASET_MAX_SEGMENTS_PER_REQUEST: NonNegativeInt = Field(
+        description="Maximum number of segments for dataset segments API (0 for unlimited)",
+        default=0,
+    )
+
 
 class WorkspaceConfig(BaseSettings):
     """
@@ -1112,6 +1142,13 @@ class SwaggerUIConfig(BaseSettings):
     )
 
 
+class TenantIsolatedTaskQueueConfig(BaseSettings):
+    TENANT_ISOLATED_TASK_CONCURRENCY: int = Field(
+        description="Number of tasks allowed to be delivered concurrently from isolated queue per tenant",
+        default=1,
+    )
+
+
 class FeatureConfig(
     # place the configs in alphabet order
     AppExecutionConfig,
@@ -1136,6 +1173,7 @@ class FeatureConfig(
     RagEtlConfig,
     RepositoryConfig,
     SecurityConfig,
+    TenantIsolatedTaskQueueConfig,
     ToolConfig,
     UpdateConfig,
     WorkflowConfig,

api/configs/middleware/__init__.py

@@ -105,31 +105,141 @@ class KeywordStoreConfig(BaseSettings):
 
 
 class DatabaseConfig(BaseSettings):
-    DB_HOST: str = Field(
-        description="Hostname or IP address of the database server.",
+    # Database type selector
+    DB_TYPE: Literal["postgresql", "mysql", "oceanbase"] = Field(
+        description="Database type to use. OceanBase is MySQL-compatible.",
+        default="postgresql",
+    )
+    
+    # PostgreSQL configuration
+    POSTGRES_HOST: str = Field(
+        description="PostgreSQL hostname or IP address.",
         default="localhost",
     )
 
-    DB_PORT: PositiveInt = Field(
-        description="Port number for database connection.",
+    POSTGRES_PORT: PositiveInt = Field(
+        description="PostgreSQL port number.",
         default=5432,
     )
 
-    DB_USERNAME: str = Field(
-        description="Username for database authentication.",
+    POSTGRES_USER: str = Field(
+        description="PostgreSQL username.",
         default="postgres",
     )
 
-    DB_PASSWORD: str = Field(
-        description="Password for database authentication.",
-        default="",
+    POSTGRES_PASSWORD: str = Field(
+        description="PostgreSQL password.",
+        default="difyai123456",
     )
 
-    DB_DATABASE: str = Field(
-        description="Name of the database to connect to.",
+    POSTGRES_DATABASE: str = Field(
+        description="PostgreSQL database name.",
         default="dify",
     )
 
+    # MySQL configuration
+    MYSQL_HOST: str = Field(
+        description="MySQL hostname or IP address.",
+        default="localhost",
+    )
+    
+    MYSQL_PORT: PositiveInt = Field(
+        description="MySQL port number.",
+        default=3306,
+    )
+
+    MYSQL_USER: str = Field(
+        description="MySQL username.",
+        default="root",
+    )
+
+    MYSQL_PASSWORD: str = Field(
+        description="MySQL password.",
+        default="difyai123456",
+    )
+
+    MYSQL_DATABASE: str = Field(
+        description="MySQL database name.",
+        default="dify",
+    )
+
+    # OceanBase configuration(MySQL-compatible)
+    OCEANBASE_HOST: str = Field(
+        description="OceanBase hostname or IP address.",
+        default="localhost",
+    )
+    
+    OCEANBASE_PORT: PositiveInt = Field(
+        description="OceanBase port number.",
+        default=2881,
+    )
+    
+    OCEANBASE_USER: str = Field(
+        description="OceanBase username.",
+        default="root@test",
+    )
+    
+    OCEANBASE_PASSWORD: str = Field(
+        description="OceanBase password.",
+        default="difyai123456",
+    )
+    
+    OCEANBASE_DATABASE: str = Field(
+        description="OceanBase database name.",
+        default="test",
+    )
+
+    # Dynamic properties based on DB_TYPE
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_HOST(self) -> str:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_HOST
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_HOST
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_HOST
+
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_PORT(self) -> int:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_PORT
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_PORT
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_PORT
+    
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_USERNAME(self) -> str:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_USER
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_USER
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_USER
+    
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_PASSWORD(self) -> str:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_PASSWORD
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_PASSWORD
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_PASSWORD
+
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_DATABASE(self) -> str:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_DATABASE
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_DATABASE
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_DATABASE
+
     DB_CHARSET: str = Field(
         description="Character set for database connection.",
         default="",
@@ -140,10 +250,10 @@ class DatabaseConfig(BaseSettings):
         default="",
     )
 
-    SQLALCHEMY_DATABASE_URI_SCHEME: str = Field(
-        description="Database URI scheme for SQLAlchemy connection.",
-        default="postgresql",
-    )
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def SQLALCHEMY_DATABASE_URI_SCHEME(self) -> str:
+        return "postgresql" if self.DB_TYPE == "postgresql" else "mysql+pymysql"
 
     @computed_field  # type: ignore[prop-decorator]
     @property
@@ -204,15 +314,15 @@ class DatabaseConfig(BaseSettings):
         # Parse DB_EXTRAS for 'options'
         db_extras_dict = dict(parse_qsl(self.DB_EXTRAS))
         options = db_extras_dict.get("options", "")
-        # Always include timezone
-        timezone_opt = "-c timezone=UTC"
-        if options:
-            # Merge user options and timezone
-            merged_options = f"{options} {timezone_opt}"
-        else:
-            merged_options = timezone_opt
-
-        connect_args = {"options": merged_options}
+        connect_args = {}
+        # Use the dynamic SQLALCHEMY_DATABASE_URI_SCHEME property
+        if self.SQLALCHEMY_DATABASE_URI_SCHEME.startswith("postgresql"):
+            timezone_opt = "-c timezone=UTC"
+            if options:
+                merged_options = f"{options} {timezone_opt}"
+            else:
+                merged_options = timezone_opt
+            connect_args = {"options": merged_options}
 
         return {
             "pool_size": self.SQLALCHEMY_POOL_SIZE,

api/configs/middleware/vdb/oceanbase_config.py

@@ -7,29 +7,29 @@ class OceanBaseVectorConfig(BaseSettings):
     Configuration settings for OceanBase Vector database
     """
 
-    OCEANBASE_VECTOR_HOST: str | None = Field(
-        description="Hostname or IP address of the OceanBase Vector server (e.g. 'localhost')",
-        default=None,
+    OCEANBASE_HOST: str = Field(
+        description="OceanBase hostname or IP address.",
+        default="localhost",
     )
-
-    OCEANBASE_VECTOR_PORT: PositiveInt | None = Field(
-        description="Port number on which the OceanBase Vector server is listening (default is 2881)",
+    
+    OCEANBASE_PORT: PositiveInt = Field(
+        description="OceanBase port number.",
         default=2881,
     )
-
-    OCEANBASE_VECTOR_USER: str | None = Field(
-        description="Username for authenticating with the OceanBase Vector database",
-        default=None,
+    
+    OCEANBASE_USER: str = Field(
+        description="OceanBase username.",
+        default="root@test",
     )
-
-    OCEANBASE_VECTOR_PASSWORD: str | None = Field(
-        description="Password for authenticating with the OceanBase Vector database",
-        default=None,
+    
+    OCEANBASE_PASSWORD: str = Field(
+        description="OceanBase password.",
+        default="difyai123456",
     )
-
-    OCEANBASE_VECTOR_DATABASE: str | None = Field(
-        description="Name of the OceanBase Vector database to connect to",
-        default=None,
+    
+    OCEANBASE_DATABASE: str = Field(
+        description="OceanBase database name.",
+        default="test",
     )
 
     OCEANBASE_ENABLE_HYBRID_SEARCH: bool = Field(

api/configs/middleware/vdb/weaviate_config.py

@@ -22,6 +22,11 @@ class WeaviateConfig(BaseSettings):
         default=True,
     )
 
+    WEAVIATE_GRPC_ENDPOINT: str | None = Field(
+        description="URL of the Weaviate gRPC server (e.g., 'grpc://localhost:50051' or 'grpcs://weaviate.example.com:443')",
+        default=None,
+    )
+
     WEAVIATE_BATCH_SIZE: PositiveInt = Field(
         description="Number of objects to be processed in a single batch operation (default is 100)",
         default=100,

api/controllers/common/errors.py

@@ -25,6 +25,12 @@ class UnsupportedFileTypeError(BaseHTTPException):
     code = 415
 
 
+class BlockedFileExtensionError(BaseHTTPException):
+    error_code = "file_extension_blocked"
+    description = "The file extension is blocked for security reasons."
+    code = 400
+
+
 class TooManyFilesError(BaseHTTPException):
     error_code = "too_many_files"
     description = "Only one file is allowed."

api/controllers/console/app/annotation.py

@@ -16,6 +16,7 @@ from fields.annotation_fields import (
     annotation_fields,
     annotation_hit_history_fields,
 )
+from libs.helper import uuid_value
 from libs.login import login_required
 from services.annotation_service import AppAnnotationService
 
@@ -175,8 +176,10 @@ class AnnotationApi(Resource):
         api.model(
             "CreateAnnotationRequest",
             {
-                "question": fields.String(required=True, description="Question text"),
-                "answer": fields.String(required=True, description="Answer text"),
+                "message_id": fields.String(description="Message ID (optional)"),
+                "question": fields.String(description="Question text (required when message_id not provided)"),
+                "answer": fields.String(description="Answer text (use 'answer' or 'content')"),
+                "content": fields.String(description="Content text (use 'answer' or 'content')"),
                 "annotation_reply": fields.Raw(description="Annotation reply data"),
             },
         )
@@ -193,11 +196,14 @@ class AnnotationApi(Resource):
         app_id = str(app_id)
         parser = (
             reqparse.RequestParser()
-            .add_argument("question", required=True, type=str, location="json")
-            .add_argument("answer", required=True, type=str, location="json")
+            .add_argument("message_id", required=False, type=uuid_value, location="json")
+            .add_argument("question", required=False, type=str, location="json")
+            .add_argument("answer", required=False, type=str, location="json")
+            .add_argument("content", required=False, type=str, location="json")
+            .add_argument("annotation_reply", required=False, type=dict, location="json")
         )
         args = parser.parse_args()
-        annotation = AppAnnotationService.insert_app_annotation_directly(args, app_id)
+        annotation = AppAnnotationService.up_insert_app_annotation_from_message(args, app_id)
         return annotation
 
     @setup_required

api/controllers/console/app/conversation.py

@@ -1,7 +1,5 @@
-from datetime import datetime
-
-import pytz
 import sqlalchemy as sa
+from flask import abort
 from flask_restx import Resource, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from sqlalchemy import func, or_
@@ -19,7 +17,7 @@ from fields.conversation_fields import (
     conversation_pagination_fields,
     conversation_with_summary_pagination_fields,
 )
-from libs.datetime_utils import naive_utc_now
+from libs.datetime_utils import naive_utc_now, parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models import Conversation, EndUser, Message, MessageAnnotation
@@ -90,25 +88,17 @@ class CompletionConversationApi(Resource):
 
         account = current_user
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             query = query.where(Conversation.created_at >= start_datetime_utc)
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=59)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
+            end_datetime_utc = end_datetime_utc.replace(second=59)
             query = query.where(Conversation.created_at < end_datetime_utc)
 
         # FIXME, the type ignore in this file
@@ -270,29 +260,21 @@ class ChatConversationApi(Resource):
 
         account = current_user
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             match args["sort_by"]:
                 case "updated_at" | "-updated_at":
                     query = query.where(Conversation.updated_at >= start_datetime_utc)
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at >= start_datetime_utc)
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=59)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
+            end_datetime_utc = end_datetime_utc.replace(second=59)
             match args["sort_by"]:
                 case "updated_at" | "-updated_at":
                     query = query.where(Conversation.updated_at <= end_datetime_utc)

api/controllers/console/app/message.py

@@ -16,7 +16,6 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.explore.error import AppSuggestedQuestionsAfterAnswerDisabledError
 from controllers.console.wraps import (
     account_initialization_required,
-    cloud_edition_billing_resource_check,
     edit_permission_required,
     setup_required,
 )
@@ -24,12 +23,11 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
-from fields.conversation_fields import annotation_fields, message_detail_fields
+from fields.conversation_fields import message_detail_fields
 from libs.helper import uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required
 from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
-from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
 from services.message_service import MessageService
@@ -194,45 +192,6 @@ class MessageFeedbackApi(Resource):
         return {"result": "success"}
 
 
-@console_ns.route("/apps/<uuid:app_id>/annotations")
-class MessageAnnotationApi(Resource):
-    @api.doc("create_message_annotation")
-    @api.doc(description="Create message annotation")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "MessageAnnotationRequest",
-            {
-                "message_id": fields.String(description="Message ID"),
-                "question": fields.String(required=True, description="Question text"),
-                "answer": fields.String(required=True, description="Answer text"),
-                "annotation_reply": fields.Raw(description="Annotation reply"),
-            },
-        )
-    )
-    @api.response(200, "Annotation created successfully", annotation_fields)
-    @api.response(403, "Insufficient permissions")
-    @marshal_with(annotation_fields)
-    @get_app_model
-    @setup_required
-    @login_required
-    @cloud_edition_billing_resource_check("annotation")
-    @account_initialization_required
-    @edit_permission_required
-    def post(self, app_model):
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("message_id", required=False, type=uuid_value, location="json")
-            .add_argument("question", required=True, type=str, location="json")
-            .add_argument("answer", required=True, type=str, location="json")
-            .add_argument("annotation_reply", required=False, type=dict, location="json")
-        )
-        args = parser.parse_args()
-        annotation = AppAnnotationService.up_insert_app_annotation_from_message(args, app_model.id)
-
-        return annotation
-
-
 @console_ns.route("/apps/<uuid:app_id>/annotations/count")
 class MessageAnnotationCountApi(Resource):
     @api.doc("get_annotation_count")

api/controllers/console/app/statistic.py

@@ -1,9 +1,7 @@
-from datetime import datetime
 from decimal import Decimal
 
-import pytz
 import sqlalchemy as sa
-from flask import jsonify
+from flask import abort, jsonify
 from flask_restx import Resource, fields, reqparse
 
 from controllers.console import api, console_ns
@@ -11,9 +9,10 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
-from libs.helper import DatetimeString
+from libs.datetime_utils import parse_time_range
+from libs.helper import DatetimeString, convert_datetime_to_date
 from libs.login import current_account_with_tenant, login_required
-from models import AppMode, Message
+from models import AppMode
 
 
 @console_ns.route("/apps/<uuid:app_id>/statistics/daily-messages")
@@ -45,8 +44,9 @@ class DailyMessageStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     COUNT(*) AS message_count
 FROM
     messages
@@ -56,26 +56,16 @@ WHERE
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
 
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -119,42 +109,39 @@ class DailyConversationStatistic(Resource):
             .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         )
         args = parser.parse_args()
+
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
+    COUNT(DISTINCT conversation_id) AS conversation_count
+FROM
+    messages
+WHERE
+    app_id = :app_id
+    AND invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        stmt = (
-            sa.select(
-                sa.func.date(
-                    sa.func.date_trunc("day", sa.text("created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz"))
-                ).label("date"),
-                sa.func.count(sa.distinct(Message.conversation_id)).label("conversation_count"),
-            )
-            .select_from(Message)
-            .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER)
-        )
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
-            stmt = stmt.where(Message.created_at >= start_datetime_utc)
+        if start_datetime_utc:
+            sql_query += " AND created_at >= :start"
+            arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-            stmt = stmt.where(Message.created_at < end_datetime_utc)
+        if end_datetime_utc:
+            sql_query += " AND created_at < :end"
+            arg_dict["end"] = end_datetime_utc
 
-        stmt = stmt.group_by("date").order_by("date")
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
         with db.engine.begin() as conn:
-            rs = conn.execute(stmt, {"tz": account.timezone})
-            for row in rs:
-                response_data.append({"date": str(row.date), "conversation_count": row.conversation_count})
+            rs = conn.execute(sa.text(sql_query), arg_dict)
+            for i in rs:
+                response_data.append({"date": str(i.date), "conversation_count": i.conversation_count})
 
         return jsonify({"data": response_data})
 
@@ -188,8 +175,9 @@ class DailyTerminalsStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     COUNT(DISTINCT messages.from_end_user_id) AS terminal_count
 FROM
     messages
@@ -198,26 +186,17 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -262,8 +241,9 @@ class DailyTokenCostStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     (SUM(messages.message_tokens) + SUM(messages.answer_tokens)) AS token_count,
     SUM(total_price) AS total_price
 FROM
@@ -273,26 +253,17 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -339,8 +310,9 @@ class AverageSessionInteractionStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', c.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("c.created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     AVG(subquery.message_count) AS interactions
 FROM
     (
@@ -357,26 +329,17 @@ FROM
             AND m.invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND c.created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND c.created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -432,8 +395,9 @@ class UserSatisfactionRateStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', m.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("m.created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     COUNT(m.id) AS message_count,
     COUNT(mf.id) AS feedback_count
 FROM
@@ -446,26 +410,17 @@ WHERE
     AND m.invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND m.created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND m.created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -515,8 +470,9 @@ class AverageResponseTimeStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     AVG(provider_response_latency) AS latency
 FROM
     messages
@@ -525,26 +481,17 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -589,8 +536,9 @@ class TokensPerSecondStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     CASE
         WHEN SUM(provider_response_latency) = 0 THEN 0
         ELSE (SUM(answer_tokens) / SUM(provider_response_latency))
@@ -602,26 +550,17 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 

api/controllers/console/app/workflow.py

@@ -102,7 +102,18 @@ class DraftWorkflowApi(Resource):
             },
         )
     )
-    @api.response(200, "Draft workflow synced successfully", workflow_fields)
+    @api.response(
+        200,
+        "Draft workflow synced successfully",
+        api.model(
+            "SyncDraftWorkflowResponse",
+            {
+                "result": fields.String,
+                "hash": fields.String,
+                "updated_at": fields.String,
+            },
+        ),
+    )
     @api.response(400, "Invalid workflow configuration")
     @api.response(403, "Permission denied")
     @edit_permission_required

api/controllers/console/app/workflow_statistic.py

@@ -1,7 +1,4 @@
-from datetime import datetime
-
-import pytz
-from flask import jsonify
+from flask import abort, jsonify
 from flask_restx import Resource, reqparse
 from sqlalchemy.orm import sessionmaker
 
@@ -9,6 +6,7 @@ from controllers.console import api, console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from extensions.ext_database import db
+from libs.datetime_utils import parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models.enums import WorkflowRunTriggeredFrom
@@ -43,23 +41,11 @@ class WorkflowDailyRunsStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        start_date = None
-        end_date = None
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_date = start_datetime_timezone.astimezone(utc_timezone)
-
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_date = end_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         response_data = self._workflow_run_repo.get_daily_runs_statistics(
             tenant_id=app_model.tenant_id,
@@ -100,23 +86,11 @@ class WorkflowDailyTerminalsStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        start_date = None
-        end_date = None
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_date = start_datetime_timezone.astimezone(utc_timezone)
-
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_date = end_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         response_data = self._workflow_run_repo.get_daily_terminals_statistics(
             tenant_id=app_model.tenant_id,
@@ -157,23 +131,11 @@ class WorkflowDailyTokenCostStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        start_date = None
-        end_date = None
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_date = start_datetime_timezone.astimezone(utc_timezone)
-
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_date = end_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         response_data = self._workflow_run_repo.get_daily_token_cost_statistics(
             tenant_id=app_model.tenant_id,
@@ -214,23 +176,11 @@ class WorkflowAverageAppInteractionStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        start_date = None
-        end_date = None
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_date = start_datetime_timezone.astimezone(utc_timezone)
-
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_date = end_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         response_data = self._workflow_run_repo.get_average_app_interaction_statistics(
             tenant_id=app_model.tenant_id,

api/controllers/console/billing/billing.py

@@ -2,6 +2,7 @@ from flask_restx import Resource, reqparse
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
+from enums.cloud_plan import CloudPlan
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService
 
@@ -16,7 +17,13 @@ class Subscription(Resource):
         current_user, current_tenant_id = current_account_with_tenant()
         parser = (
             reqparse.RequestParser()
-            .add_argument("plan", type=str, required=True, location="args", choices=["professional", "team"])
+            .add_argument(
+                "plan",
+                type=str,
+                required=True,
+                location="args",
+                choices=[CloudPlan.PROFESSIONAL, CloudPlan.TEAM],
+            )
             .add_argument("interval", type=str, required=True, location="args", choices=["month", "year"])
         )
         args = parser.parse_args()

api/controllers/console/datasets/datasets_document.py

@@ -746,7 +746,7 @@ class DocumentApi(DocumentResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": document.created_at.timestamp(),
+                "created_at": int(document.created_at.timestamp()),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -779,7 +779,7 @@ class DocumentApi(DocumentResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": document.created_at.timestamp(),
+                "created_at": int(document.created_at.timestamp()),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,

api/controllers/console/files.py

@@ -8,6 +8,7 @@ import services
 from configs import dify_config
 from constants import DOCUMENT_EXTENSIONS
 from controllers.common.errors import (
+    BlockedFileExtensionError,
     FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
@@ -83,6 +84,8 @@ class FileApi(Resource):
             raise FileTooLargeError(file_too_large_error.description)
         except services.errors.file.UnsupportedFileTypeError:
             raise UnsupportedFileTypeError()
+        except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
+            raise BlockedFileExtensionError(blocked_extension_error.description)
 
         return upload_file, 201
 

api/controllers/console/workspace/workspace.py

@@ -21,6 +21,7 @@ from controllers.console.wraps import (
     cloud_edition_billing_resource_check,
     setup_required,
 )
+from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
@@ -83,7 +84,7 @@ class TenantListApi(Resource):
                 "name": tenant.name,
                 "status": tenant.status,
                 "created_at": tenant.created_at,
-                "plan": features.billing.subscription.plan if features.billing.enabled else "sandbox",
+                "plan": features.billing.subscription.plan if features.billing.enabled else CloudPlan.SANDBOX,
                 "current": tenant.id == current_tenant_id if current_tenant_id else False,
             }
 

api/controllers/console/wraps.py

@@ -10,6 +10,7 @@ from flask import abort, request
 
 from configs import dify_config
 from controllers.console.workspace.error import AccountNotInitializedError
+from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.login import current_account_with_tenant
@@ -133,7 +134,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str):
             features = FeatureService.get_features(current_tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
-                    if features.billing.subscription.plan == "sandbox":
+                    if features.billing.subscription.plan == CloudPlan.SANDBOX:
                         abort(
                             403,
                             "To unlock this feature and elevate your Dify experience, please upgrade to a paid plan.",

api/controllers/service_api/dataset/document.py

@@ -592,7 +592,7 @@ class DocumentApi(DatasetApiResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": document.created_at.timestamp(),
+                "created_at": int(document.created_at.timestamp()),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -625,7 +625,7 @@ class DocumentApi(DatasetApiResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": document.created_at.timestamp(),
+                "created_at": int(document.created_at.timestamp()),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,

api/controllers/service_api/dataset/segment.py

@@ -2,6 +2,7 @@ from flask import request
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import NotFound
 
+from configs import dify_config
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import ProviderNotInitializeError
 from controllers.service_api.wraps import (
@@ -107,6 +108,10 @@ class SegmentApi(DatasetApiResource):
         # validate args
         args = segment_create_parser.parse_args()
         if args["segments"] is not None:
+            segments_limit = dify_config.DATASET_MAX_SEGMENTS_PER_REQUEST
+            if segments_limit > 0 and len(args["segments"]) > segments_limit:
+                raise ValueError(f"Exceeded maximum segments limit of {segments_limit}.")
+
             for args_item in args["segments"]:
                 SegmentService.segment_create_args_validate(args_item, document)
             segments = SegmentService.multi_create_segment(args["segments"], document, dataset)

api/controllers/service_api/wraps.py

@@ -13,6 +13,7 @@ from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
+from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.datetime_utils import naive_utc_now
@@ -66,6 +67,7 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
 
             kwargs["app_model"] = app_model
 
+            # If caller needs end-user context, attach EndUser to current_user
             if fetch_user_arg:
                 if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
                     user_id = request.args.get("user")
@@ -74,7 +76,6 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
                 elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
                     user_id = request.form.get("user")
                 else:
-                    # use default-user
                     user_id = None
 
                 if not user_id and fetch_user_arg.required:
@@ -89,6 +90,28 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
                 # Set EndUser as current logged-in user for flask_login.current_user
                 current_app.login_manager._update_request_context_with_user(end_user)  # type: ignore
                 user_logged_in.send(current_app._get_current_object(), user=end_user)  # type: ignore
+            else:
+                # For service API without end-user context, ensure an Account is logged in
+                # so services relying on current_account_with_tenant() work correctly.
+                tenant_owner_info = (
+                    db.session.query(Tenant, Account)
+                    .join(TenantAccountJoin, Tenant.id == TenantAccountJoin.tenant_id)
+                    .join(Account, TenantAccountJoin.account_id == Account.id)
+                    .where(
+                        Tenant.id == app_model.tenant_id,
+                        TenantAccountJoin.role == "owner",
+                        Tenant.status == TenantStatus.NORMAL,
+                    )
+                    .one_or_none()
+                )
+
+                if tenant_owner_info:
+                    tenant_model, account = tenant_owner_info
+                    account.current_tenant = tenant_model
+                    current_app.login_manager._update_request_context_with_user(account)  # type: ignore
+                    user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
+                else:
+                    raise Unauthorized("Tenant owner account not found or tenant is not active.")
 
             return view_func(*args, **kwargs)
 
@@ -138,7 +161,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: s
             features = FeatureService.get_features(api_token.tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
-                    if features.billing.subscription.plan == "sandbox":
+                    if features.billing.subscription.plan == CloudPlan.SANDBOX:
                         raise Forbidden(
                             "To unlock this feature and elevate your Dify experience, please upgrade to a paid plan."
                         )

api/core/app/apps/agent_chat/app_runner.py

@@ -144,7 +144,7 @@ class AgentChatAppRunner(AppRunner):
             prompt_template_entity=app_config.prompt_template,
             inputs=dict(inputs),
             files=list(files),
-            query=query or "",
+            query=query,
             memory=memory,
         )
 
@@ -172,7 +172,7 @@ class AgentChatAppRunner(AppRunner):
             prompt_template_entity=app_config.prompt_template,
             inputs=dict(inputs),
             files=list(files),
-            query=query or "",
+            query=query,
             memory=memory,
         )
 

api/core/app/apps/base_app_runner.py

@@ -79,7 +79,7 @@ class AppRunner:
         prompt_template_entity: PromptTemplateEntity,
         inputs: Mapping[str, str],
         files: Sequence["File"],
-        query: str | None = None,
+        query: str = "",
         context: str | None = None,
         memory: TokenBufferMemory | None = None,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
@@ -105,7 +105,7 @@ class AppRunner:
                 app_mode=AppMode.value_of(app_record.mode),
                 prompt_template_entity=prompt_template_entity,
                 inputs=inputs,
-                query=query or "",
+                query=query,
                 files=files,
                 context=context,
                 memory=memory,

api/core/app/apps/message_based_app_generator.py

@@ -190,7 +190,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             override_model_configs=json.dumps(override_model_configs) if override_model_configs else None,
             conversation_id=conversation.id,
             inputs=application_generate_entity.inputs,
-            query=application_generate_entity.query or "",
+            query=application_generate_entity.query,
             message="",
             message_tokens=0,
             message_unit_price=0,

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

@@ -41,18 +41,14 @@ from core.workflow.repositories.workflow_execution_repository import WorkflowExe
 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 extensions.ext_redis import redis_client
 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.datasource_provider_service import DatasourceProviderService
-from services.feature_service import FeatureService
-from services.file_service import FileService
+from services.rag_pipeline.rag_pipeline_task_proxy import RagPipelineTaskProxy
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
-from tasks.rag_pipeline.priority_rag_pipeline_run_task import priority_rag_pipeline_run_task
-from tasks.rag_pipeline.rag_pipeline_run_task import rag_pipeline_run_task
 
 logger = logging.getLogger(__name__)
 
@@ -248,34 +244,7 @@ class PipelineGenerator(BaseAppGenerator):
                 )
 
         if rag_pipeline_invoke_entities:
-            # store the rag_pipeline_invoke_entities to object storage
-            text = [item.model_dump() for item in rag_pipeline_invoke_entities]
-            name = "rag_pipeline_invoke_entities.json"
-            # Convert list to proper JSON string
-            json_text = json.dumps(text)
-            upload_file = FileService(db.engine).upload_text(json_text, name, user.id, dataset.tenant_id)
-            features = FeatureService.get_features(dataset.tenant_id)
-            if features.billing.enabled and features.billing.subscription.plan == "sandbox":
-                tenant_pipeline_task_key = f"tenant_pipeline_task:{dataset.tenant_id}"
-                tenant_self_pipeline_task_queue = f"tenant_self_pipeline_task_queue:{dataset.tenant_id}"
-
-                if redis_client.get(tenant_pipeline_task_key):
-                    # Add to waiting queue using List operations (lpush)
-                    redis_client.lpush(tenant_self_pipeline_task_queue, upload_file.id)
-                else:
-                    # Set flag and execute task
-                    redis_client.set(tenant_pipeline_task_key, 1, ex=60 * 60)
-                    rag_pipeline_run_task.delay(  # type: ignore
-                        rag_pipeline_invoke_entities_file_id=upload_file.id,
-                        tenant_id=dataset.tenant_id,
-                    )
-
-            else:
-                priority_rag_pipeline_run_task.delay(  # type: ignore
-                    rag_pipeline_invoke_entities_file_id=upload_file.id,
-                    tenant_id=dataset.tenant_id,
-                )
-
+            RagPipelineTaskProxy(dataset.tenant_id, user.id, rag_pipeline_invoke_entities).delay()
         # return batch, dataset, documents
         return {
             "batch": batch,

api/core/app/entities/app_invoke_entities.py

@@ -129,7 +129,7 @@ class EasyUIBasedAppGenerateEntity(AppGenerateEntity):
     app_config: EasyUIBasedAppConfig = None  # type: ignore
     model_conf: ModelConfigWithCredentialsEntity
 
-    query: str | None = None
+    query: str = ""
 
     # pydantic configs
     model_config = ConfigDict(protected_namespaces=())

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -121,7 +121,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         if self._application_generate_entity.app_config.app_mode != AppMode.COMPLETION:
             # start generate conversation name thread
             self._conversation_name_generate_thread = self._message_cycle_manager.generate_conversation_name(
-                conversation_id=self._conversation_id, query=self._application_generate_entity.query or ""
+                conversation_id=self._conversation_id, query=self._application_generate_entity.query
             )
 
         generator = self._wrapper_process_stream_response(trace_manager=self._application_generate_entity.trace_manager)

api/core/app/task_pipeline/message_cycle_manager.py

@@ -140,7 +140,27 @@ class MessageCycleManager:
         if not self._application_generate_entity.app_config.additional_features:
             raise ValueError("Additional features not found")
         if self._application_generate_entity.app_config.additional_features.show_retrieve_source:
-            self._task_state.metadata.retriever_resources = event.retriever_resources
+            merged_resources = [r for r in self._task_state.metadata.retriever_resources or [] if r]
+            existing_ids = {(r.dataset_id, r.document_id) for r in merged_resources if r.dataset_id and r.document_id}
+
+            # Add new unique resources from the event
+            for resource in event.retriever_resources or []:
+                if not resource:
+                    continue
+
+                is_duplicate = (
+                    resource.dataset_id
+                    and resource.document_id
+                    and (resource.dataset_id, resource.document_id) in existing_ids
+                )
+
+                if not is_duplicate:
+                    merged_resources.append(resource)
+
+            for i, resource in enumerate(merged_resources, 1):
+                resource.position = i
+
+            self._task_state.metadata.retriever_resources = merged_resources
 
     def message_file_to_stream_response(self, event: QueueMessageFileEvent) -> MessageFileStreamResponse | None:
         """

api/core/entities/document_task.py

@@ -0,0 +1,15 @@
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+
+@dataclass
+class DocumentTask:
+    """Document task entity for document indexing operations.
+
+    This class represents a document indexing task that can be queued
+    and processed by the document indexing system.
+    """
+
+    tenant_id: str
+    dataset_id: str
+    document_ids: Sequence[str]

api/core/entities/provider_configuration.py

@@ -1533,6 +1533,9 @@ class ProviderConfiguration(BaseModel):
             # Return composite sort key: (model_type value, model position index)
             return (model.model_type.value, position_index)
 
+        # Deduplicate
+        provider_models = list({(m.model, m.model_type, m.fetch_from): m for m in provider_models}.values())
+
         # Sort using the composite sort key
         return sorted(provider_models, key=get_sort_key)
 

api/core/file/models.py

@@ -74,6 +74,10 @@ class File(BaseModel):
         storage_key: str | None = None,
         dify_model_identity: str | None = FILE_MODEL_IDENTITY,
         url: str | None = None,
+        # Legacy compatibility fields - explicitly handle known extra fields
+        tool_file_id: str | None = None,
+        upload_file_id: str | None = None,
+        datasource_file_id: str | None = None,
     ):
         super().__init__(
             id=id,

api/core/helper/code_executor/javascript/javascript_transformer.py

@@ -6,10 +6,7 @@ from core.helper.code_executor.template_transformer import TemplateTransformer
 class NodeJsTemplateTransformer(TemplateTransformer):
     @classmethod
     def get_runner_script(cls) -> str:
-        runner_script = dedent(
-            f"""
-            // declare main function
-            {cls._code_placeholder}
+        runner_script = dedent(f"""            {cls._code_placeholder}
 
             // decode and prepare input object
             var inputs_obj = JSON.parse(Buffer.from('{cls._inputs_placeholder}', 'base64').toString('utf-8'))
@@ -21,6 +18,5 @@ class NodeJsTemplateTransformer(TemplateTransformer):
             var output_json = JSON.stringify(output_obj)
             var result = `<<RESULT>>${{output_json}}<<RESULT>>`
             console.log(result)
-            """
-        )
+            """)
         return runner_script

api/core/helper/code_executor/python3/python3_transformer.py

@@ -6,9 +6,7 @@ from core.helper.code_executor.template_transformer import TemplateTransformer
 class Python3TemplateTransformer(TemplateTransformer):
     @classmethod
     def get_runner_script(cls) -> str:
-        runner_script = dedent(f"""
-            # declare main function
-            {cls._code_placeholder}
+        runner_script = dedent(f"""            {cls._code_placeholder}
 
             import json
             from base64 import b64decode

api/core/ops/tencent_trace/client.py

@@ -5,6 +5,7 @@ Tencent APM Trace Client - handles network operations, metrics, and API communic
 from __future__ import annotations
 
 import importlib
+import json
 import logging
 import os
 import socket
@@ -110,6 +111,7 @@ class TencentTraceClient:
         self.span_contexts: dict[int, trace_api.SpanContext] = {}
 
         self.meter: Meter | None = None
+        self.meter_provider: MeterProvider | None = None
         self.hist_llm_duration: Histogram | None = None
         self.hist_token_usage: Histogram | None = None
         self.hist_time_to_first_token: Histogram | None = None
@@ -119,7 +121,6 @@ class TencentTraceClient:
 
         # Metrics exporter and instruments
         try:
-            from opentelemetry import metrics
             from opentelemetry.sdk.metrics import Histogram, MeterProvider
             from opentelemetry.sdk.metrics.export import AggregationTemporality, PeriodicExportingMetricReader
 
@@ -202,9 +203,11 @@ class TencentTraceClient:
                 )
 
             if metric_reader is not None:
+                # Use instance-level MeterProvider instead of global to support config changes
+                # without worker restart. Each TencentTraceClient manages its own MeterProvider.
                 provider = MeterProvider(resource=self.resource, metric_readers=[metric_reader])
-                metrics.set_meter_provider(provider)
-                self.meter = metrics.get_meter("dify-sdk", dify_config.project.version)
+                self.meter_provider = provider
+                self.meter = provider.get_meter("dify-sdk", dify_config.project.version)
 
                 # LLM operation duration histogram
                 self.hist_llm_duration = self.meter.create_histogram(
@@ -244,6 +247,7 @@ class TencentTraceClient:
                 self.metric_reader = metric_reader
             else:
                 self.meter = None
+                self.meter_provider = None
                 self.hist_llm_duration = None
                 self.hist_token_usage = None
                 self.hist_time_to_first_token = None
@@ -253,6 +257,7 @@ class TencentTraceClient:
         except Exception:
             logger.exception("[Tencent APM] Metrics initialization failed; metrics disabled")
             self.meter = None
+            self.meter_provider = None
             self.hist_llm_duration = None
             self.hist_token_usage = None
             self.hist_time_to_first_token = None
@@ -279,6 +284,14 @@ class TencentTraceClient:
             if attributes:
                 for k, v in attributes.items():
                     attrs[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v  # type: ignore[assignment]
+
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
+                LLM_OPERATION_DURATION,
+                latency_seconds,
+                json.dumps(attrs, ensure_ascii=False),
+            )
+
             self.hist_llm_duration.record(latency_seconds, attrs)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record LLM duration", exc_info=True)
@@ -317,6 +330,13 @@ class TencentTraceClient:
                 "server.address": server_address,
             }
 
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %d | Attributes: %s",
+                GEN_AI_TOKEN_USAGE,
+                token_count,
+                json.dumps(attributes, ensure_ascii=False),
+            )
+
             self.hist_token_usage.record(token_count, attributes)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record token usage", exc_info=True)
@@ -344,6 +364,13 @@ class TencentTraceClient:
                 "stream": "true",
             }
 
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
+                GEN_AI_SERVER_TIME_TO_FIRST_TOKEN,
+                ttft_seconds,
+                json.dumps(attributes, ensure_ascii=False),
+            )
+
             self.hist_time_to_first_token.record(ttft_seconds, attributes)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record time to first token", exc_info=True)
@@ -371,6 +398,13 @@ class TencentTraceClient:
                 "stream": "true",
             }
 
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
+                GEN_AI_STREAMING_TIME_TO_GENERATE,
+                ttg_seconds,
+                json.dumps(attributes, ensure_ascii=False),
+            )
+
             self.hist_time_to_generate.record(ttg_seconds, attributes)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record time to generate", exc_info=True)
@@ -390,6 +424,14 @@ class TencentTraceClient:
             if attributes:
                 for k, v in attributes.items():
                     attrs[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v  # type: ignore[assignment]
+
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
+                GEN_AI_TRACE_DURATION,
+                duration_seconds,
+                json.dumps(attrs, ensure_ascii=False),
+            )
+
             self.hist_trace_duration.record(duration_seconds, attrs)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record trace duration", exc_info=True)
@@ -474,11 +516,19 @@ class TencentTraceClient:
 
             if self.tracer_provider:
                 self.tracer_provider.shutdown()
+
+            # Shutdown instance-level meter provider
+            if self.meter_provider is not None:
+                try:
+                    self.meter_provider.shutdown()  # type: ignore[attr-defined]
+                except Exception:
+                    logger.debug("[Tencent APM] Error shutting down meter provider", exc_info=True)
+
             if self.metric_reader is not None:
                 try:
                     self.metric_reader.shutdown()  # type: ignore[attr-defined]
                 except Exception:
-                    pass
+                    logger.debug("[Tencent APM] Error shutting down metric reader", exc_info=True)
 
         except Exception:
             logger.exception("[Tencent APM] Error during client shutdown")

api/core/rag/datasource/vdb/elasticsearch/elasticsearch_vector.py

@@ -147,7 +147,8 @@ class ElasticSearchVector(BaseVector):
 
     def _get_version(self) -> str:
         info = self._client.info()
-        return cast(str, info["version"]["number"])
+        # remove any suffix like "-SNAPSHOT" from the version string
+        return cast(str, info["version"]["number"]).split("-")[0]
 
     def _check_version(self):
         if parse_version(self._version) < parse_version("8.0.0"):

api/core/rag/datasource/vdb/oceanbase/oceanbase_vector.py

@@ -37,13 +37,13 @@ class OceanBaseVectorConfig(BaseModel):
     @classmethod
     def validate_config(cls, values: dict):
         if not values["host"]:
-            raise ValueError("config OCEANBASE_VECTOR_HOST is required")
+            raise ValueError("config OCEANBASE_HOST is required")
         if not values["port"]:
-            raise ValueError("config OCEANBASE_VECTOR_PORT is required")
+            raise ValueError("config OCEANBASE_PORT is required")
         if not values["user"]:
-            raise ValueError("config OCEANBASE_VECTOR_USER is required")
+            raise ValueError("config OCEANBASE_USER is required")
         if not values["database"]:
-            raise ValueError("config OCEANBASE_VECTOR_DATABASE is required")
+            raise ValueError("config OCEANBASE_DATABASE is required")
         return values
 
 
@@ -316,11 +316,11 @@ class OceanBaseVectorFactory(AbstractVectorFactory):
         return OceanBaseVector(
             collection_name,
             OceanBaseVectorConfig(
-                host=dify_config.OCEANBASE_VECTOR_HOST or "",
-                port=dify_config.OCEANBASE_VECTOR_PORT or 0,
-                user=dify_config.OCEANBASE_VECTOR_USER or "",
-                password=(dify_config.OCEANBASE_VECTOR_PASSWORD or ""),
-                database=dify_config.OCEANBASE_VECTOR_DATABASE or "",
+                host=dify_config.OCEANBASE_HOST or "",
+                port=dify_config.OCEANBASE_PORT or 0,
+                user=dify_config.OCEANBASE_USER or "",
+                password=(dify_config.OCEANBASE_PASSWORD or ""),
+                database=dify_config.OCEANBASE_DATABASE or "",
                 enable_hybrid_search=dify_config.OCEANBASE_ENABLE_HYBRID_SEARCH or False,
             ),
         )

api/core/rag/datasource/vdb/weaviate/weaviate_vector.py

@@ -39,11 +39,13 @@ class WeaviateConfig(BaseModel):
 
     Attributes:
         endpoint: Weaviate server endpoint URL
+        grpc_endpoint: Optional Weaviate gRPC server endpoint URL
         api_key: Optional API key for authentication
         batch_size: Number of objects to batch per insert operation
     """
 
     endpoint: str
+    grpc_endpoint: str | None = None
     api_key: str | None = None
     batch_size: int = 100
 
@@ -88,9 +90,22 @@ class WeaviateVector(BaseVector):
         http_secure = p.scheme == "https"
         http_port = p.port or (443 if http_secure else 80)
 
-        grpc_host = host
-        grpc_secure = http_secure
-        grpc_port = 443 if grpc_secure else 50051
+        # Parse gRPC configuration
+        if config.grpc_endpoint:
+            # Urls without scheme won't be parsed correctly in some python versions,
+            # see https://bugs.python.org/issue27657
+            grpc_endpoint_with_scheme = (
+                config.grpc_endpoint if "://" in config.grpc_endpoint else f"grpc://{config.grpc_endpoint}"
+            )
+            grpc_p = urlparse(grpc_endpoint_with_scheme)
+            grpc_host = grpc_p.hostname or "localhost"
+            grpc_port = grpc_p.port or (443 if grpc_p.scheme == "grpcs" else 50051)
+            grpc_secure = grpc_p.scheme == "grpcs"
+        else:
+            # Infer from HTTP endpoint as fallback
+            grpc_host = host
+            grpc_secure = http_secure
+            grpc_port = 443 if grpc_secure else 50051
 
         client = weaviate.connect_to_custom(
             http_host=host,
@@ -432,6 +447,7 @@ class WeaviateVectorFactory(AbstractVectorFactory):
             collection_name=collection_name,
             config=WeaviateConfig(
                 endpoint=dify_config.WEAVIATE_ENDPOINT or "",
+                grpc_endpoint=dify_config.WEAVIATE_GRPC_ENDPOINT or "",
                 api_key=dify_config.WEAVIATE_API_KEY,
                 batch_size=dify_config.WEAVIATE_BATCH_SIZE,
             ),

api/core/rag/pipeline/queue.py

@@ -0,0 +1,79 @@
+import json
+from collections.abc import Sequence
+from typing import Any
+
+from pydantic import BaseModel, ValidationError
+
+from extensions.ext_redis import redis_client
+
+_DEFAULT_TASK_TTL = 60 * 60  # 1 hour
+
+
+class TaskWrapper(BaseModel):
+    data: Any
+
+    def serialize(self) -> str:
+        return self.model_dump_json()
+
+    @classmethod
+    def deserialize(cls, serialized_data: str) -> "TaskWrapper":
+        return cls.model_validate_json(serialized_data)
+
+
+class TenantIsolatedTaskQueue:
+    """
+    Simple queue for tenant isolated tasks, used for rag related tenant tasks isolation.
+    It uses Redis list to store tasks, and Redis key to store task waiting flag.
+    Support tasks that can be serialized by json.
+    """
+
+    def __init__(self, tenant_id: str, unique_key: str):
+        self._tenant_id = tenant_id
+        self._unique_key = unique_key
+        self._queue = f"tenant_self_{unique_key}_task_queue:{tenant_id}"
+        self._task_key = f"tenant_{unique_key}_task:{tenant_id}"
+
+    def get_task_key(self):
+        return redis_client.get(self._task_key)
+
+    def set_task_waiting_time(self, ttl: int = _DEFAULT_TASK_TTL):
+        redis_client.setex(self._task_key, ttl, 1)
+
+    def delete_task_key(self):
+        redis_client.delete(self._task_key)
+
+    def push_tasks(self, tasks: Sequence[Any]):
+        serialized_tasks = []
+        for task in tasks:
+            # Store str list directly, maintaining full compatibility for pipeline scenarios
+            if isinstance(task, str):
+                serialized_tasks.append(task)
+            else:
+                # Use TaskWrapper to do JSON serialization for non-string tasks
+                wrapper = TaskWrapper(data=task)
+                serialized_data = wrapper.serialize()
+                serialized_tasks.append(serialized_data)
+
+        redis_client.lpush(self._queue, *serialized_tasks)
+
+    def pull_tasks(self, count: int = 1) -> Sequence[Any]:
+        if count <= 0:
+            return []
+
+        tasks = []
+        for _ in range(count):
+            serialized_task = redis_client.rpop(self._queue)
+            if not serialized_task:
+                break
+
+            if isinstance(serialized_task, bytes):
+                serialized_task = serialized_task.decode("utf-8")
+
+            try:
+                wrapper = TaskWrapper.deserialize(serialized_task)
+                tasks.append(wrapper.data)
+            except (json.JSONDecodeError, ValidationError, TypeError, ValueError):
+                # Fall back to raw string for legacy format or invalid JSON
+                tasks.append(serialized_task)
+
+        return tasks

api/core/tools/__base/tool.py

@@ -210,12 +210,13 @@ class Tool(ABC):
             meta=meta,
         )
 
-    def create_json_message(self, object: dict) -> ToolInvokeMessage:
+    def create_json_message(self, object: dict, suppress_output: bool = False) -> ToolInvokeMessage:
         """
         create a json message
         """
         return ToolInvokeMessage(
-            type=ToolInvokeMessage.MessageType.JSON, message=ToolInvokeMessage.JsonMessage(json_object=object)
+            type=ToolInvokeMessage.MessageType.JSON,
+            message=ToolInvokeMessage.JsonMessage(json_object=object, suppress_output=suppress_output),
         )
 
     def create_variable_message(

api/core/tools/entities/tool_entities.py

@@ -129,6 +129,7 @@ class ToolInvokeMessage(BaseModel):
 
     class JsonMessage(BaseModel):
         json_object: dict
+        suppress_output: bool = Field(default=False, description="Whether to suppress JSON output in result string")
 
     class BlobMessage(BaseModel):
         blob: bytes

api/core/tools/mcp_tool/tool.py

@@ -1,16 +1,19 @@
 import base64
 import json
+import logging
 from collections.abc import Generator
 from typing import Any
 
 from core.mcp.auth_client import MCPClientWithAuthRetry
 from core.mcp.error import MCPConnectionError
-from core.mcp.types import CallToolResult, ImageContent, TextContent
+from core.mcp.types import AudioContent, CallToolResult, ImageContent, TextContent
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.entities.tool_entities import ToolEntity, ToolInvokeMessage, ToolProviderType
 from core.tools.errors import ToolInvokeError
 
+logger = logging.getLogger(__name__)
+
 
 class MCPTool(Tool):
     def __init__(
@@ -52,6 +55,11 @@ class MCPTool(Tool):
                 yield from self._process_text_content(content)
             elif isinstance(content, ImageContent):
                 yield self._process_image_content(content)
+            elif isinstance(content, AudioContent):
+                yield self._process_audio_content(content)
+            else:
+                logger.warning("Unsupported content type=%s", type(content))
+
         # handle MCP structured output
         if self.entity.output_schema and result.structuredContent:
             for k, v in result.structuredContent.items():
@@ -97,6 +105,10 @@ class MCPTool(Tool):
         """Process image content and return a blob message."""
         return self.create_blob_message(blob=base64.b64decode(content.data), meta={"mime_type": content.mimeType})
 
+    def _process_audio_content(self, content: AudioContent) -> ToolInvokeMessage:
+        """Process audio content and return a blob message."""
+        return self.create_blob_message(blob=base64.b64decode(content.data), meta={"mime_type": content.mimeType})
+
     def fork_tool_runtime(self, runtime: ToolRuntime) -> "MCPTool":
         return MCPTool(
             entity=self.entity,

api/core/tools/tool_engine.py

@@ -228,29 +228,41 @@ class ToolEngine:
         """
         Handle tool response
         """
-        result = ""
+        parts: list[str] = []
+        json_parts: list[str] = []
+
         for response in tool_response:
             if response.type == ToolInvokeMessage.MessageType.TEXT:
-                result += cast(ToolInvokeMessage.TextMessage, response.message).text
+                parts.append(cast(ToolInvokeMessage.TextMessage, response.message).text)
             elif response.type == ToolInvokeMessage.MessageType.LINK:
-                result += (
+                parts.append(
                     f"result link: {cast(ToolInvokeMessage.TextMessage, response.message).text}."
                     + " please tell user to check it."
                 )
             elif response.type in {ToolInvokeMessage.MessageType.IMAGE_LINK, ToolInvokeMessage.MessageType.IMAGE}:
-                result += (
+                parts.append(
                     "image has been created and sent to user already, "
                     + "you do not need to create it, just tell the user to check it now."
                 )
             elif response.type == ToolInvokeMessage.MessageType.JSON:
-                result += json.dumps(
-                    safe_json_value(cast(ToolInvokeMessage.JsonMessage, response.message).json_object),
-                    ensure_ascii=False,
+                json_message = cast(ToolInvokeMessage.JsonMessage, response.message)
+                if json_message.suppress_output:
+                    continue
+                json_parts.append(
+                    json.dumps(
+                        safe_json_value(cast(ToolInvokeMessage.JsonMessage, response.message).json_object),
+                        ensure_ascii=False,
+                    )
                 )
             else:
-                result += str(response.message)
+                parts.append(str(response.message))
 
-        return result
+        # Add JSON parts, avoiding duplicates from text parts.
+        if json_parts:
+            existing_parts = set(parts)
+            parts.extend(p for p in json_parts if p not in existing_parts)
+
+        return "".join(parts)
 
     @staticmethod
     def _extract_tool_response_binary_and_text(

api/core/tools/tool_manager.py

@@ -14,6 +14,7 @@ from sqlalchemy.orm import Session
 from yarl import URL
 
 import contexts
+from configs import dify_config
 from core.helper.provider_cache import ToolProviderCredentialsCache
 from core.plugin.impl.tool import PluginToolManager
 from core.tools.__base.tool_provider import ToolProviderController
@@ -33,7 +34,6 @@ from services.tools.mcp_tools_manage_service import MCPToolManageService
 if TYPE_CHECKING:
     from core.workflow.nodes.tool.entities import ToolEntity
 
-from configs import dify_config
 from core.agent.entities import AgentToolEntity
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.helper.module_import_helper import load_single_subclass_from_source
@@ -621,12 +621,28 @@ class ToolManager:
         """
         # according to multi credentials, select the one with is_default=True first, then created_at oldest
         # for compatibility with old version
-        sql = """
+        if dify_config.SQLALCHEMY_DATABASE_URI_SCHEME == "postgresql":
+            # PostgreSQL: Use DISTINCT ON
+            sql = """
                 SELECT DISTINCT ON (tenant_id, provider) id
                 FROM tool_builtin_providers
                 WHERE tenant_id = :tenant_id
                 ORDER BY tenant_id, provider, is_default DESC, created_at DESC
                 """
+        else:
+            # MySQL: Use window function to achieve same result
+            sql = """
+                SELECT id FROM (
+                    SELECT id, 
+                           ROW_NUMBER() OVER (
+                               PARTITION BY tenant_id, provider 
+                               ORDER BY is_default DESC, created_at DESC
+                           ) as rn
+                    FROM tool_builtin_providers
+                    WHERE tenant_id = :tenant_id
+                ) ranked WHERE rn = 1
+                """
+        
         with Session(db.engine, autoflush=False) as session:
             ids = [row.id for row in session.execute(sa.text(sql), {"tenant_id": tenant_id}).all()]
             return session.query(BuiltinToolProvider).where(BuiltinToolProvider.id.in_(ids)).all()

api/core/tools/workflow_as_tool/tool.py

@@ -117,7 +117,7 @@ class WorkflowTool(Tool):
         self._latest_usage = self._derive_usage_from_result(data)
 
         yield self.create_text_message(json.dumps(outputs, ensure_ascii=False))
-        yield self.create_json_message(outputs)
+        yield self.create_json_message(outputs, suppress_output=True)
 
     @property
     def latest_usage(self) -> LLMUsage:

api/core/workflow/runtime/variable_pool.py

@@ -153,7 +153,11 @@ class VariablePool(BaseModel):
             return None
 
         node_id, name = self._selector_to_keys(selector)
-        segment: Segment | None = self.variable_dictionary[node_id].get(name)
+        node_map = self.variable_dictionary.get(node_id)
+        if node_map is None:
+            return None
+
+        segment: Segment | None = node_map.get(name)
 
         if segment is None:
             return None

api/docker/entrypoint.sh

@@ -32,7 +32,7 @@ if [[ "${MODE}" == "worker" ]]; then
 
   exec celery -A celery_entrypoint.celery worker -P ${CELERY_WORKER_CLASS:-gevent} $CONCURRENCY_OPTION \
     --max-tasks-per-child ${MAX_TASKS_PER_CHILD:-50} --loglevel ${LOG_LEVEL:-INFO} \
-    -Q ${CELERY_QUEUES:-dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline} \
+    -Q ${CELERY_QUEUES:-dataset,priority_dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline} \
     --prefetch-multiplier=1
 
 elif [[ "${MODE}" == "beat" ]]; then

api/enums/cloud_plan.py

@@ -0,0 +1,15 @@
+from enum import StrEnum, auto
+
+
+class CloudPlan(StrEnum):
+    """
+    Enum representing user plan types in the cloud platform.
+
+    SANDBOX: Free/default plan with limited features
+    PROFESSIONAL: Professional paid plan
+    TEAM: Team collaboration paid plan
+    """
+
+    SANDBOX = auto()
+    PROFESSIONAL = auto()
+    TEAM = auto()

api/libs/broadcast_channel/channel.py

@@ -0,0 +1,134 @@
+"""
+Broadcast channel for Pub/Sub messaging.
+"""
+
+import types
+from abc import abstractmethod
+from collections.abc import Iterator
+from contextlib import AbstractContextManager
+from typing import Protocol, Self
+
+
+class Subscription(AbstractContextManager["Subscription"], Protocol):
+    """A subscription to a topic that provides an iterator over received messages.
+    The subscription can be used as a context manager and will automatically
+    close when exiting the context.
+
+    Note: `Subscription` instances are not thread-safe. Each thread should create its own
+    subscription.
+    """
+
+    @abstractmethod
+    def __iter__(self) -> Iterator[bytes]:
+        """`__iter__` returns an iterator used to consume the message from this subscription.
+
+        If the caller did not enter the context, `__iter__` may lazily perform the setup before
+        yielding messages; otherwise `__enter__` handles it.”
+
+        If the subscription is closed, then the returned  iterator exits without
+        raising any error.
+        """
+        ...
+
+    @abstractmethod
+    def close(self) -> None:
+        """close closes the subscription, releases any resources associated with it."""
+        ...
+
+    def __enter__(self) -> Self:
+        """`__enter__` does the setup logic of the subscription (if any), and return itself."""
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: types.TracebackType | None,
+    ) -> bool | None:
+        self.close()
+        return None
+
+    @abstractmethod
+    def receive(self, timeout: float | None = 0.1) -> bytes | None:
+        """Receive the next message from the broadcast channel.
+
+        If `timeout` is specified, this method returns `None` if no message is
+        received within the given period. If `timeout` is `None`, the call blocks
+        until a message is received.
+
+        Calling receive with `timeout=None` is highly discouraged, as it is impossible to
+        cancel a blocking subscription.
+
+        :param timeout: timeout for receive message, in seconds.
+
+        Returns:
+            bytes: The received message as a byte string, or
+            None: If the timeout expires before a message is received.
+
+        Raises:
+            SubscriptionClosed: If the subscription has already been closed.
+        """
+        ...
+
+
+class Producer(Protocol):
+    """Producer is an interface for message publishing. It is already bound to a specific topic.
+
+    `Producer` implementations must be thread-safe and support concurrent use by multiple threads.
+    """
+
+    @abstractmethod
+    def publish(self, payload: bytes) -> None:
+        """Publish a message to the bounded topic."""
+        ...
+
+
+class Subscriber(Protocol):
+    """Subscriber is an interface for subscription creation. It is already bound to a specific topic.
+
+    `Subscriber` implementations must be thread-safe and support concurrent use by multiple threads.
+    """
+
+    @abstractmethod
+    def subscribe(self) -> Subscription:
+        pass
+
+
+class Topic(Producer, Subscriber, Protocol):
+    """A named channel for publishing and subscribing to messages.
+
+    Topics provide both read and write access. For restricted access,
+    use as_producer() for write-only view or as_subscriber() for read-only view.
+
+    `Topic` implementations must be thread-safe and support concurrent use by multiple threads.
+    """
+
+    @abstractmethod
+    def as_producer(self) -> Producer:
+        """as_producer creates a write-only view for this topic."""
+        ...
+
+    @abstractmethod
+    def as_subscriber(self) -> Subscriber:
+        """as_subscriber create a read-only view for this topic."""
+        ...
+
+
+class BroadcastChannel(Protocol):
+    """A broadcasting channel is a channel supporting broadcasting semantics.
+
+    Each channel is identified by a topic, different topics are isolated and do not affect each other.
+
+    There can be multiple subscriptions to a specific topic. When a publisher publishes a message to
+    a specific topic, all subscription should receive the published message.
+
+    There are no restriction for the persistence of messages. Once a subscription is created, it
+    should receive all subsequent messages published.
+
+    `BroadcastChannel` implementations must be thread-safe and support concurrent use by multiple threads.
+    """
+
+    @abstractmethod
+    def topic(self, topic: str) -> "Topic":
+        """topic returns a `Topic` instance for the given topic name."""
+        ...

api/libs/broadcast_channel/exc.py

@@ -0,0 +1,12 @@
+class BroadcastChannelError(Exception):
+    """`BroadcastChannelError` is the base class for all exceptions related
+    to `BroadcastChannel`."""
+
+    pass
+
+
+class SubscriptionClosedError(BroadcastChannelError):
+    """SubscriptionClosedError means that the subscription has been closed and
+    methods for consuming messages should not be called."""
+
+    pass

api/libs/broadcast_channel/redis/__init__.py

@@ -0,0 +1,3 @@
+from .channel import BroadcastChannel
+
+__all__ = ["BroadcastChannel"]

api/libs/broadcast_channel/redis/channel.py

@@ -0,0 +1,200 @@
+import logging
+import queue
+import threading
+import types
+from collections.abc import Generator, Iterator
+from typing import Self
+
+from libs.broadcast_channel.channel import Producer, Subscriber, Subscription
+from libs.broadcast_channel.exc import SubscriptionClosedError
+from redis import Redis
+from redis.client import PubSub
+
+_logger = logging.getLogger(__name__)
+
+
+class BroadcastChannel:
+    """
+    Redis Pub/Sub based broadcast channel implementation.
+
+    Provides "at most once" delivery semantics for messages published to channels.
+    Uses Redis PUBLISH/SUBSCRIBE commands for real-time message delivery.
+
+    The `redis_client` used to construct BroadcastChannel should have `decode_responses` set to `False`.
+    """
+
+    def __init__(
+        self,
+        redis_client: Redis,
+    ):
+        self._client = redis_client
+
+    def topic(self, topic: str) -> "Topic":
+        return Topic(self._client, topic)
+
+
+class Topic:
+    def __init__(self, redis_client: Redis, topic: str):
+        self._client = redis_client
+        self._topic = topic
+
+    def as_producer(self) -> Producer:
+        return self
+
+    def publish(self, payload: bytes) -> None:
+        self._client.publish(self._topic, payload)
+
+    def as_subscriber(self) -> Subscriber:
+        return self
+
+    def subscribe(self) -> Subscription:
+        return _RedisSubscription(
+            pubsub=self._client.pubsub(),
+            topic=self._topic,
+        )
+
+
+class _RedisSubscription(Subscription):
+    def __init__(
+        self,
+        pubsub: PubSub,
+        topic: str,
+    ):
+        # The _pubsub is None only if the subscription is closed.
+        self._pubsub: PubSub | None = pubsub
+        self._topic = topic
+        self._closed = threading.Event()
+        self._queue: queue.Queue[bytes] = queue.Queue(maxsize=1024)
+        self._dropped_count = 0
+        self._listener_thread: threading.Thread | None = None
+        self._start_lock = threading.Lock()
+        self._started = False
+
+    def _start_if_needed(self) -> None:
+        with self._start_lock:
+            if self._started:
+                return
+            if self._closed.is_set():
+                raise SubscriptionClosedError("The Redis subscription is closed")
+            if self._pubsub is None:
+                raise SubscriptionClosedError("The Redis subscription has been cleaned up")
+
+            self._pubsub.subscribe(self._topic)
+            _logger.debug("Subscribed to channel %s", self._topic)
+
+            self._listener_thread = threading.Thread(
+                target=self._listen,
+                name=f"redis-broadcast-{self._topic}",
+                daemon=True,
+            )
+            self._listener_thread.start()
+            self._started = True
+
+    def _listen(self) -> None:
+        pubsub = self._pubsub
+        assert pubsub is not None, "PubSub should not be None while starting listening."
+        while not self._closed.is_set():
+            raw_message = pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1)
+
+            if raw_message is None:
+                continue
+
+            if raw_message.get("type") != "message":
+                continue
+
+            channel_field = raw_message.get("channel")
+            if isinstance(channel_field, bytes):
+                channel_name = channel_field.decode("utf-8")
+            elif isinstance(channel_field, str):
+                channel_name = channel_field
+            else:
+                channel_name = str(channel_field)
+
+            if channel_name != self._topic:
+                _logger.warning("Ignoring message from unexpected channel %s", channel_name)
+                continue
+
+            payload_bytes: bytes | None = raw_message.get("data")
+            if not isinstance(payload_bytes, bytes):
+                _logger.error("Received invalid data from channel %s, type=%s", self._topic, type(payload_bytes))
+                continue
+
+            self._enqueue_message(payload_bytes)
+
+        _logger.debug("Listener thread stopped for channel %s", self._topic)
+        pubsub.unsubscribe(self._topic)
+        pubsub.close()
+        _logger.debug("PubSub closed for topic %s", self._topic)
+        self._pubsub = None
+
+    def _enqueue_message(self, payload: bytes) -> None:
+        while not self._closed.is_set():
+            try:
+                self._queue.put_nowait(payload)
+                return
+            except queue.Full:
+                try:
+                    self._queue.get_nowait()
+                    self._dropped_count += 1
+                    _logger.debug(
+                        "Dropped message from Redis subscription, topic=%s, total_dropped=%d",
+                        self._topic,
+                        self._dropped_count,
+                    )
+                except queue.Empty:
+                    continue
+        return
+
+    def _message_iterator(self) -> Generator[bytes, None, None]:
+        while not self._closed.is_set():
+            try:
+                item = self._queue.get(timeout=0.1)
+            except queue.Empty:
+                continue
+
+            yield item
+
+    def __iter__(self) -> Iterator[bytes]:
+        if self._closed.is_set():
+            raise SubscriptionClosedError("The Redis subscription is closed")
+        self._start_if_needed()
+        return iter(self._message_iterator())
+
+    def receive(self, timeout: float | None = None) -> bytes | None:
+        if self._closed.is_set():
+            raise SubscriptionClosedError("The Redis subscription is closed")
+        self._start_if_needed()
+
+        try:
+            item = self._queue.get(timeout=timeout)
+        except queue.Empty:
+            return None
+
+        return item
+
+    def __enter__(self) -> Self:
+        self._start_if_needed()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: types.TracebackType | None,
+    ) -> bool | None:
+        self.close()
+        return None
+
+    def close(self) -> None:
+        if self._closed.is_set():
+            return
+
+        self._closed.set()
+        # NOTE: PubSub is not thread-safe. More specifically, the `PubSub.close` method and the `PubSub.get_message`
+        # method should NOT be called concurrently.
+        #
+        # Due to the restriction above, the PubSub cleanup logic happens inside the consumer thread.
+        listener = self._listener_thread
+        if listener is not None:
+            listener.join(timeout=1.0)
+            self._listener_thread = None

api/libs/datetime_utils.py

@@ -2,6 +2,8 @@ import abc
 import datetime
 from typing import Protocol
 
+import pytz
+
 
 class _NowFunction(Protocol):
     @abc.abstractmethod
@@ -20,3 +22,51 @@ def naive_utc_now() -> datetime.datetime:
     representing current UTC time.
     """
     return _now_func(datetime.UTC).replace(tzinfo=None)
+
+
+def parse_time_range(
+    start: str | None, end: str | None, tzname: str
+) -> tuple[datetime.datetime | None, datetime.datetime | None]:
+    """
+    Parse time range strings and convert to UTC datetime objects.
+    Handles DST ambiguity and non-existent times gracefully.
+
+    Args:
+        start: Start time string (YYYY-MM-DD HH:MM)
+        end: End time string (YYYY-MM-DD HH:MM)
+        tzname: Timezone name
+
+    Returns:
+        tuple: (start_datetime_utc, end_datetime_utc)
+
+    Raises:
+        ValueError: When time range is invalid or start > end
+    """
+    tz = pytz.timezone(tzname)
+    utc = pytz.utc
+
+    def _parse(time_str: str | None, label: str) -> datetime.datetime | None:
+        if not time_str:
+            return None
+
+        try:
+            dt = datetime.datetime.strptime(time_str, "%Y-%m-%d %H:%M").replace(second=0)
+        except ValueError as e:
+            raise ValueError(f"Invalid {label} time format: {e}")
+
+        try:
+            return tz.localize(dt, is_dst=None).astimezone(utc)
+        except pytz.AmbiguousTimeError:
+            return tz.localize(dt, is_dst=False).astimezone(utc)
+        except pytz.NonExistentTimeError:
+            dt += datetime.timedelta(hours=1)
+            return tz.localize(dt, is_dst=None).astimezone(utc)
+
+    start_dt = _parse(start, "start")
+    end_dt = _parse(end, "end")
+
+    # Range validation
+    if start_dt and end_dt and start_dt > end_dt:
+        raise ValueError("start must be earlier than or equal to end")
+
+    return start_dt, end_dt

api/libs/helper.py

@@ -177,6 +177,15 @@ def timezone(timezone_string):
     raise ValueError(error)
 
 
+def convert_datetime_to_date(field, target_timezone: str = ":tz"):
+    if dify_config.SQLALCHEMY_DATABASE_URI_SCHEME == "postgresql":
+        return f"DATE(DATE_TRUNC('day', {field} AT TIME ZONE 'UTC' AT TIME ZONE {target_timezone}))"
+    elif "mysql" in dify_config.SQLALCHEMY_DATABASE_URI_SCHEME:
+        return f"DATE(CONVERT_TZ({field}, 'UTC', {target_timezone}))"
+    else:
+        raise NotImplementedError(f"Unsupported database URI scheme: {dify_config.SQLALCHEMY_DATABASE_URI_SCHEME}")
+
+
 def generate_string(n):
     letters_digits = string.ascii_letters + string.digits
     result = ""

api/migrations/versions/00bacef91f18_rename_api_provider_description.py

@@ -8,6 +8,12 @@ Create Date: 2024-01-07 04:07:34.482983
 import sqlalchemy as sa
 from alembic import op
 
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '00bacef91f18'
 down_revision = '8ec536f3c800'
@@ -17,17 +23,31 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('description', sa.Text(), nullable=False))
-        batch_op.drop_column('description_str')
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('description', sa.Text(), nullable=False))
+            batch_op.drop_column('description_str')
+    else:
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('description', models.types.LongText(), nullable=False))
+            batch_op.drop_column('description_str')
 
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('description_str', sa.TEXT(), autoincrement=False, nullable=False))
-        batch_op.drop_column('description')
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('description_str', sa.TEXT(), autoincrement=False, nullable=False))
+            batch_op.drop_column('description')
+    else:
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('description_str', models.types.LongText(), autoincrement=False, nullable=False))
+            batch_op.drop_column('description')
 
     # ### end Alembic commands ###

api/migrations/versions/04c602f5dc9b_update_appmodelconfig_and_add_table_.py

@@ -7,9 +7,14 @@ Create Date: 2024-06-12 07:49:07.666510
 """
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 
 import models.types
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '04c602f5dc9b'
 down_revision = '4ff534e1eb11'
@@ -19,15 +24,28 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tracing_app_configs',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('app_id', models.types.StringUUID(), nullable=False),
-    sa.Column('tracing_provider', sa.String(length=255), nullable=True),
-    sa.Column('tracing_config', sa.JSON(), nullable=True),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tracing_app_config_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tracing_app_configs',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('tracing_provider', sa.String(length=255), nullable=True),
+        sa.Column('tracing_config', sa.JSON(), nullable=True), 
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tracing_app_config_pkey')
+        )
+    else:
+        op.create_table('tracing_app_configs',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('tracing_provider', sa.String(length=255), nullable=True),
+        sa.Column('tracing_config', sa.JSON(), nullable=True),  
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tracing_app_config_pkey')
+        )
 
     # ### end Alembic commands ###
 

api/migrations/versions/053da0c1d756_add_api_tool_privacy.py

@@ -7,8 +7,15 @@ Create Date: 2024-01-12 06:47:21.656262
 """
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 from sqlalchemy.dialects import postgresql
 
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '053da0c1d756'
 down_revision = '4829e54d2fee'
@@ -18,16 +25,31 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tool_conversation_variables',
-    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('user_id', postgresql.UUID(), nullable=False),
-    sa.Column('tenant_id', postgresql.UUID(), nullable=False),
-    sa.Column('conversation_id', postgresql.UUID(), nullable=False),
-    sa.Column('variables_str', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tool_conversation_variables_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tool_conversation_variables',
+        sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('user_id', postgresql.UUID(), nullable=False),
+        sa.Column('tenant_id', postgresql.UUID(), nullable=False),
+        sa.Column('conversation_id', postgresql.UUID(), nullable=False),
+        sa.Column('variables_str', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_conversation_variables_pkey')
+        )
+    else:
+        op.create_table('tool_conversation_variables',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('user_id', models.types.StringUUID(), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
+        sa.Column('variables_str', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_conversation_variables_pkey')
+        )
+    
     with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
         batch_op.add_column(sa.Column('privacy_policy', sa.String(length=255), nullable=True))
         batch_op.alter_column('icon',

api/migrations/versions/114eed84c228_remove_tool_id_from_model_invoke.py

@@ -9,6 +9,12 @@ import sqlalchemy as sa
 from alembic import op
 from sqlalchemy.dialects import postgresql
 
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '114eed84c228'
 down_revision = 'c71211c8f604'
@@ -26,7 +32,13 @@ def upgrade():
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('tool_model_invokes', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('tool_id', postgresql.UUID(), autoincrement=False, nullable=False))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_model_invokes', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('tool_id', postgresql.UUID(), autoincrement=False, nullable=False))
+    else:
+        with op.batch_alter_table('tool_model_invokes', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('tool_id', models.types.StringUUID(), autoincrement=False, nullable=False))
 
     # ### end Alembic commands ###

api/migrations/versions/161cadc1af8d_add_dataset_permission_tenant_id.py

@@ -8,7 +8,11 @@ Create Date: 2024-07-05 14:30:59.472593
 import sqlalchemy as sa
 from alembic import op
 
-import models as models
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '161cadc1af8d'
@@ -19,9 +23,16 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('dataset_permissions', schema=None) as batch_op:
-        # Step 1: Add column without NOT NULL constraint
-        op.add_column('dataset_permissions', sa.Column('tenant_id', sa.UUID(), nullable=False))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('dataset_permissions', schema=None) as batch_op:
+            # Step 1: Add column without NOT NULL constraint
+            op.add_column('dataset_permissions', sa.Column('tenant_id', sa.UUID(), nullable=False))
+    else:
+        with op.batch_alter_table('dataset_permissions', schema=None) as batch_op:
+            # Step 1: Add column without NOT NULL constraint
+            op.add_column('dataset_permissions', sa.Column('tenant_id', models.types.StringUUID(), nullable=False))
 
     # ### end Alembic commands ###
 

api/migrations/versions/16fa53d9faec_add_provider_model_support.py

@@ -7,8 +7,15 @@ Create Date: 2023-08-06 16:57:51.248337
 """
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 from sqlalchemy.dialects import postgresql
 
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '16fa53d9faec'
 down_revision = '8d2d099ceb74'
@@ -18,44 +25,87 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('provider_models',
-    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', postgresql.UUID(), nullable=False),
-    sa.Column('provider_name', sa.String(length=40), nullable=False),
-    sa.Column('model_name', sa.String(length=40), nullable=False),
-    sa.Column('model_type', sa.String(length=40), nullable=False),
-    sa.Column('encrypted_config', sa.Text(), nullable=True),
-    sa.Column('is_valid', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='provider_model_pkey'),
-    sa.UniqueConstraint('tenant_id', 'provider_name', 'model_name', 'model_type', name='unique_provider_model_name')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('provider_models',
+        sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', postgresql.UUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('model_name', sa.String(length=40), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('encrypted_config', sa.Text(), nullable=True),
+        sa.Column('is_valid', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='provider_model_pkey'),
+        sa.UniqueConstraint('tenant_id', 'provider_name', 'model_name', 'model_type', name='unique_provider_model_name')
+        )
+    else:
+        op.create_table('provider_models',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('model_name', sa.String(length=40), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('encrypted_config', models.types.LongText(), nullable=True),
+        sa.Column('is_valid', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='provider_model_pkey'),
+        sa.UniqueConstraint('tenant_id', 'provider_name', 'model_name', 'model_type', name='unique_provider_model_name')
+        )
+    
     with op.batch_alter_table('provider_models', schema=None) as batch_op:
         batch_op.create_index('provider_model_tenant_id_provider_idx', ['tenant_id', 'provider_name'], unique=False)
 
-    op.create_table('tenant_default_models',
-    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', postgresql.UUID(), nullable=False),
-    sa.Column('provider_name', sa.String(length=40), nullable=False),
-    sa.Column('model_name', sa.String(length=40), nullable=False),
-    sa.Column('model_type', sa.String(length=40), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tenant_default_model_pkey')
-    )
+    if _is_pg(conn):
+        op.create_table('tenant_default_models',
+        sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', postgresql.UUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('model_name', sa.String(length=40), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_default_model_pkey')
+        )
+    else:
+        op.create_table('tenant_default_models',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('model_name', sa.String(length=40), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_default_model_pkey')
+        )
+    
     with op.batch_alter_table('tenant_default_models', schema=None) as batch_op:
         batch_op.create_index('tenant_default_model_tenant_id_provider_type_idx', ['tenant_id', 'provider_name', 'model_type'], unique=False)
 
-    op.create_table('tenant_preferred_model_providers',
-    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', postgresql.UUID(), nullable=False),
-    sa.Column('provider_name', sa.String(length=40), nullable=False),
-    sa.Column('preferred_provider_type', sa.String(length=40), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tenant_preferred_model_provider_pkey')
-    )
+    if _is_pg(conn):
+        op.create_table('tenant_preferred_model_providers',
+        sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', postgresql.UUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('preferred_provider_type', sa.String(length=40), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_preferred_model_provider_pkey')
+        )
+    else:
+        op.create_table('tenant_preferred_model_providers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('preferred_provider_type', sa.String(length=40), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_preferred_model_provider_pkey')
+        )
+    
     with op.batch_alter_table('tenant_preferred_model_providers', schema=None) as batch_op:
         batch_op.create_index('tenant_preferred_model_provider_tenant_provider_idx', ['tenant_id', 'provider_name'], unique=False)
 

api/migrations/versions/17b5ab037c40_add_keyworg_table_storage_type.py

@@ -8,6 +8,10 @@ Create Date: 2024-04-01 09:48:54.232201
 import sqlalchemy as sa
 from alembic import op
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '17b5ab037c40'
 down_revision = 'a8f9b3c45e4a'
@@ -17,9 +21,14 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-
-    with op.batch_alter_table('dataset_keyword_tables', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('data_source_type', sa.String(length=255), server_default=sa.text("'database'::character varying"), nullable=False))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('dataset_keyword_tables', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('data_source_type', sa.String(length=255), server_default=sa.text("'database'::character varying"), nullable=False))
+    else:
+        with op.batch_alter_table('dataset_keyword_tables', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('data_source_type', sa.String(length=255), server_default=sa.text("'database'"), nullable=False))
 
     # ### end Alembic commands ###
 

api/migrations/versions/2024_08_13_0633-63a83fcf12ba_support_conversation_variables.py

@@ -10,6 +10,10 @@ from alembic import op
 
 import models as models
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '63a83fcf12ba'
 down_revision = '1787fbae959a'
@@ -19,21 +23,39 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('workflow__conversation_variables',
-    sa.Column('id', models.types.StringUUID(), nullable=False),
-    sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
-    sa.Column('app_id', models.types.StringUUID(), nullable=False),
-    sa.Column('data', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', 'conversation_id', name=op.f('workflow__conversation_variables_pkey'))
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('workflow__conversation_variables',
+        sa.Column('id', models.types.StringUUID(), nullable=False),
+        sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('data', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', 'conversation_id', name=op.f('workflow__conversation_variables_pkey'))
+        )
+    else:
+        op.create_table('workflow__conversation_variables',
+        sa.Column('id', models.types.StringUUID(), nullable=False),
+        sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('data', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', 'conversation_id', name=op.f('workflow__conversation_variables_pkey'))
+        )
+    
     with op.batch_alter_table('workflow__conversation_variables', schema=None) as batch_op:
         batch_op.create_index(batch_op.f('workflow__conversation_variables_app_id_idx'), ['app_id'], unique=False)
         batch_op.create_index(batch_op.f('workflow__conversation_variables_created_at_idx'), ['created_at'], unique=False)
 
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('conversation_variables', sa.Text(), server_default='{}', nullable=False))
+    if _is_pg(conn):
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('conversation_variables', sa.Text(), server_default='{}', nullable=False))
+    else:
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('conversation_variables', models.types.LongText(), default='{}', nullable=False))
 
     # ### end Alembic commands ###
 

api/migrations/versions/2024_08_15_0956-0251a1c768cc_add_tidb_auth_binding.py

@@ -7,9 +7,14 @@ Create Date: 2024-08-15 09:56:59.012490
 """
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 
 import models as models
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '0251a1c768cc'
 down_revision = 'bbadea11becb'
@@ -19,18 +24,35 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tidb_auth_bindings',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
-    sa.Column('cluster_id', sa.String(length=255), nullable=False),
-    sa.Column('cluster_name', sa.String(length=255), nullable=False),
-    sa.Column('active', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('status', sa.String(length=255), server_default=sa.text("'CREATING'::character varying"), nullable=False),
-    sa.Column('account', sa.String(length=255), nullable=False),
-    sa.Column('password', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tidb_auth_bindings_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tidb_auth_bindings',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
+        sa.Column('cluster_id', sa.String(length=255), nullable=False),
+        sa.Column('cluster_name', sa.String(length=255), nullable=False),
+        sa.Column('active', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('status', sa.String(length=255), server_default=sa.text("'CREATING'::character varying"), nullable=False),
+        sa.Column('account', sa.String(length=255), nullable=False),
+        sa.Column('password', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tidb_auth_bindings_pkey')
+        )
+    else:
+        op.create_table('tidb_auth_bindings',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
+        sa.Column('cluster_id', sa.String(length=255), nullable=False),
+        sa.Column('cluster_name', sa.String(length=255), nullable=False),
+        sa.Column('active', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('status', sa.String(length=255), server_default=sa.text("'CREATING'"), nullable=False),
+        sa.Column('account', sa.String(length=255), nullable=False),
+        sa.Column('password', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tidb_auth_bindings_pkey')
+        )
+    
     with op.batch_alter_table('tidb_auth_bindings', schema=None) as batch_op:
         batch_op.create_index('tidb_auth_bindings_active_idx', ['active'], unique=False)
         batch_op.create_index('tidb_auth_bindings_status_idx', ['status'], unique=False)

api/migrations/versions/2024_09_11_1012-d57ba9ebb251_add_parent_message_id_to_messages.py

@@ -10,6 +10,10 @@ from alembic import op
 
 import models as models
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = 'd57ba9ebb251'
 down_revision = '675b5321501b'
@@ -22,8 +26,14 @@ def upgrade():
     with op.batch_alter_table('messages', schema=None) as batch_op:
         batch_op.add_column(sa.Column('parent_message_id', models.types.StringUUID(), nullable=True))
 
-    # Set parent_message_id for existing messages to uuid_nil() to distinguish them from new messages with actual parent IDs or NULLs
-    op.execute('UPDATE messages SET parent_message_id = uuid_nil() WHERE parent_message_id IS NULL')
+    # Set parent_message_id for existing messages to distinguish them from new messages with actual parent IDs or NULLs
+    conn = op.get_bind()
+    if _is_pg(conn):
+        # PostgreSQL: Use uuid_nil() function
+        op.execute('UPDATE messages SET parent_message_id = uuid_nil() WHERE parent_message_id IS NULL')
+    else:
+        # MySQL: Use a specific UUID value to represent nil
+        op.execute("UPDATE messages SET parent_message_id = '00000000-0000-0000-0000-000000000000' WHERE parent_message_id IS NULL")
 
     # ### end Alembic commands ###
 

api/migrations/versions/2024_09_24_0922-6af6a521a53e_update_retrieval_resource.py

@@ -6,7 +6,11 @@ Create Date: 2024-09-24 09:22:43.570120
 
 """
 from alembic import op
-import models as models
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
@@ -19,30 +23,58 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
-        batch_op.alter_column('document_id',
-               existing_type=sa.UUID(),
-               nullable=True)
-        batch_op.alter_column('data_source_type',
-               existing_type=sa.TEXT(),
-               nullable=True)
-        batch_op.alter_column('segment_id',
-               existing_type=sa.UUID(),
-               nullable=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
+            batch_op.alter_column('document_id',
+                   existing_type=sa.UUID(),
+                   nullable=True)
+            batch_op.alter_column('data_source_type',
+                   existing_type=sa.TEXT(),
+                   nullable=True)
+            batch_op.alter_column('segment_id',
+                   existing_type=sa.UUID(),
+                   nullable=True)
+    else:
+        with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
+            batch_op.alter_column('document_id',
+                   existing_type=models.types.StringUUID(),
+                   nullable=True)
+            batch_op.alter_column('data_source_type',
+                   existing_type=models.types.LongText(),
+                   nullable=True)
+            batch_op.alter_column('segment_id',
+                   existing_type=models.types.StringUUID(),
+                   nullable=True)
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
-        batch_op.alter_column('segment_id',
-               existing_type=sa.UUID(),
-               nullable=False)
-        batch_op.alter_column('data_source_type',
-               existing_type=sa.TEXT(),
-               nullable=False)
-        batch_op.alter_column('document_id',
-               existing_type=sa.UUID(),
-               nullable=False)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
+            batch_op.alter_column('segment_id',
+                   existing_type=sa.UUID(),
+                   nullable=False)
+            batch_op.alter_column('data_source_type',
+                   existing_type=sa.TEXT(),
+                   nullable=False)
+            batch_op.alter_column('document_id',
+                   existing_type=sa.UUID(),
+                   nullable=False)
+    else:
+        with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
+            batch_op.alter_column('segment_id',
+                   existing_type=models.types.StringUUID(),
+                   nullable=False)
+            batch_op.alter_column('data_source_type',
+                   existing_type=models.types.LongText(),
+                   nullable=False)
+            batch_op.alter_column('document_id',
+                   existing_type=models.types.StringUUID(),
+                   nullable=False)
 
     # ### end Alembic commands ###

api/migrations/versions/2024_09_25_0434-33f5fac87f29_external_knowledge_api.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '33f5fac87f29'
@@ -19,34 +24,66 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('external_knowledge_apis',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('description', sa.String(length=255), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('settings', sa.Text(), nullable=True),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='external_knowledge_apis_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('external_knowledge_apis',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.String(length=255), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('settings', sa.Text(), nullable=True),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='external_knowledge_apis_pkey')
+        )
+    else:
+        op.create_table('external_knowledge_apis',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.String(length=255), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('settings', models.types.LongText(), nullable=True),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='external_knowledge_apis_pkey')
+        )
+    
     with op.batch_alter_table('external_knowledge_apis', schema=None) as batch_op:
         batch_op.create_index('external_knowledge_apis_name_idx', ['name'], unique=False)
         batch_op.create_index('external_knowledge_apis_tenant_idx', ['tenant_id'], unique=False)
 
-    op.create_table('external_knowledge_bindings',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('external_knowledge_api_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('external_knowledge_id', sa.Text(), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='external_knowledge_bindings_pkey')
-    )
+    if _is_pg(conn):
+        op.create_table('external_knowledge_bindings',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('external_knowledge_api_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('external_knowledge_id', sa.Text(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='external_knowledge_bindings_pkey')
+        )
+    else:
+        op.create_table('external_knowledge_bindings',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('external_knowledge_api_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('external_knowledge_id', sa.String(length=512), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='external_knowledge_bindings_pkey')
+        )
+    
     with op.batch_alter_table('external_knowledge_bindings', schema=None) as batch_op:
         batch_op.create_index('external_knowledge_bindings_dataset_idx', ['dataset_id'], unique=False)
         batch_op.create_index('external_knowledge_bindings_external_knowledge_api_idx', ['external_knowledge_api_id'], unique=False)

api/migrations/versions/2024_10_10_0516-bbadea11becb_add_name_and_size_to_tool_files.py

@@ -16,6 +16,10 @@ branch_labels = None
 depends_on = None
 
 
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
 def upgrade():
     def _has_name_or_size_column() -> bool:
         # We cannot access the database in offline mode, so assume
@@ -46,14 +50,26 @@ def upgrade():
     if _has_name_or_size_column():
         return
 
-    with op.batch_alter_table("tool_files", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("name", sa.String(), nullable=True))
-        batch_op.add_column(sa.Column("size", sa.Integer(), nullable=True))
-    op.execute("UPDATE tool_files SET name = '' WHERE name IS NULL")
-    op.execute("UPDATE tool_files SET size = -1 WHERE size IS NULL")
-    with op.batch_alter_table("tool_files", schema=None) as batch_op:
-        batch_op.alter_column("name", existing_type=sa.String(), nullable=False)
-        batch_op.alter_column("size", existing_type=sa.Integer(), nullable=False)
+    if _is_pg(conn):
+        # PostgreSQL: Keep original syntax
+        with op.batch_alter_table("tool_files", schema=None) as batch_op:
+            batch_op.add_column(sa.Column("name", sa.String(), nullable=True))
+            batch_op.add_column(sa.Column("size", sa.Integer(), nullable=True))
+        op.execute("UPDATE tool_files SET name = '' WHERE name IS NULL")
+        op.execute("UPDATE tool_files SET size = -1 WHERE size IS NULL")
+        with op.batch_alter_table("tool_files", schema=None) as batch_op:
+            batch_op.alter_column("name", existing_type=sa.String(), nullable=False)
+            batch_op.alter_column("size", existing_type=sa.Integer(), nullable=False)
+    else:
+        # MySQL: Use compatible syntax
+        with op.batch_alter_table("tool_files", schema=None) as batch_op:
+            batch_op.add_column(sa.Column("name", sa.String(length=255), nullable=True))
+            batch_op.add_column(sa.Column("size", sa.Integer(), nullable=True))
+        op.execute("UPDATE tool_files SET name = '' WHERE name IS NULL")
+        op.execute("UPDATE tool_files SET size = -1 WHERE size IS NULL")
+        with op.batch_alter_table("tool_files", schema=None) as batch_op:
+            batch_op.alter_column("name", existing_type=sa.String(length=255), nullable=False)
+            batch_op.alter_column("size", existing_type=sa.Integer(), nullable=False)
     # ### end Alembic commands ###
 
 

api/migrations/versions/2024_10_22_0959-43fa78bc3b7d_add_white_list.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '43fa78bc3b7d'
@@ -19,13 +24,25 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('whitelists',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
-    sa.Column('category', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='whitelists_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('whitelists',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
+        sa.Column('category', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='whitelists_pkey')
+        )
+    else:
+        op.create_table('whitelists',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
+        sa.Column('category', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='whitelists_pkey')
+        )
+    
     with op.batch_alter_table('whitelists', schema=None) as batch_op:
         batch_op.create_index('whitelists_tenant_idx', ['tenant_id'], unique=False)
 

api/migrations/versions/2024_10_28_0720-08ec4f75af5e_add_tenant_plugin_permisisons.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '08ec4f75af5e'
@@ -19,14 +24,26 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('account_plugin_permissions',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('install_permission', sa.String(length=16), server_default='everyone', nullable=False),
-    sa.Column('debug_permission', sa.String(length=16), server_default='noone', nullable=False),
-    sa.PrimaryKeyConstraint('id', name='account_plugin_permission_pkey'),
-    sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('account_plugin_permissions',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('install_permission', sa.String(length=16), server_default='everyone', nullable=False),
+        sa.Column('debug_permission', sa.String(length=16), server_default='noone', nullable=False),
+        sa.PrimaryKeyConstraint('id', name='account_plugin_permission_pkey'),
+        sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin')
+        )
+    else:
+        op.create_table('account_plugin_permissions',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('install_permission', sa.String(length=16), server_default='everyone', nullable=False),
+        sa.Column('debug_permission', sa.String(length=16), server_default='noone', nullable=False),
+        sa.PrimaryKeyConstraint('id', name='account_plugin_permission_pkey'),
+        sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin')
+        )
 
     # ### end Alembic commands ###
 

api/migrations/versions/2024_11_01_0540-f4d7ce70a7ca_update_upload_files_source_url.py

@@ -10,6 +10,10 @@ import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = 'f4d7ce70a7ca'
 down_revision = '93ad8c19c40b'
@@ -19,23 +23,43 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('upload_files', schema=None) as batch_op:
-        batch_op.alter_column('source_url',
-               existing_type=sa.VARCHAR(length=255),
-               type_=sa.TEXT(),
-               existing_nullable=False,
-               existing_server_default=sa.text("''::character varying"))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('upload_files', schema=None) as batch_op:
+            batch_op.alter_column('source_url',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=sa.TEXT(),
+                   existing_nullable=False,
+                   existing_server_default=sa.text("''::character varying"))
+    else:
+        with op.batch_alter_table('upload_files', schema=None) as batch_op:
+            batch_op.alter_column('source_url',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=models.types.LongText(),
+                   existing_nullable=False,
+                   existing_default=sa.text("''"))
 
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('upload_files', schema=None) as batch_op:
-        batch_op.alter_column('source_url',
-               existing_type=sa.TEXT(),
-               type_=sa.VARCHAR(length=255),
-               existing_nullable=False,
-               existing_server_default=sa.text("''::character varying"))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('upload_files', schema=None) as batch_op:
+            batch_op.alter_column('source_url',
+                   existing_type=sa.TEXT(),
+                   type_=sa.VARCHAR(length=255),
+                   existing_nullable=False,
+                   existing_server_default=sa.text("''::character varying"))
+    else:
+        with op.batch_alter_table('upload_files', schema=None) as batch_op:
+            batch_op.alter_column('source_url',
+                   existing_type=models.types.LongText(),
+                   type_=sa.VARCHAR(length=255),
+                   existing_nullable=False,
+                   existing_default=sa.text("''"))
 
     # ### end Alembic commands ###

api/migrations/versions/2024_11_01_0622-d07474999927_update_type_of_custom_disclaimer_to_text.py

@@ -7,6 +7,9 @@ Create Date: 2024-11-01 06:22:27.981398
 """
 from alembic import op
 import models as models
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
@@ -19,49 +22,91 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
+    conn = op.get_bind()
+    
     op.execute("UPDATE recommended_apps SET custom_disclaimer = '' WHERE custom_disclaimer IS NULL")
     op.execute("UPDATE sites SET custom_disclaimer = '' WHERE custom_disclaimer IS NULL")
     op.execute("UPDATE tool_api_providers SET custom_disclaimer = '' WHERE custom_disclaimer IS NULL")
 
-    with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.VARCHAR(length=255),
-               type_=sa.TEXT(),
-               nullable=False)
+    if _is_pg(conn):
+        with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=sa.TEXT(),
+                   nullable=False)
 
-    with op.batch_alter_table('sites', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.VARCHAR(length=255),
-               type_=sa.TEXT(),
-               nullable=False)
+        with op.batch_alter_table('sites', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=sa.TEXT(),
+                   nullable=False)
 
-    with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.VARCHAR(length=255),
-               type_=sa.TEXT(),
-               nullable=False)
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=sa.TEXT(),
+                   nullable=False)
+    else:
+        with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=models.types.LongText(),
+                   nullable=False)
+
+        with op.batch_alter_table('sites', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=models.types.LongText(),
+                   nullable=False)
+
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=models.types.LongText(),
+                   nullable=False)
 
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.TEXT(),
-               type_=sa.VARCHAR(length=255),
-               nullable=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.TEXT(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
 
-    with op.batch_alter_table('sites', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.TEXT(),
-               type_=sa.VARCHAR(length=255),
-               nullable=True)
+        with op.batch_alter_table('sites', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.TEXT(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
 
-    with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.TEXT(),
-               type_=sa.VARCHAR(length=255),
-               nullable=True)
+        with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.TEXT(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
+    else:
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=models.types.LongText(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
+
+        with op.batch_alter_table('sites', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=models.types.LongText(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
+
+        with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=models.types.LongText(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
 
     # ### end Alembic commands ###

api/migrations/versions/2024_11_01_0623-09a8d1878d9b_update_workflows_graph_features_and_.py

@@ -10,6 +10,10 @@ import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '09a8d1878d9b'
 down_revision = 'd07474999927'
@@ -19,55 +23,103 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('conversations', schema=None) as batch_op:
-        batch_op.alter_column('inputs',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=False)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('conversations', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   nullable=False)
 
-    with op.batch_alter_table('messages', schema=None) as batch_op:
-        batch_op.alter_column('inputs',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=False)
+        with op.batch_alter_table('messages', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   nullable=False)
+    else:
+        with op.batch_alter_table('conversations', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=sa.JSON(),
+                   nullable=False)
+
+        with op.batch_alter_table('messages', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=sa.JSON(),
+                   nullable=False)
 
     op.execute("UPDATE workflows SET updated_at = created_at WHERE updated_at IS NULL")
     op.execute("UPDATE workflows SET graph = '' WHERE graph IS NULL")
     op.execute("UPDATE workflows SET features = '' WHERE features IS NULL")
-
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.alter_column('graph',
-               existing_type=sa.TEXT(),
-               nullable=False)
-        batch_op.alter_column('features',
-               existing_type=sa.TEXT(),
-               nullable=False)
-        batch_op.alter_column('updated_at',
-               existing_type=postgresql.TIMESTAMP(),
-               nullable=False)
-
+    if _is_pg(conn):
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.alter_column('graph',
+                existing_type=sa.TEXT(),
+                nullable=False)
+            batch_op.alter_column('features',
+                existing_type=sa.TEXT(),
+                nullable=False)
+            batch_op.alter_column('updated_at',
+                existing_type=postgresql.TIMESTAMP(),
+                nullable=False)
+    else:
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.alter_column('graph',
+                existing_type=models.types.LongText(),
+                nullable=False)
+            batch_op.alter_column('features',
+                existing_type=models.types.LongText(),
+                nullable=False)
+            batch_op.alter_column('updated_at',
+                existing_type=sa.TIMESTAMP(),
+                nullable=False)
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.alter_column('updated_at',
-               existing_type=postgresql.TIMESTAMP(),
-               nullable=True)
-        batch_op.alter_column('features',
-               existing_type=sa.TEXT(),
-               nullable=True)
-        batch_op.alter_column('graph',
-               existing_type=sa.TEXT(),
-               nullable=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.alter_column('updated_at',
+                existing_type=postgresql.TIMESTAMP(),
+                nullable=True)
+            batch_op.alter_column('features',
+                existing_type=sa.TEXT(),
+                nullable=True)
+            batch_op.alter_column('graph',
+                existing_type=sa.TEXT(),
+                nullable=True)
+    else:
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.alter_column('updated_at',
+                existing_type=sa.TIMESTAMP(),
+                nullable=True)
+            batch_op.alter_column('features',
+                existing_type=models.types.LongText(),
+                nullable=True)
+            batch_op.alter_column('graph',
+                existing_type=models.types.LongText(),
+                nullable=True)
 
-    with op.batch_alter_table('messages', schema=None) as batch_op:
-        batch_op.alter_column('inputs',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=True)
+    if _is_pg(conn):
+        with op.batch_alter_table('messages', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   nullable=True)
 
-    with op.batch_alter_table('conversations', schema=None) as batch_op:
-        batch_op.alter_column('inputs',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=True)
+        with op.batch_alter_table('conversations', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   nullable=True)
+    else:
+        with op.batch_alter_table('messages', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=sa.JSON(),
+                   nullable=True)
+
+        with op.batch_alter_table('conversations', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=sa.JSON(),
+                   nullable=True)
 
     # ### end Alembic commands ###

api/migrations/versions/2024_11_22_0701-e19037032219_parent_child_index.py

@@ -8,6 +8,11 @@ Create Date: 2024-11-22 07:01:17.550037
 from alembic import op
 import models as models
 import sqlalchemy as sa
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 
 # revision identifiers, used by Alembic.
@@ -19,27 +24,53 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('child_chunks',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('document_id', models.types.StringUUID(), nullable=False),
-    sa.Column('segment_id', models.types.StringUUID(), nullable=False),
-    sa.Column('position', sa.Integer(), nullable=False),
-    sa.Column('content', sa.Text(), nullable=False),
-    sa.Column('word_count', sa.Integer(), nullable=False),
-    sa.Column('index_node_id', sa.String(length=255), nullable=True),
-    sa.Column('index_node_hash', sa.String(length=255), nullable=True),
-    sa.Column('type', sa.String(length=255), server_default=sa.text("'automatic'::character varying"), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('indexing_at', sa.DateTime(), nullable=True),
-    sa.Column('completed_at', sa.DateTime(), nullable=True),
-    sa.Column('error', sa.Text(), nullable=True),
-    sa.PrimaryKeyConstraint('id', name='child_chunk_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('child_chunks',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('segment_id', models.types.StringUUID(), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('content', sa.Text(), nullable=False),
+        sa.Column('word_count', sa.Integer(), nullable=False),
+        sa.Column('index_node_id', sa.String(length=255), nullable=True),
+        sa.Column('index_node_hash', sa.String(length=255), nullable=True),
+        sa.Column('type', sa.String(length=255), server_default=sa.text("'automatic'::character varying"), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('indexing_at', sa.DateTime(), nullable=True),
+        sa.Column('completed_at', sa.DateTime(), nullable=True),
+        sa.Column('error', sa.Text(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='child_chunk_pkey')
+        )
+    else:
+        op.create_table('child_chunks',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('segment_id', models.types.StringUUID(), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('content', models.types.LongText(), nullable=False),
+        sa.Column('word_count', sa.Integer(), nullable=False),
+        sa.Column('index_node_id', sa.String(length=255), nullable=True),
+        sa.Column('index_node_hash', sa.String(length=255), nullable=True),
+        sa.Column('type', sa.String(length=255), server_default=sa.text("'automatic'"), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('indexing_at', sa.DateTime(), nullable=True),
+        sa.Column('completed_at', sa.DateTime(), nullable=True),
+        sa.Column('error', models.types.LongText(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='child_chunk_pkey')
+        )
+    
     with op.batch_alter_table('child_chunks', schema=None) as batch_op:
         batch_op.create_index('child_chunk_dataset_id_idx', ['tenant_id', 'dataset_id', 'document_id', 'segment_id', 'index_node_id'], unique=False)
 

api/migrations/versions/2024_12_19_1746-11b07f66c737_remove_unused_tool_providers.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '11b07f66c737'
@@ -25,15 +30,30 @@ def upgrade():
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tool_providers',
-    sa.Column('id', sa.UUID(), server_default=sa.text('uuid_generate_v4()'), autoincrement=False, nullable=False),
-    sa.Column('tenant_id', sa.UUID(), autoincrement=False, nullable=False),
-    sa.Column('tool_name', sa.VARCHAR(length=40), autoincrement=False, nullable=False),
-    sa.Column('encrypted_credentials', sa.TEXT(), autoincrement=False, nullable=True),
-    sa.Column('is_enabled', sa.BOOLEAN(), server_default=sa.text('false'), autoincrement=False, nullable=False),
-    sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), autoincrement=False, nullable=False),
-    sa.Column('updated_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), autoincrement=False, nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tool_provider_pkey'),
-    sa.UniqueConstraint('tenant_id', 'tool_name', name='unique_tool_provider_tool_name')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tool_providers',
+        sa.Column('id', sa.UUID(), server_default=sa.text('uuid_generate_v4()'), autoincrement=False, nullable=False),
+        sa.Column('tenant_id', sa.UUID(), autoincrement=False, nullable=False),
+        sa.Column('tool_name', sa.VARCHAR(length=40), autoincrement=False, nullable=False),
+        sa.Column('encrypted_credentials', sa.TEXT(), autoincrement=False, nullable=True),
+        sa.Column('is_enabled', sa.BOOLEAN(), server_default=sa.text('false'), autoincrement=False, nullable=False),
+        sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), autoincrement=False, nullable=False),
+        sa.Column('updated_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), autoincrement=False, nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'tool_name', name='unique_tool_provider_tool_name')
+        )
+    else:
+        op.create_table('tool_providers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), autoincrement=False, nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), autoincrement=False, nullable=False),
+        sa.Column('tool_name', sa.VARCHAR(length=40), autoincrement=False, nullable=False),
+        sa.Column('encrypted_credentials', models.types.LongText(), autoincrement=False, nullable=True),
+        sa.Column('is_enabled', sa.BOOLEAN(), server_default=sa.text('false'), autoincrement=False, nullable=False),
+        sa.Column('created_at', sa.TIMESTAMP(), server_default=sa.func.current_timestamp(), autoincrement=False, nullable=False),
+        sa.Column('updated_at', sa.TIMESTAMP(), server_default=sa.func.current_timestamp(), autoincrement=False, nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'tool_name', name='unique_tool_provider_tool_name')
+        )
     # ### end Alembic commands ###

api/migrations/versions/2024_12_25_1137-923752d42eb6_add_auto_disabled_dataset_logs.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '923752d42eb6'
@@ -19,15 +24,29 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('dataset_auto_disable_logs',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('document_id', models.types.StringUUID(), nullable=False),
-    sa.Column('notified', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='dataset_auto_disable_log_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('dataset_auto_disable_logs',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('notified', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='dataset_auto_disable_log_pkey')
+        )
+    else:
+        op.create_table('dataset_auto_disable_logs',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('notified', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='dataset_auto_disable_log_pkey')
+        )
+    
     with op.batch_alter_table('dataset_auto_disable_logs', schema=None) as batch_op:
         batch_op.create_index('dataset_auto_disable_log_created_atx', ['created_at'], unique=False)
         batch_op.create_index('dataset_auto_disable_log_dataset_idx', ['dataset_id'], unique=False)

api/migrations/versions/2025_01_14_0617-f051706725cc_add_rate_limit_logs.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = 'f051706725cc'
@@ -19,14 +24,27 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('rate_limit_logs',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('subscription_plan', sa.String(length=255), nullable=False),
-    sa.Column('operation', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='rate_limit_log_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('rate_limit_logs',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('subscription_plan', sa.String(length=255), nullable=False),
+        sa.Column('operation', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='rate_limit_log_pkey')
+        )
+    else:
+        op.create_table('rate_limit_logs',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('subscription_plan', sa.String(length=255), nullable=False),
+        sa.Column('operation', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='rate_limit_log_pkey')
+        )
+    
     with op.batch_alter_table('rate_limit_logs', schema=None) as batch_op:
         batch_op.create_index('rate_limit_log_operation_idx', ['operation'], unique=False)
         batch_op.create_index('rate_limit_log_tenant_idx', ['tenant_id'], unique=False)

api/migrations/versions/2025_02_27_0917-d20049ed0af6_add_metadata_function.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = 'd20049ed0af6'
@@ -19,34 +24,68 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('dataset_metadata_bindings',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('metadata_id', models.types.StringUUID(), nullable=False),
-    sa.Column('document_id', models.types.StringUUID(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='dataset_metadata_binding_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        # PostgreSQL: Keep original syntax
+        op.create_table('dataset_metadata_bindings',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('metadata_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='dataset_metadata_binding_pkey')
+        )
+    else:
+        # MySQL: Use compatible syntax
+        op.create_table('dataset_metadata_bindings',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('metadata_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='dataset_metadata_binding_pkey')
+        )
+    
     with op.batch_alter_table('dataset_metadata_bindings', schema=None) as batch_op:
         batch_op.create_index('dataset_metadata_binding_dataset_idx', ['dataset_id'], unique=False)
         batch_op.create_index('dataset_metadata_binding_document_idx', ['document_id'], unique=False)
         batch_op.create_index('dataset_metadata_binding_metadata_idx', ['metadata_id'], unique=False)
         batch_op.create_index('dataset_metadata_binding_tenant_idx', ['tenant_id'], unique=False)
 
-    op.create_table('dataset_metadatas',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('type', sa.String(length=255), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.PrimaryKeyConstraint('id', name='dataset_metadata_pkey')
-    )
+    if _is_pg(conn):
+        # PostgreSQL: Keep original syntax
+        op.create_table('dataset_metadatas',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('type', sa.String(length=255), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='dataset_metadata_pkey')
+        )
+    else:
+        # MySQL: Use compatible syntax
+        op.create_table('dataset_metadatas',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('type', sa.String(length=255), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='dataset_metadata_pkey')
+        )
+    
     with op.batch_alter_table('dataset_metadatas', schema=None) as batch_op:
         batch_op.create_index('dataset_metadata_dataset_idx', ['dataset_id'], unique=False)
         batch_op.create_index('dataset_metadata_tenant_idx', ['tenant_id'], unique=False)
@@ -54,23 +93,31 @@ def upgrade():
     with op.batch_alter_table('datasets', schema=None) as batch_op:
         batch_op.add_column(sa.Column('built_in_field_enabled', sa.Boolean(), server_default=sa.text('false'), nullable=False))
 
-    with op.batch_alter_table('documents', schema=None) as batch_op:
-        batch_op.alter_column('doc_metadata',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               type_=postgresql.JSONB(astext_type=sa.Text()),
-               existing_nullable=True)
-        batch_op.create_index('document_metadata_idx', ['doc_metadata'], unique=False, postgresql_using='gin')
+    if _is_pg(conn):
+        with op.batch_alter_table('documents', schema=None) as batch_op:
+            batch_op.alter_column('doc_metadata',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   type_=postgresql.JSONB(astext_type=sa.Text()),
+                   existing_nullable=True)
+            batch_op.create_index('document_metadata_idx', ['doc_metadata'], unique=False, postgresql_using='gin')
+    else:
+        pass
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('documents', schema=None) as batch_op:
-        batch_op.drop_index('document_metadata_idx', postgresql_using='gin')
-        batch_op.alter_column('doc_metadata',
-               existing_type=postgresql.JSONB(astext_type=sa.Text()),
-               type_=postgresql.JSON(astext_type=sa.Text()),
-               existing_nullable=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('documents', schema=None) as batch_op:
+            batch_op.drop_index('document_metadata_idx', postgresql_using='gin')
+            batch_op.alter_column('doc_metadata',
+                   existing_type=postgresql.JSONB(astext_type=sa.Text()),
+                   type_=postgresql.JSON(astext_type=sa.Text()),
+                   existing_nullable=True)
+    else:
+        pass
 
     with op.batch_alter_table('datasets', schema=None) as batch_op:
         batch_op.drop_column('built_in_field_enabled')

api/migrations/versions/2025_03_03_1436-ee79d9b1c156_add_marked_name_and_marked_comment_in_.py

@@ -17,10 +17,23 @@ branch_labels = None
 depends_on = None
 
 
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
 def upgrade():
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('marked_name', sa.String(), nullable=False, server_default=''))
-        batch_op.add_column(sa.Column('marked_comment', sa.String(), nullable=False, server_default=''))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        # PostgreSQL: Keep original syntax
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('marked_name', sa.String(), nullable=False, server_default=''))
+            batch_op.add_column(sa.Column('marked_comment', sa.String(), nullable=False, server_default=''))
+    else:
+        # MySQL: Use compatible syntax
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('marked_name', sa.String(length=255), nullable=False, server_default=''))
+            batch_op.add_column(sa.Column('marked_comment', sa.String(length=255), nullable=False, server_default=''))
 
 
 def downgrade():

api/migrations/versions/2025_05_15_1531-2adcbe1f5dfb_add_workflowdraftvariable_model.py

@@ -8,9 +8,14 @@ Create Date: 2025-05-15 15:31:03.128680
 
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 
 import models as models
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = "2adcbe1f5dfb"
 down_revision = "d28f2004b072"
@@ -20,24 +25,46 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table(
-        "workflow_draft_variables",
-        sa.Column("id", models.types.StringUUID(), server_default=sa.text("uuid_generate_v4()"), nullable=False),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.Column("app_id", models.types.StringUUID(), nullable=False),
-        sa.Column("last_edited_at", sa.DateTime(), nullable=True),
-        sa.Column("node_id", sa.String(length=255), nullable=False),
-        sa.Column("name", sa.String(length=255), nullable=False),
-        sa.Column("description", sa.String(length=255), nullable=False),
-        sa.Column("selector", sa.String(length=255), nullable=False),
-        sa.Column("value_type", sa.String(length=20), nullable=False),
-        sa.Column("value", sa.Text(), nullable=False),
-        sa.Column("visible", sa.Boolean(), nullable=False),
-        sa.Column("editable", sa.Boolean(), nullable=False),
-        sa.PrimaryKeyConstraint("id", name=op.f("workflow_draft_variables_pkey")),
-        sa.UniqueConstraint("app_id", "node_id", "name", name=op.f("workflow_draft_variables_app_id_key")),
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table(
+            "workflow_draft_variables",
+            sa.Column("id", models.types.StringUUID(), server_default=sa.text("uuid_generate_v4()"), nullable=False),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.Column("app_id", models.types.StringUUID(), nullable=False),
+            sa.Column("last_edited_at", sa.DateTime(), nullable=True),
+            sa.Column("node_id", sa.String(length=255), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("description", sa.String(length=255), nullable=False),
+            sa.Column("selector", sa.String(length=255), nullable=False),
+            sa.Column("value_type", sa.String(length=20), nullable=False),
+            sa.Column("value", sa.Text(), nullable=False),
+            sa.Column("visible", sa.Boolean(), nullable=False),
+            sa.Column("editable", sa.Boolean(), nullable=False),
+            sa.PrimaryKeyConstraint("id", name=op.f("workflow_draft_variables_pkey")),
+            sa.UniqueConstraint("app_id", "node_id", "name", name=op.f("workflow_draft_variables_app_id_key")),
+        )
+    else:
+        op.create_table(
+            "workflow_draft_variables",
+            sa.Column("id", models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.Column("app_id", models.types.StringUUID(), nullable=False),
+            sa.Column("last_edited_at", sa.DateTime(), nullable=True),
+            sa.Column("node_id", sa.String(length=255), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("description", sa.String(length=255), nullable=False),
+            sa.Column("selector", sa.String(length=255), nullable=False),
+            sa.Column("value_type", sa.String(length=20), nullable=False),
+            sa.Column("value", models.types.LongText(), nullable=False),
+            sa.Column("visible", sa.Boolean(), nullable=False),
+            sa.Column("editable", sa.Boolean(), nullable=False),
+            sa.PrimaryKeyConstraint("id", name=op.f("workflow_draft_variables_pkey")),
+            sa.UniqueConstraint("app_id", "node_id", "name", name=op.f("workflow_draft_variables_app_id_key")),
+        )
 
     # ### end Alembic commands ###
 

api/migrations/versions/2025_06_06_1424-4474872b0ee6_workflow_draft_varaibles_add_node_execution_id.py

@@ -7,6 +7,10 @@ Create Date: 2025-06-06 14:24:44.213018
 """
 from alembic import op
 import models as models
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 
 
@@ -18,19 +22,30 @@ depends_on = None
 
 
 def upgrade():
-    # `CREATE INDEX CONCURRENTLY` cannot run within a transaction, so use the `autocommit_block`
-    # context manager to wrap the index creation statement.
-    # Reference:
-    #
-    # - https://www.postgresql.org/docs/current/sql-createindex.html#:~:text=Another%20difference%20is,CREATE%20INDEX%20CONCURRENTLY%20cannot.
-    # - https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.migration.MigrationContext.autocommit_block
-    with op.get_context().autocommit_block():
+    # ### commands auto generated by Alembic - please adjust! ###
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        # `CREATE INDEX CONCURRENTLY` cannot run within a transaction, so use the `autocommit_block`
+        # context manager to wrap the index creation statement.
+        # Reference:
+        #
+        # - https://www.postgresql.org/docs/current/sql-createindex.html#:~:text=Another%20difference%20is,CREATE%20INDEX%20CONCURRENTLY%20cannot.
+        # - https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.migration.MigrationContext.autocommit_block
+        with op.get_context().autocommit_block():
+            op.create_index(
+                op.f('workflow_node_executions_tenant_id_idx'),
+                "workflow_node_executions",
+                ['tenant_id', 'workflow_id', 'node_id', sa.literal_column('created_at DESC')],
+                unique=False,
+                postgresql_concurrently=True,
+            )
+    else:
         op.create_index(
             op.f('workflow_node_executions_tenant_id_idx'),
             "workflow_node_executions",
             ['tenant_id', 'workflow_id', 'node_id', sa.literal_column('created_at DESC')],
             unique=False,
-            postgresql_concurrently=True,
         )
 
     with op.batch_alter_table('workflow_draft_variables', schema=None) as batch_op:
@@ -51,8 +66,13 @@ def downgrade():
     # Reference:
     #
     # https://www.postgresql.org/docs/current/sql-createindex.html#:~:text=Another%20difference%20is,CREATE%20INDEX%20CONCURRENTLY%20cannot.
-    with op.get_context().autocommit_block():
-        op.drop_index(op.f('workflow_node_executions_tenant_id_idx'), postgresql_concurrently=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.get_context().autocommit_block():
+            op.drop_index(op.f('workflow_node_executions_tenant_id_idx'), postgresql_concurrently=True)
+    else:
+        op.drop_index(op.f('workflow_node_executions_tenant_id_idx'))
 
     with op.batch_alter_table('workflow_draft_variables', schema=None) as batch_op:
         batch_op.drop_column('node_execution_id')

api/migrations/versions/2025_06_25_0936-58eb7bdb93fe_add_mcp_server_tool_and_app_server.py

@@ -8,6 +8,11 @@ Create Date: 2025-06-25 09:36:07.510570
 from alembic import op
 import models as models
 import sqlalchemy as sa
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 
 # revision identifiers, used by Alembic.
@@ -19,40 +24,80 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('app_mcp_servers',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('app_id', models.types.StringUUID(), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('description', sa.String(length=255), nullable=False),
-    sa.Column('server_code', sa.String(length=255), nullable=False),
-    sa.Column('status', sa.String(length=255), server_default=sa.text("'normal'::character varying"), nullable=False),
-    sa.Column('parameters', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='app_mcp_server_pkey'),
-    sa.UniqueConstraint('tenant_id', 'app_id', name='unique_app_mcp_server_tenant_app_id'),
-    sa.UniqueConstraint('server_code', name='unique_app_mcp_server_server_code')
-    )
-    op.create_table('tool_mcp_providers',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('name', sa.String(length=40), nullable=False),
-    sa.Column('server_identifier', sa.String(length=24), nullable=False),
-    sa.Column('server_url', sa.Text(), nullable=False),
-    sa.Column('server_url_hash', sa.String(length=64), nullable=False),
-    sa.Column('icon', sa.String(length=255), nullable=True),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('user_id', models.types.StringUUID(), nullable=False),
-    sa.Column('encrypted_credentials', sa.Text(), nullable=True),
-    sa.Column('authed', sa.Boolean(), nullable=False),
-    sa.Column('tools', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tool_mcp_provider_pkey'),
-    sa.UniqueConstraint('tenant_id', 'name', name='unique_mcp_provider_name'),
-    sa.UniqueConstraint('tenant_id', 'server_identifier', name='unique_mcp_provider_server_identifier'),
-    sa.UniqueConstraint('tenant_id', 'server_url_hash', name='unique_mcp_provider_server_url')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('app_mcp_servers',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.String(length=255), nullable=False),
+        sa.Column('server_code', sa.String(length=255), nullable=False),
+        sa.Column('status', sa.String(length=255), server_default=sa.text("'normal'::character varying"), nullable=False),
+        sa.Column('parameters', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='app_mcp_server_pkey'),
+        sa.UniqueConstraint('tenant_id', 'app_id', name='unique_app_mcp_server_tenant_app_id'),
+        sa.UniqueConstraint('server_code', name='unique_app_mcp_server_server_code')
+        )
+    else:
+        op.create_table('app_mcp_servers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.String(length=255), nullable=False),
+        sa.Column('server_code', sa.String(length=255), nullable=False),
+        sa.Column('status', sa.String(length=255), server_default=sa.text("'normal'"), nullable=False),
+        sa.Column('parameters', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='app_mcp_server_pkey'),
+        sa.UniqueConstraint('tenant_id', 'app_id', name='unique_app_mcp_server_tenant_app_id'),
+        sa.UniqueConstraint('server_code', name='unique_app_mcp_server_server_code')
+        )
+    if _is_pg(conn):
+        op.create_table('tool_mcp_providers',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('name', sa.String(length=40), nullable=False),
+        sa.Column('server_identifier', sa.String(length=24), nullable=False),
+        sa.Column('server_url', sa.Text(), nullable=False),
+        sa.Column('server_url_hash', sa.String(length=64), nullable=False),
+        sa.Column('icon', sa.String(length=255), nullable=True),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('user_id', models.types.StringUUID(), nullable=False),
+        sa.Column('encrypted_credentials', sa.Text(), nullable=True),
+        sa.Column('authed', sa.Boolean(), nullable=False),
+        sa.Column('tools', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_mcp_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'name', name='unique_mcp_provider_name'),
+        sa.UniqueConstraint('tenant_id', 'server_identifier', name='unique_mcp_provider_server_identifier'),
+        sa.UniqueConstraint('tenant_id', 'server_url_hash', name='unique_mcp_provider_server_url')
+        )
+    else:
+        op.create_table('tool_mcp_providers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('name', sa.String(length=40), nullable=False),
+        sa.Column('server_identifier', sa.String(length=24), nullable=False),
+        sa.Column('server_url', models.types.LongText(), nullable=False),
+        sa.Column('server_url_hash', sa.String(length=64), nullable=False),
+        sa.Column('icon', sa.String(length=255), nullable=True),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('user_id', models.types.StringUUID(), nullable=False),
+        sa.Column('encrypted_credentials', models.types.LongText(), nullable=True),
+        sa.Column('authed', sa.Boolean(), nullable=False),
+        sa.Column('tools', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_mcp_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'name', name='unique_mcp_provider_name'),
+        sa.UniqueConstraint('tenant_id', 'server_identifier', name='unique_mcp_provider_server_identifier'),
+        sa.UniqueConstraint('tenant_id', 'server_url_hash', name='unique_mcp_provider_server_url')
+        )
 
     # ### end Alembic commands ###
 

api/migrations/versions/2025_07_02_2332-1c9ba48be8e4_add_uuidv7_function_in_sql.py

@@ -27,6 +27,10 @@ import models as models
 import sqlalchemy as sa
 
 
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
 # revision identifiers, used by Alembic.
 revision = '1c9ba48be8e4'
 down_revision = '58eb7bdb93fe'
@@ -40,7 +44,11 @@ def upgrade():
     # The ability to specify source timestamp has been removed because its type signature is incompatible with
     # PostgreSQL 18's `uuidv7` function. This capability is rarely needed in practice, as IDs can be
     # generated and controlled within the application layer.
-    op.execute(sa.text(r"""
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        # PostgreSQL: Create uuidv7 functions
+        op.execute(sa.text(r"""
 /* Main function to generate a uuidv7 value with millisecond precision */
 CREATE FUNCTION uuidv7() RETURNS uuid
 AS
@@ -63,7 +71,7 @@ COMMENT ON FUNCTION uuidv7 IS
     'Generate a uuid-v7 value with a 48-bit timestamp (millisecond precision) and 74 bits of randomness';
 """))
 
-    op.execute(sa.text(r"""
+        op.execute(sa.text(r"""
 CREATE FUNCTION uuidv7_boundary(timestamptz) RETURNS uuid
 AS
 $$
@@ -79,8 +87,15 @@ COMMENT ON FUNCTION uuidv7_boundary(timestamptz) IS
     'Generate a non-random uuidv7 with the given timestamp (first 48 bits) and all random bits to 0. As the smallest possible uuidv7 for that timestamp, it may be used as a boundary for partitions.';
 """
 ))
+    else:
+        pass
 
 
 def downgrade():
-    op.execute(sa.text("DROP FUNCTION uuidv7"))
-    op.execute(sa.text("DROP FUNCTION uuidv7_boundary"))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.execute(sa.text("DROP FUNCTION uuidv7"))
+        op.execute(sa.text("DROP FUNCTION uuidv7_boundary"))
+    else:
+        pass

api/migrations/versions/2025_07_04_1705-71f5020c6470_tool_oauth.py

@@ -8,6 +8,11 @@ Create Date: 2025-06-24 17:05:43.118647
 from alembic import op
 import models as models
 import sqlalchemy as sa
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 
 # revision identifiers, used by Alembic.
@@ -19,31 +24,63 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tool_oauth_system_clients',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('plugin_id', sa.String(length=512), nullable=False),
-    sa.Column('provider', sa.String(length=255), nullable=False),
-    sa.Column('encrypted_oauth_params', sa.Text(), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tool_oauth_system_client_pkey'),
-    sa.UniqueConstraint('plugin_id', 'provider', name='tool_oauth_system_client_plugin_id_provider_idx')
-    )
-    op.create_table('tool_oauth_tenant_clients',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('plugin_id', sa.String(length=512), nullable=False),
-    sa.Column('provider', sa.String(length=255), nullable=False),
-    sa.Column('enabled', sa.Boolean(), server_default=sa.text('true'), nullable=False),
-    sa.Column('encrypted_oauth_params', sa.Text(), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tool_oauth_tenant_client_pkey'),
-    sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', name='unique_tool_oauth_tenant_client')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tool_oauth_system_clients',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('plugin_id', sa.String(length=512), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('encrypted_oauth_params', sa.Text(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_oauth_system_client_pkey'),
+        sa.UniqueConstraint('plugin_id', 'provider', name='tool_oauth_system_client_plugin_id_provider_idx')
+        )
+    else:
+        op.create_table('tool_oauth_system_clients',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('plugin_id', sa.String(length=512), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('encrypted_oauth_params', models.types.LongText(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_oauth_system_client_pkey'),
+        sa.UniqueConstraint('plugin_id', 'provider', name='tool_oauth_system_client_plugin_id_provider_idx')
+        )
+    if _is_pg(conn):
+        op.create_table('tool_oauth_tenant_clients',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('plugin_id', sa.String(length=512), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('enabled', sa.Boolean(), server_default=sa.text('true'), nullable=False),
+        sa.Column('encrypted_oauth_params', sa.Text(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_oauth_tenant_client_pkey'),
+        sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', name='unique_tool_oauth_tenant_client')
+        )
+    else:
+        op.create_table('tool_oauth_tenant_clients',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('plugin_id', sa.String(length=255), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('enabled', sa.Boolean(), server_default=sa.text('true'), nullable=False),
+        sa.Column('encrypted_oauth_params', models.types.LongText(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_oauth_tenant_client_pkey'),
+        sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', name='unique_tool_oauth_tenant_client')
+        )
 
-    with op.batch_alter_table('tool_builtin_providers', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('name', sa.String(length=256), server_default=sa.text("'API KEY 1'::character varying"), nullable=False))
-        batch_op.add_column(sa.Column('is_default', sa.Boolean(), server_default=sa.text('false'), nullable=False))
-        batch_op.add_column(sa.Column('credential_type', sa.String(length=32), server_default=sa.text("'api-key'::character varying"), nullable=False))
-        batch_op.drop_constraint(batch_op.f('unique_builtin_tool_provider'), type_='unique')
-        batch_op.create_unique_constraint(batch_op.f('unique_builtin_tool_provider'), ['tenant_id', 'provider', 'name'])
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_builtin_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('name', sa.String(length=256), server_default=sa.text("'API KEY 1'::character varying"), nullable=False))
+            batch_op.add_column(sa.Column('is_default', sa.Boolean(), server_default=sa.text('false'), nullable=False))
+            batch_op.add_column(sa.Column('credential_type', sa.String(length=32), server_default=sa.text("'api-key'::character varying"), nullable=False))
+            batch_op.drop_constraint(batch_op.f('unique_builtin_tool_provider'), type_='unique')
+            batch_op.create_unique_constraint(batch_op.f('unique_builtin_tool_provider'), ['tenant_id', 'provider', 'name'])
+    else:
+        with op.batch_alter_table('tool_builtin_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('name', sa.String(length=256), server_default=sa.text("'API KEY 1'"), nullable=False))
+            batch_op.add_column(sa.Column('is_default', sa.Boolean(), server_default=sa.text('false'), nullable=False))
+            batch_op.add_column(sa.Column('credential_type', sa.String(length=32), server_default=sa.text("'api-key'"), nullable=False))
+            batch_op.drop_constraint(batch_op.f('unique_builtin_tool_provider'), type_='unique')
+            batch_op.create_unique_constraint(batch_op.f('unique_builtin_tool_provider'), ['tenant_id', 'provider', 'name'])
 
     # ### end Alembic commands ###
 

api/migrations/versions/2025_07_23_1508-8bcc02c9bd07_add_tenant_plugin_autoupgrade_table.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '8bcc02c9bd07'
@@ -19,19 +24,36 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tenant_plugin_auto_upgrade_strategies',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('strategy_setting', sa.String(length=16), server_default='fix_only', nullable=False),
-    sa.Column('upgrade_time_of_day', sa.Integer(), nullable=False),
-    sa.Column('upgrade_mode', sa.String(length=16), server_default='exclude', nullable=False),
-    sa.Column('exclude_plugins', sa.ARRAY(sa.String(length=255)), nullable=False),
-    sa.Column('include_plugins', sa.ARRAY(sa.String(length=255)), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tenant_plugin_auto_upgrade_strategy_pkey'),
-    sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin_auto_upgrade_strategy')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tenant_plugin_auto_upgrade_strategies',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('strategy_setting', sa.String(length=16), server_default='fix_only', nullable=False),
+        sa.Column('upgrade_time_of_day', sa.Integer(), nullable=False),
+        sa.Column('upgrade_mode', sa.String(length=16), server_default='exclude', nullable=False),
+        sa.Column('exclude_plugins', sa.ARRAY(sa.String(length=255)), nullable=False),
+        sa.Column('include_plugins', sa.ARRAY(sa.String(length=255)), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_plugin_auto_upgrade_strategy_pkey'),
+        sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin_auto_upgrade_strategy')
+        )
+    else:
+        op.create_table('tenant_plugin_auto_upgrade_strategies',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('strategy_setting', sa.String(length=16), server_default='fix_only', nullable=False),
+        sa.Column('upgrade_time_of_day', sa.Integer(), nullable=False),
+        sa.Column('upgrade_mode', sa.String(length=16), server_default='exclude', nullable=False),
+        sa.Column('exclude_plugins', sa.JSON(), nullable=False),
+        sa.Column('include_plugins', sa.JSON(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_plugin_auto_upgrade_strategy_pkey'),
+        sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin_auto_upgrade_strategy')
+        )
     # ### end Alembic commands ###
 
 

api/migrations/versions/2025_07_24_1450-532b3f888abf_manual_dataset_field_update.py

@@ -7,6 +7,10 @@ Create Date: 2025-07-24 14:50:48.779833
 """
 from alembic import op
 import models as models
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 
 
@@ -18,8 +22,18 @@ depends_on = None
 
 
 def upgrade():
-    op.execute("ALTER TABLE tidb_auth_bindings ALTER COLUMN status SET DEFAULT 'CREATING'::character varying")
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.execute("ALTER TABLE tidb_auth_bindings ALTER COLUMN status SET DEFAULT 'CREATING'::character varying")
+    else:
+        op.execute("ALTER TABLE tidb_auth_bindings ALTER COLUMN status SET DEFAULT 'CREATING'")
 
 
 def downgrade():
-    op.execute("ALTER TABLE tidb_auth_bindings ALTER COLUMN status SET DEFAULT 'CREATING'")
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.execute("ALTER TABLE tidb_auth_bindings ALTER COLUMN status SET DEFAULT 'CREATING'::character varying")
+    else:
+        op.execute("ALTER TABLE tidb_auth_bindings ALTER COLUMN status SET DEFAULT 'CREATING'")

api/migrations/versions/2025_08_09_1553-e8446f481c1e_add_provider_credential_pool_support.py

@@ -11,6 +11,10 @@ import models as models
 import sqlalchemy as sa
 from sqlalchemy.sql import table, column
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = 'e8446f481c1e'
 down_revision = 'fa8b0fa6f407'
@@ -20,16 +24,30 @@ depends_on = None
 
 def upgrade():
     # Create provider_credentials table
-    op.create_table('provider_credentials',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('provider_name', sa.String(length=255), nullable=False),
-    sa.Column('credential_name', sa.String(length=255), nullable=False),
-    sa.Column('encrypted_config', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='provider_credential_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('provider_credentials',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=255), nullable=False),
+        sa.Column('credential_name', sa.String(length=255), nullable=False),
+        sa.Column('encrypted_config', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='provider_credential_pkey')
+        )
+    else:
+        op.create_table('provider_credentials',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=255), nullable=False),
+        sa.Column('credential_name', sa.String(length=255), nullable=False),
+        sa.Column('encrypted_config', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='provider_credential_pkey')
+        )
 
     # Create index for provider_credentials
     with op.batch_alter_table('provider_credentials', schema=None) as batch_op:
@@ -60,27 +78,49 @@ def upgrade():
 
 def migrate_existing_providers_data():
     """migrate providers table data to provider_credentials"""
-
+    conn = op.get_bind()
     # Define table structure for data manipulation
-    providers_table = table('providers',
-        column('id', models.types.StringUUID()),
-        column('tenant_id', models.types.StringUUID()),
-        column('provider_name', sa.String()),
-        column('encrypted_config', sa.Text()),
-        column('created_at', sa.DateTime()),
-        column('updated_at', sa.DateTime()),
-        column('credential_id', models.types.StringUUID()),
-    )
+    if _is_pg(conn):
+        providers_table = table('providers',
+            column('id', models.types.StringUUID()),
+            column('tenant_id', models.types.StringUUID()),
+            column('provider_name', sa.String()),
+            column('encrypted_config', sa.Text()),
+            column('created_at', sa.DateTime()),
+            column('updated_at', sa.DateTime()),
+            column('credential_id', models.types.StringUUID()),
+        )
+    else:
+        providers_table = table('providers',
+            column('id', models.types.StringUUID()),
+            column('tenant_id', models.types.StringUUID()),
+            column('provider_name', sa.String()),
+            column('encrypted_config', models.types.LongText()),
+            column('created_at', sa.DateTime()),
+            column('updated_at', sa.DateTime()),
+            column('credential_id', models.types.StringUUID()),
+        )
 
-    provider_credential_table = table('provider_credentials',
-        column('id', models.types.StringUUID()),
-        column('tenant_id', models.types.StringUUID()),
-        column('provider_name', sa.String()),
-        column('credential_name', sa.String()),
-        column('encrypted_config', sa.Text()),
-        column('created_at', sa.DateTime()),
-        column('updated_at', sa.DateTime())
-    )
+    if _is_pg(conn):
+        provider_credential_table = table('provider_credentials',
+            column('id', models.types.StringUUID()),
+            column('tenant_id', models.types.StringUUID()),
+            column('provider_name', sa.String()),
+            column('credential_name', sa.String()),
+            column('encrypted_config', sa.Text()),
+            column('created_at', sa.DateTime()),
+            column('updated_at', sa.DateTime())
+        )
+    else:
+        provider_credential_table = table('provider_credentials',
+            column('id', models.types.StringUUID()),
+            column('tenant_id', models.types.StringUUID()),
+            column('provider_name', sa.String()),
+            column('credential_name', sa.String()),
+            column('encrypted_config', models.types.LongText()),
+            column('created_at', sa.DateTime()),
+            column('updated_at', sa.DateTime())
+        )
 
     # Get database connection
     conn = op.get_bind()
@@ -123,8 +163,14 @@ def migrate_existing_providers_data():
 
 def downgrade():
     # Re-add encrypted_config column to providers table
-    with op.batch_alter_table('providers', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('encrypted_config', sa.Text(), nullable=True))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('encrypted_config', sa.Text(), nullable=True))
+    else:
+        with op.batch_alter_table('providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('encrypted_config', models.types.LongText(), nullable=True))
 
     # Migrate data back from provider_credentials to providers
 

api/migrations/versions/2025_08_13_1605-0e154742a5fa_add_provider_model_multi_credential.py

@@ -13,6 +13,10 @@ import sqlalchemy as sa
 from sqlalchemy.sql import table, column
 
 
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
 # revision identifiers, used by Alembic.
 revision = '0e154742a5fa'
 down_revision = 'e8446f481c1e'
@@ -22,18 +26,34 @@ depends_on = None
 
 def upgrade():
     # Create provider_model_credentials table
-    op.create_table('provider_model_credentials',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('provider_name', sa.String(length=255), nullable=False),
-    sa.Column('model_name', sa.String(length=255), nullable=False),
-    sa.Column('model_type', sa.String(length=40), nullable=False),
-    sa.Column('credential_name', sa.String(length=255), nullable=False),
-    sa.Column('encrypted_config', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='provider_model_credential_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('provider_model_credentials',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=255), nullable=False),
+        sa.Column('model_name', sa.String(length=255), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('credential_name', sa.String(length=255), nullable=False),
+        sa.Column('encrypted_config', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='provider_model_credential_pkey')
+        )
+    else:
+        op.create_table('provider_model_credentials',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=255), nullable=False),
+        sa.Column('model_name', sa.String(length=255), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('credential_name', sa.String(length=255), nullable=False),
+        sa.Column('encrypted_config', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='provider_model_credential_pkey')
+        )
 
     # Create index for provider_model_credentials
     with op.batch_alter_table('provider_model_credentials', schema=None) as batch_op:
@@ -66,31 +86,57 @@ def upgrade():
 
 def migrate_existing_provider_models_data():
     """migrate provider_models table data to provider_model_credentials"""
-
+    conn = op.get_bind()
     # Define table structure for data manipulation
-    provider_models_table = table('provider_models',
-        column('id', models.types.StringUUID()),
-        column('tenant_id', models.types.StringUUID()),
-        column('provider_name', sa.String()),
-        column('model_name', sa.String()),
-        column('model_type', sa.String()),
-        column('encrypted_config', sa.Text()),
-        column('created_at', sa.DateTime()),
-        column('updated_at', sa.DateTime()),
-        column('credential_id', models.types.StringUUID()),
-    )
+    if _is_pg(conn):
+        provider_models_table = table('provider_models',
+            column('id', models.types.StringUUID()),
+            column('tenant_id', models.types.StringUUID()),
+            column('provider_name', sa.String()),
+            column('model_name', sa.String()),
+            column('model_type', sa.String()),
+            column('encrypted_config', sa.Text()),
+            column('created_at', sa.DateTime()),
+            column('updated_at', sa.DateTime()),
+            column('credential_id', models.types.StringUUID()),
+        )
+    else:
+        provider_models_table = table('provider_models',
+            column('id', models.types.StringUUID()),
+            column('tenant_id', models.types.StringUUID()),
+            column('provider_name', sa.String()),
+            column('model_name', sa.String()),
+            column('model_type', sa.String()),
+            column('encrypted_config', models.types.LongText()),
+            column('created_at', sa.DateTime()),
+            column('updated_at', sa.DateTime()),
+            column('credential_id', models.types.StringUUID()),
+        )
 
-    provider_model_credentials_table = table('provider_model_credentials',
-        column('id', models.types.StringUUID()),
-        column('tenant_id', models.types.StringUUID()),
-        column('provider_name', sa.String()),
-        column('model_name', sa.String()),
-        column('model_type', sa.String()),
-        column('credential_name', sa.String()),
-        column('encrypted_config', sa.Text()),
-        column('created_at', sa.DateTime()),
-        column('updated_at', sa.DateTime())
-    )
+    if _is_pg(conn):
+        provider_model_credentials_table = table('provider_model_credentials',
+            column('id', models.types.StringUUID()),
+            column('tenant_id', models.types.StringUUID()),
+            column('provider_name', sa.String()),
+            column('model_name', sa.String()),
+            column('model_type', sa.String()),
+            column('credential_name', sa.String()),
+            column('encrypted_config', sa.Text()),
+            column('created_at', sa.DateTime()),
+            column('updated_at', sa.DateTime())
+        )
+    else:
+        provider_model_credentials_table = table('provider_model_credentials',
+            column('id', models.types.StringUUID()),
+            column('tenant_id', models.types.StringUUID()),
+            column('provider_name', sa.String()),
+            column('model_name', sa.String()),
+            column('model_type', sa.String()),
+            column('credential_name', sa.String()),
+            column('encrypted_config', models.types.LongText()),
+            column('created_at', sa.DateTime()),
+            column('updated_at', sa.DateTime())
+        )
 
 
     # Get database connection
@@ -137,8 +183,14 @@ def migrate_existing_provider_models_data():
 
 def downgrade():
     # Re-add encrypted_config column to provider_models table
-    with op.batch_alter_table('provider_models', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('encrypted_config', sa.Text(), nullable=True))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('provider_models', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('encrypted_config', sa.Text(), nullable=True))
+    else:
+        with op.batch_alter_table('provider_models', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('encrypted_config', models.types.LongText(), nullable=True))
 
     if not context.is_offline_mode():
         # Migrate data back from provider_model_credentials to provider_models

api/migrations/versions/2025_08_20_1747-8d289573e1da_add_oauth_provider_apps.py

@@ -8,6 +8,11 @@ Create Date: 2025-08-20 17:47:17.015695
 from alembic import op
 import models as models
 import sqlalchemy as sa
+from libs.uuid_utils import uuidv7
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 
 # revision identifiers, used by Alembic.
@@ -19,17 +24,33 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('oauth_provider_apps',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('app_icon', sa.String(length=255), nullable=False),
-    sa.Column('app_label', sa.JSON(), server_default='{}', nullable=False),
-    sa.Column('client_id', sa.String(length=255), nullable=False),
-    sa.Column('client_secret', sa.String(length=255), nullable=False),
-    sa.Column('redirect_uris', sa.JSON(), server_default='[]', nullable=False),
-    sa.Column('scope', sa.String(length=255), server_default=sa.text("'read:name read:email read:avatar read:interface_language read:timezone'"), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='oauth_provider_app_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('oauth_provider_apps',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('app_icon', sa.String(length=255), nullable=False),
+        sa.Column('app_label', sa.JSON(), server_default='{}', nullable=False),
+        sa.Column('client_id', sa.String(length=255), nullable=False),
+        sa.Column('client_secret', sa.String(length=255), nullable=False),
+        sa.Column('redirect_uris', sa.JSON(), server_default='[]', nullable=False),
+        sa.Column('scope', sa.String(length=255), server_default=sa.text("'read:name read:email read:avatar read:interface_language read:timezone'"), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='oauth_provider_app_pkey')
+        )
+    else:
+        op.create_table('oauth_provider_apps',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('app_icon', sa.String(length=255), nullable=False),
+        sa.Column('app_label', sa.JSON(), default='{}', nullable=False),
+        sa.Column('client_id', sa.String(length=255), nullable=False),
+        sa.Column('client_secret', sa.String(length=255), nullable=False),
+        sa.Column('redirect_uris', sa.JSON(), default='[]', nullable=False),
+        sa.Column('scope', sa.String(length=255), server_default=sa.text("'read:name read:email read:avatar read:interface_language read:timezone'"), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='oauth_provider_app_pkey')
+        )
+    
     with op.batch_alter_table('oauth_provider_apps', schema=None) as batch_op:
         batch_op.create_index('oauth_provider_app_client_id_idx', ['client_id'], unique=False)
 

api/migrations/versions/2025_09_08_1007-c20211f18133_add_headers_to_mcp_provider.py

@@ -7,6 +7,10 @@ Create Date: 2025-08-29 10:07:54.163626
 """
 from alembic import op
 import models as models
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 
 
@@ -19,7 +23,12 @@ depends_on = None
 
 def upgrade():
     # Add encrypted_headers column to tool_mcp_providers table
-    op.add_column('tool_mcp_providers', sa.Column('encrypted_headers', sa.Text(), nullable=True))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.add_column('tool_mcp_providers', sa.Column('encrypted_headers', sa.Text(), nullable=True))
+    else:
+        op.add_column('tool_mcp_providers', sa.Column('encrypted_headers', models.types.LongText(), nullable=True))
     
 
 def downgrade():

api/migrations/versions/2025_09_11_1537-cf7c38a32b2d_add_credential_status_for_provider_table.py

@@ -7,6 +7,9 @@ Create Date: 2025-09-11 15:37:17.771298
 """
 from alembic import op
 import models as models
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 
 
@@ -19,8 +22,14 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('providers', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('credential_status', sa.String(length=20), server_default=sa.text("'active'::character varying"), nullable=True))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('credential_status', sa.String(length=20), server_default=sa.text("'active'::character varying"), nullable=True))
+    else:
+        with op.batch_alter_table('providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('credential_status', sa.String(length=20), server_default=sa.text("'active'"), nullable=True))
 
     # ### end Alembic commands ###
 

api/migrations/versions/2025_09_17_1515-68519ad5cd18_knowledge_pipeline_migrate.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from libs.uuid_utils import uuidv7
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '68519ad5cd18'
@@ -19,152 +24,314 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('datasource_oauth_params',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('plugin_id', sa.String(length=255), nullable=False),
-    sa.Column('provider', sa.String(length=255), nullable=False),
-    sa.Column('system_credentials', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='datasource_oauth_config_pkey'),
-    sa.UniqueConstraint('plugin_id', 'provider', name='datasource_oauth_config_datasource_id_provider_idx')
-    )
-    op.create_table('datasource_oauth_tenant_params',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('provider', sa.String(length=255), nullable=False),
-    sa.Column('plugin_id', sa.String(length=255), nullable=False),
-    sa.Column('client_params', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
-    sa.Column('enabled', sa.Boolean(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='datasource_oauth_tenant_config_pkey'),
-    sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', name='datasource_oauth_tenant_config_unique')
-    )
-    op.create_table('datasource_providers',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('provider', sa.String(length=255), nullable=False),
-    sa.Column('plugin_id', sa.String(length=255), nullable=False),
-    sa.Column('auth_type', sa.String(length=255), nullable=False),
-    sa.Column('encrypted_credentials', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
-    sa.Column('avatar_url', sa.Text(), nullable=True),
-    sa.Column('is_default', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('expires_at', sa.Integer(), server_default='-1', nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='datasource_provider_pkey'),
-    sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', 'name', name='datasource_provider_unique_name')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('datasource_oauth_params',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('plugin_id', sa.String(length=255), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('system_credentials', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='datasource_oauth_config_pkey'),
+        sa.UniqueConstraint('plugin_id', 'provider', name='datasource_oauth_config_datasource_id_provider_idx')
+        )
+    else:
+        op.create_table('datasource_oauth_params',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('plugin_id', sa.String(length=255), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('system_credentials', sa.JSON(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='datasource_oauth_config_pkey'),
+        sa.UniqueConstraint('plugin_id', 'provider', name='datasource_oauth_config_datasource_id_provider_idx')
+        )
+    if _is_pg(conn):
+        op.create_table('datasource_oauth_tenant_params',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('plugin_id', sa.String(length=255), nullable=False),
+        sa.Column('client_params', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column('enabled', sa.Boolean(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='datasource_oauth_tenant_config_pkey'),
+        sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', name='datasource_oauth_tenant_config_unique')
+        )
+    else:
+        op.create_table('datasource_oauth_tenant_params',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('plugin_id', sa.String(length=255), nullable=False),
+        sa.Column('client_params', sa.JSON(), nullable=False),
+        sa.Column('enabled', sa.Boolean(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='datasource_oauth_tenant_config_pkey'),
+        sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', name='datasource_oauth_tenant_config_unique')
+        )
+    if _is_pg(conn):
+        op.create_table('datasource_providers',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('provider', sa.String(length=255), nullable=False),
+        sa.Column('plugin_id', sa.String(length=255), nullable=False),
+        sa.Column('auth_type', sa.String(length=255), nullable=False),
+        sa.Column('encrypted_credentials', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column('avatar_url', sa.Text(), nullable=True),
+        sa.Column('is_default', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('expires_at', sa.Integer(), server_default='-1', nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='datasource_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', 'name', name='datasource_provider_unique_name')
+        )
+    else:
+        op.create_table('datasource_providers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('provider', sa.String(length=128), nullable=False),
+        sa.Column('plugin_id', sa.String(length=255), nullable=False),
+        sa.Column('auth_type', sa.String(length=255), nullable=False),
+        sa.Column('encrypted_credentials', sa.JSON(), nullable=False),
+        sa.Column('avatar_url', models.types.LongText(), nullable=True),
+        sa.Column('is_default', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('expires_at', sa.Integer(), server_default='-1', nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='datasource_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'plugin_id', 'provider', 'name', name='datasource_provider_unique_name')
+        )
     with op.batch_alter_table('datasource_providers', schema=None) as batch_op:
         batch_op.create_index('datasource_provider_auth_type_provider_idx', ['tenant_id', 'plugin_id', 'provider'], unique=False)
 
-    op.create_table('document_pipeline_execution_logs',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('pipeline_id', models.types.StringUUID(), nullable=False),
-    sa.Column('document_id', models.types.StringUUID(), nullable=False),
-    sa.Column('datasource_type', sa.String(length=255), nullable=False),
-    sa.Column('datasource_info', sa.Text(), nullable=False),
-    sa.Column('datasource_node_id', sa.String(length=255), nullable=False),
-    sa.Column('input_data', sa.JSON(), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=True),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='document_pipeline_execution_log_pkey')
-    )
+    if _is_pg(conn):
+        op.create_table('document_pipeline_execution_logs',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('pipeline_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('datasource_type', sa.String(length=255), nullable=False),
+        sa.Column('datasource_info', sa.Text(), nullable=False),
+        sa.Column('datasource_node_id', sa.String(length=255), nullable=False),
+        sa.Column('input_data', sa.JSON(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='document_pipeline_execution_log_pkey')
+        )
+    else:
+        op.create_table('document_pipeline_execution_logs',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('pipeline_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('datasource_type', sa.String(length=255), nullable=False),
+        sa.Column('datasource_info', models.types.LongText(), nullable=False),
+        sa.Column('datasource_node_id', sa.String(length=255), nullable=False),
+        sa.Column('input_data', sa.JSON(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='document_pipeline_execution_log_pkey')
+        )
     with op.batch_alter_table('document_pipeline_execution_logs', schema=None) as batch_op:
         batch_op.create_index('document_pipeline_execution_logs_document_id_idx', ['document_id'], unique=False)
 
-    op.create_table('pipeline_built_in_templates',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('description', sa.Text(), nullable=False),
-    sa.Column('chunk_structure', sa.String(length=255), nullable=False),
-    sa.Column('icon', sa.JSON(), nullable=False),
-    sa.Column('yaml_content', sa.Text(), nullable=False),
-    sa.Column('copyright', sa.String(length=255), nullable=False),
-    sa.Column('privacy_policy', sa.String(length=255), nullable=False),
-    sa.Column('position', sa.Integer(), nullable=False),
-    sa.Column('install_count', sa.Integer(), nullable=False),
-    sa.Column('language', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.PrimaryKeyConstraint('id', name='pipeline_built_in_template_pkey')
-    )
-    op.create_table('pipeline_customized_templates',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('description', sa.Text(), nullable=False),
-    sa.Column('chunk_structure', sa.String(length=255), nullable=False),
-    sa.Column('icon', sa.JSON(), nullable=False),
-    sa.Column('position', sa.Integer(), nullable=False),
-    sa.Column('yaml_content', sa.Text(), nullable=False),
-    sa.Column('install_count', sa.Integer(), nullable=False),
-    sa.Column('language', sa.String(length=255), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='pipeline_customized_template_pkey')
-    )
+    if _is_pg(conn):
+        op.create_table('pipeline_built_in_templates',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.Text(), nullable=False),
+        sa.Column('chunk_structure', sa.String(length=255), nullable=False),
+        sa.Column('icon', sa.JSON(), nullable=False),
+        sa.Column('yaml_content', sa.Text(), nullable=False),
+        sa.Column('copyright', sa.String(length=255), nullable=False),
+        sa.Column('privacy_policy', sa.String(length=255), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('install_count', sa.Integer(), nullable=False),
+        sa.Column('language', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='pipeline_built_in_template_pkey')
+        )
+    else:
+        op.create_table('pipeline_built_in_templates',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', models.types.LongText(), nullable=False),
+        sa.Column('chunk_structure', sa.String(length=255), nullable=False),
+        sa.Column('icon', sa.JSON(), nullable=False),
+        sa.Column('yaml_content', models.types.LongText(), nullable=False),
+        sa.Column('copyright', sa.String(length=255), nullable=False),
+        sa.Column('privacy_policy', sa.String(length=255), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('install_count', sa.Integer(), nullable=False),
+        sa.Column('language', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='pipeline_built_in_template_pkey')
+        )
+    if _is_pg(conn):
+        op.create_table('pipeline_customized_templates',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.Text(), nullable=False),
+        sa.Column('chunk_structure', sa.String(length=255), nullable=False),
+        sa.Column('icon', sa.JSON(), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('yaml_content', sa.Text(), nullable=False),
+        sa.Column('install_count', sa.Integer(), nullable=False),
+        sa.Column('language', sa.String(length=255), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='pipeline_customized_template_pkey')
+        )
+    else:
+        # MySQL: Use compatible syntax
+        op.create_table('pipeline_customized_templates',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', models.types.LongText(), nullable=False),
+        sa.Column('chunk_structure', sa.String(length=255), nullable=False),
+        sa.Column('icon', sa.JSON(), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('yaml_content', models.types.LongText(), nullable=False),
+        sa.Column('install_count', sa.Integer(), nullable=False),
+        sa.Column('language', sa.String(length=255), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='pipeline_customized_template_pkey')
+        )
     with op.batch_alter_table('pipeline_customized_templates', schema=None) as batch_op:
         batch_op.create_index('pipeline_customized_template_tenant_idx', ['tenant_id'], unique=False)
 
-    op.create_table('pipeline_recommended_plugins',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('plugin_id', sa.Text(), nullable=False),
-    sa.Column('provider_name', sa.Text(), nullable=False),
-    sa.Column('position', sa.Integer(), nullable=False),
-    sa.Column('active', sa.Boolean(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='pipeline_recommended_plugin_pkey')
-    )
-    op.create_table('pipelines',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('description', sa.Text(), server_default=sa.text("''::character varying"), nullable=False),
-    sa.Column('workflow_id', models.types.StringUUID(), nullable=True),
-    sa.Column('is_public', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('is_published', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=True),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='pipeline_pkey')
-    )
-    op.create_table('workflow_draft_variable_files',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False, comment='The tenant to which the WorkflowDraftVariableFile belongs, referencing Tenant.id'),
-    sa.Column('app_id', models.types.StringUUID(), nullable=False, comment='The application to which the WorkflowDraftVariableFile belongs, referencing App.id'),
-    sa.Column('user_id', models.types.StringUUID(), nullable=False, comment='The owner to of the WorkflowDraftVariableFile, referencing Account.id'),
-    sa.Column('upload_file_id', models.types.StringUUID(), nullable=False, comment='Reference to UploadFile containing the large variable data'),
-    sa.Column('size', sa.BigInteger(), nullable=False, comment='Size of the original variable content in bytes'),
-    sa.Column('length', sa.Integer(), nullable=True, comment='Length of the original variable content. For array and array-like types, this represents the number of elements. For object types, it indicates the number of keys. For other types, the value is NULL.'),
-    sa.Column('value_type', sa.String(20), nullable=False),
-    sa.PrimaryKeyConstraint('id', name=op.f('workflow_draft_variable_files_pkey'))
-    )
-    op.create_table('workflow_node_execution_offload',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('app_id', models.types.StringUUID(), nullable=False),
-    sa.Column('node_execution_id', models.types.StringUUID(), nullable=True),
-    sa.Column('type', sa.String(20), nullable=False),
-    sa.Column('file_id', models.types.StringUUID(), nullable=False),
-    sa.PrimaryKeyConstraint('id', name=op.f('workflow_node_execution_offload_pkey')),
-    sa.UniqueConstraint('node_execution_id', 'type', name=op.f('workflow_node_execution_offload_node_execution_id_key'))
-    )
-    with op.batch_alter_table('datasets', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('keyword_number', sa.Integer(), server_default=sa.text('10'), nullable=True))
-        batch_op.add_column(sa.Column('icon_info', postgresql.JSONB(astext_type=sa.Text()), nullable=True))
-        batch_op.add_column(sa.Column('runtime_mode', sa.String(length=255), server_default=sa.text("'general'::character varying"), nullable=True))
-        batch_op.add_column(sa.Column('pipeline_id', models.types.StringUUID(), nullable=True))
-        batch_op.add_column(sa.Column('chunk_structure', sa.String(length=255), nullable=True))
-        batch_op.add_column(sa.Column('enable_api', sa.Boolean(), server_default=sa.text('true'), nullable=False))
+    if _is_pg(conn):
+        op.create_table('pipeline_recommended_plugins',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('plugin_id', sa.Text(), nullable=False),
+        sa.Column('provider_name', sa.Text(), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('active', sa.Boolean(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='pipeline_recommended_plugin_pkey')
+        )
+    else:
+        op.create_table('pipeline_recommended_plugins',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('plugin_id', models.types.LongText(), nullable=False),
+        sa.Column('provider_name', models.types.LongText(), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('active', sa.Boolean(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='pipeline_recommended_plugin_pkey')
+        )
+    if _is_pg(conn):
+        op.create_table('pipelines',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.Text(), server_default=sa.text("''::character varying"), nullable=False),
+        sa.Column('workflow_id', models.types.StringUUID(), nullable=True),
+        sa.Column('is_public', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('is_published', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='pipeline_pkey')
+        )
+    else:
+        op.create_table('pipelines',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', models.types.LongText(), default=sa.text("''"), nullable=False),
+        sa.Column('workflow_id', models.types.StringUUID(), nullable=True),
+        sa.Column('is_public', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('is_published', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='pipeline_pkey')
+        )
+    if _is_pg(conn):
+        op.create_table('workflow_draft_variable_files',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False, comment='The tenant to which the WorkflowDraftVariableFile belongs, referencing Tenant.id'),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False, comment='The application to which the WorkflowDraftVariableFile belongs, referencing App.id'),
+        sa.Column('user_id', models.types.StringUUID(), nullable=False, comment='The owner to of the WorkflowDraftVariableFile, referencing Account.id'),
+        sa.Column('upload_file_id', models.types.StringUUID(), nullable=False, comment='Reference to UploadFile containing the large variable data'),
+        sa.Column('size', sa.BigInteger(), nullable=False, comment='Size of the original variable content in bytes'),
+        sa.Column('length', sa.Integer(), nullable=True, comment='Length of the original variable content. For array and array-like types, this represents the number of elements. For object types, it indicates the number of keys. For other types, the value is NULL.'),
+        sa.Column('value_type', sa.String(20), nullable=False),
+        sa.PrimaryKeyConstraint('id', name=op.f('workflow_draft_variable_files_pkey'))
+        )
+    else:
+        op.create_table('workflow_draft_variable_files',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False, comment='The tenant to which the WorkflowDraftVariableFile belongs, referencing Tenant.id'),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False, comment='The application to which the WorkflowDraftVariableFile belongs, referencing App.id'),
+        sa.Column('user_id', models.types.StringUUID(), nullable=False, comment='The owner to of the WorkflowDraftVariableFile, referencing Account.id'),
+        sa.Column('upload_file_id', models.types.StringUUID(), nullable=False, comment='Reference to UploadFile containing the large variable data'),
+        sa.Column('size', sa.BigInteger(), nullable=False, comment='Size of the original variable content in bytes'),
+        sa.Column('length', sa.Integer(), nullable=True, comment='Length of the original variable content. For array and array-like types, this represents the number of elements. For object types, it indicates the number of keys. For other types, the value is NULL.'),
+        sa.Column('value_type', sa.String(20), nullable=False),
+        sa.PrimaryKeyConstraint('id', name=op.f('workflow_draft_variable_files_pkey'))
+        )
+    if _is_pg(conn):
+        op.create_table('workflow_node_execution_offload',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuidv7()'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('node_execution_id', models.types.StringUUID(), nullable=True),
+        sa.Column('type', sa.String(20), nullable=False),
+        sa.Column('file_id', models.types.StringUUID(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name=op.f('workflow_node_execution_offload_pkey')),
+        sa.UniqueConstraint('node_execution_id', 'type', name=op.f('workflow_node_execution_offload_node_execution_id_key'))
+        )
+    else:
+        op.create_table('workflow_node_execution_offload',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('node_execution_id', models.types.StringUUID(), nullable=True),
+        sa.Column('type', sa.String(20), nullable=False),
+        sa.Column('file_id', models.types.StringUUID(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name=op.f('workflow_node_execution_offload_pkey')),
+        sa.UniqueConstraint('node_execution_id', 'type', name=op.f('workflow_node_execution_offload_node_execution_id_key'))
+        )
+    if _is_pg(conn):
+        with op.batch_alter_table('datasets', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('keyword_number', sa.Integer(), server_default=sa.text('10'), nullable=True))
+            batch_op.add_column(sa.Column('icon_info', postgresql.JSONB(astext_type=sa.Text()), nullable=True))
+            batch_op.add_column(sa.Column('runtime_mode', sa.String(length=255), server_default=sa.text("'general'::character varying"), nullable=True))
+            batch_op.add_column(sa.Column('pipeline_id', models.types.StringUUID(), nullable=True))
+            batch_op.add_column(sa.Column('chunk_structure', sa.String(length=255), nullable=True))
+            batch_op.add_column(sa.Column('enable_api', sa.Boolean(), server_default=sa.text('true'), nullable=False))
+    else:
+        with op.batch_alter_table('datasets', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('keyword_number', sa.Integer(), server_default=sa.text('10'), nullable=True))
+            batch_op.add_column(sa.Column('icon_info', sa.JSON(), nullable=True))
+            batch_op.add_column(sa.Column('runtime_mode', sa.String(length=255), server_default=sa.text("'general'"), nullable=True))
+            batch_op.add_column(sa.Column('pipeline_id', models.types.StringUUID(), nullable=True))
+            batch_op.add_column(sa.Column('chunk_structure', sa.String(length=255), nullable=True))
+            batch_op.add_column(sa.Column('enable_api', sa.Boolean(), server_default=sa.text('true'), nullable=False))
 
     with op.batch_alter_table('workflow_draft_variables', schema=None) as batch_op:
         batch_op.add_column(sa.Column('file_id', models.types.StringUUID(), nullable=True, comment='Reference to WorkflowDraftVariableFile if variable is offloaded to external storage'))
@@ -175,9 +342,12 @@ def upgrade():
                     comment='Indicates whether the current value is the default for a conversation variable. Always `FALSE` for other types of variables.',)
             )
         batch_op.create_index('workflow_draft_variable_file_id_idx', ['file_id'], unique=False)
-
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('rag_pipeline_variables', sa.Text(), server_default='{}', nullable=False))
+    if _is_pg(conn):
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('rag_pipeline_variables', sa.Text(), server_default='{}', nullable=False))
+    else:
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('rag_pipeline_variables', models.types.LongText(), default='{}', nullable=False))
 
     # ### end Alembic commands ###
 

api/migrations/versions/2025_10_21_1430-ae662b25d9bc_remove_builtin_template_user.py

@@ -7,6 +7,10 @@ Create Date: 2025-10-21 14:30:28.566192
 """
 from alembic import op
 import models as models
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 
 
@@ -29,8 +33,15 @@ def upgrade():
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('pipeline_built_in_templates', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('created_by', sa.UUID(), autoincrement=False, nullable=False))
-        batch_op.add_column(sa.Column('updated_by', sa.UUID(), autoincrement=False, nullable=True))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('pipeline_built_in_templates', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('created_by', sa.UUID(), autoincrement=False, nullable=False))
+            batch_op.add_column(sa.Column('updated_by', sa.UUID(), autoincrement=False, nullable=True))
+    else:
+        with op.batch_alter_table('pipeline_built_in_templates', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('created_by', models.types.StringUUID(), autoincrement=False, nullable=False))
+            batch_op.add_column(sa.Column('updated_by', models.types.StringUUID(), autoincrement=False, nullable=True))
 
     # ### end Alembic commands ###

api/migrations/versions/2025_10_22_1611-03f8dcbc611e_add_workflowpause_model.py

@@ -9,7 +9,10 @@ Create Date: 2025-10-22 16:11:31.805407
 from alembic import op
 import models as models
 import sqlalchemy as sa
+from libs.uuid_utils import uuidv7
 
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = "03f8dcbc611e"
@@ -19,19 +22,32 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table(
-        "workflow_pauses",
-        sa.Column("workflow_id", models.types.StringUUID(), nullable=False),
-        sa.Column("workflow_run_id", models.types.StringUUID(), nullable=False),
-        sa.Column("resumed_at", sa.DateTime(), nullable=True),
-        sa.Column("state_object_key", sa.String(length=255), nullable=False),
-        sa.Column("id", models.types.StringUUID(), server_default=sa.text("uuidv7()"), nullable=False),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.PrimaryKeyConstraint("id", name=op.f("workflow_pauses_pkey")),
-        sa.UniqueConstraint("workflow_run_id", name=op.f("workflow_pauses_workflow_run_id_key")),
-    )
-
+    if _is_pg(op.get_context().bind):
+        op.create_table(
+            "workflow_pauses",
+            sa.Column("workflow_id", models.types.StringUUID(), nullable=False),
+            sa.Column("workflow_run_id", models.types.StringUUID(), nullable=False),
+            sa.Column("resumed_at", sa.DateTime(), nullable=True),
+            sa.Column("state_object_key", sa.String(length=255), nullable=False),
+            sa.Column("id", models.types.StringUUID(), server_default=sa.text("uuidv7()"), nullable=False),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.PrimaryKeyConstraint("id", name=op.f("workflow_pauses_pkey")),
+            sa.UniqueConstraint("workflow_run_id", name=op.f("workflow_pauses_workflow_run_id_key")),
+        )
+    else:
+        op.create_table(
+            "workflow_pauses",
+            sa.Column("workflow_id", models.types.StringUUID(), nullable=False),
+            sa.Column("workflow_run_id", models.types.StringUUID(), nullable=False),
+            sa.Column("resumed_at", sa.DateTime(), nullable=True),
+            sa.Column("state_object_key", sa.String(length=255), nullable=False),
+            sa.Column("id", models.types.StringUUID(), default=lambda: str(uuidv7()), nullable=False),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.PrimaryKeyConstraint("id", name=op.f("workflow_pauses_pkey")),
+            sa.UniqueConstraint("workflow_run_id", name=op.f("workflow_pauses_workflow_run_id_key")),
+        )
     # ### end Alembic commands ###