Merge branch 'fix/polyfill-toSplice' into deploy/dev

api/tasks/annotation/batch_import_annotations_task.py

@@ -31,7 +31,6 @@ def batch_import_annotations_task(job_id: str, content_list: list[dict], app_id:
     start_at = time.perf_counter()
     indexing_cache_key = f"app_annotation_batch_import_{str(job_id)}"
     active_jobs_key = f"annotation_import_active:{tenant_id}"
-    
     # get app info
     app = db.session.query(App).where(App.id == app_id, App.tenant_id == tenant_id, App.status == "normal").first()
 
@@ -100,6 +99,5 @@ def batch_import_annotations_task(job_id: str, content_list: list[dict], app_id:
             except Exception as cleanup_error:
                 # Log but don't fail if cleanup fails - the job will be auto-expired
                 logger.warning("Failed to clean up active job tracking for %s: %s", job_id, cleanup_error)
-            
             # Close database session
             db.session.close()

api/tasks/clean_dataset_task.py

@@ -9,6 +9,7 @@ from core.rag.index_processor.index_processor_factory import IndexProcessorFacto
 from core.tools.utils.web_reader_tool import get_image_upload_file_ids
 from extensions.ext_database import db
 from extensions.ext_storage import storage
+from models import WorkflowType
 from models.dataset import (
     AppDatasetJoin,
     Dataset,
@@ -18,9 +19,11 @@ from models.dataset import (
     DatasetQuery,
     Document,
     DocumentSegment,
+    Pipeline,
     SegmentAttachmentBinding,
 )
 from models.model import UploadFile
+from models.workflow import Workflow
 
 logger = logging.getLogger(__name__)
 
@@ -34,6 +37,7 @@ def clean_dataset_task(
     index_struct: str,
     collection_binding_id: str,
     doc_form: str,
+    pipeline_id: str | None = None,
 ):
     """
     Clean dataset when dataset deleted.
@@ -135,6 +139,14 @@ def clean_dataset_task(
         # delete dataset metadata
         db.session.query(DatasetMetadata).where(DatasetMetadata.dataset_id == dataset_id).delete()
         db.session.query(DatasetMetadataBinding).where(DatasetMetadataBinding.dataset_id == dataset_id).delete()
+        # delete pipeline and workflow
+        if pipeline_id:
+            db.session.query(Pipeline).where(Pipeline.id == pipeline_id).delete()
+            db.session.query(Workflow).where(
+                Workflow.tenant_id == tenant_id,
+                Workflow.app_id == pipeline_id,
+                Workflow.type == WorkflowType.RAG_PIPELINE,
+            ).delete()
         # delete files
         if documents:
             for document in documents:

api/tasks/delete_account_task.py

@@ -2,6 +2,7 @@ import logging
 
 from celery import shared_task
 
+from configs import dify_config
 from extensions.ext_database import db
 from models import Account
 from services.billing_service import BillingService
@@ -14,7 +15,8 @@ logger = logging.getLogger(__name__)
 def delete_account_task(account_id):
     account = db.session.query(Account).where(Account.id == account_id).first()
     try:
-        BillingService.delete_account(account_id)
+        if dify_config.BILLING_ENABLED:
+            BillingService.delete_account(account_id)
     except Exception:
         logger.exception("Failed to delete account %s from billing service.", account_id)
         raise

api/tasks/document_indexing_sync_task.py

@@ -2,7 +2,6 @@ import logging
 import time
 
 import click
-import sqlalchemy as sa
 from celery import shared_task
 from sqlalchemy import select
 
@@ -12,7 +11,7 @@ from core.rag.index_processor.index_processor_factory import IndexProcessorFacto
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from models.dataset import Dataset, Document, DocumentSegment
-from models.source import DataSourceOauthBinding
+from services.datasource_provider_service import DatasourceProviderService
 
 logger = logging.getLogger(__name__)
 
@@ -48,27 +47,36 @@ def document_indexing_sync_task(dataset_id: str, document_id: str):
         page_id = data_source_info["notion_page_id"]
         page_type = data_source_info["type"]
         page_edited_time = data_source_info["last_edited_time"]
+        credential_id = data_source_info.get("credential_id")
 
-        data_source_binding = (
-            db.session.query(DataSourceOauthBinding)
-            .where(
-                sa.and_(
-                    DataSourceOauthBinding.tenant_id == document.tenant_id,
-                    DataSourceOauthBinding.provider == "notion",
-                    DataSourceOauthBinding.disabled == False,
-                    DataSourceOauthBinding.source_info["workspace_id"] == f'"{workspace_id}"',
-                )
-            )
-            .first()
+        # Get credentials from datasource provider
+        datasource_provider_service = DatasourceProviderService()
+        credential = datasource_provider_service.get_datasource_credentials(
+            tenant_id=document.tenant_id,
+            credential_id=credential_id,
+            provider="notion_datasource",
+            plugin_id="langgenius/notion_datasource",
         )
-        if not data_source_binding:
-            raise ValueError("Data source binding not found.")
+
+        if not credential:
+            logger.error(
+                "Datasource credential not found for document %s, tenant_id: %s, credential_id: %s",
+                document_id,
+                document.tenant_id,
+                credential_id,
+            )
+            document.indexing_status = "error"
+            document.error = "Datasource credential not found. Please reconnect your Notion workspace."
+            document.stopped_at = naive_utc_now()
+            db.session.commit()
+            db.session.close()
+            return
 
         loader = NotionExtractor(
             notion_workspace_id=workspace_id,
             notion_obj_id=page_id,
             notion_page_type=page_type,
-            notion_access_token=data_source_binding.access_token,
+            notion_access_token=credential.get("integration_secret"),
             tenant_id=document.tenant_id,
         )