test: mock to sqlite3

api/controllers/console/datasets/external.py

@@ -1,4 +1,5 @@
-from datetime import datetime
+from __future__ import annotations
+
 from typing import Any
 from uuid import UUID
 
@@ -31,6 +32,7 @@
 from models import Account
 from services.dataset_service import DatasetService
 from services.enterprise import rbac_service as enterprise_rbac_service
+from services.entities.external_knowledge_entities.external_knowledge_entities import ExternalDatasetCreatePayload
 from services.external_knowledge_service import ExternalDatasetService
 from services.hit_testing_service import HitTestingService
 from services.knowledge_service import BedrockRetrievalSetting, ExternalDatasetTestService
@@ -41,22 +43,14 @@
     settings: dict[str, Any]
 
 
-class ExternalDatasetCreatePayload(BaseModel):
-    external_knowledge_api_id: str
-    external_knowledge_id: str
-    name: str = Field(..., min_length=1, max_length=100)
-    description: str | None = Field(None, max_length=400)
-    external_retrieval_model: dict[str, Any] | None = None
-
-
 class ExternalHitTestingPayload(BaseModel):
     query: str
     external_retrieval_model: dict[str, Any] | None = None
     metadata_filtering_conditions: dict[str, Any] | None = None
 
 
 class BedrockRetrievalPayload(BaseModel):
-    retrieval_setting: "BedrockRetrievalSetting"
+    retrieval_setting: BedrockRetrievalSetting
     query: str
     knowledge_id: str
 
@@ -84,8 +78,8 @@
 
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | str) -> str:
         if isinstance(value, datetime):
             return value.isoformat()
         return value
 
@@ -331,7 +325,6 @@
     def post(self, session: Session, current_tenant_id: str, current_user: Account):
         # The role of the current user in the ta table must be admin, owner, or editor
         payload = ExternalDatasetCreatePayload.model_validate(console_ns.payload or {})
-        args = payload.model_dump(exclude_none=True)
 
         # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
         if not current_user.is_dataset_editor:
@@ -341,7 +334,7 @@
             dataset = ExternalDatasetService.create_external_dataset(
                 tenant_id=current_tenant_id,
                 user_id=current_user.id,
-                args=args,
+                args=payload,
                 session=session,
             )
         except services.errors.dataset.DatasetNameDuplicateError:

api/services/entities/external_knowledge_entities/external_knowledge_entities.py

@@ -1,6 +1,6 @@
 from typing import Any, Literal, Union
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 
 class AuthorizationConfig(BaseModel):
@@ -24,3 +24,19 @@ class ExternalKnowledgeApiSetting(BaseModel):
     request_method: str
     headers: dict[str, Any] | None = None
     params: dict[str, Any] | None = None
+
+
+class ExternalDatasetCreatePayload(BaseModel):
+    """Validated fields required to create an external dataset binding.
+
+    The console controller owns HTTP concerns, but the service also needs this
+    contract when creating the tenant-scoped dataset and external knowledge
+    binding. Keep it outside controllers so service imports do not depend on
+    Flask blueprint initialization.
+    """
+
+    external_knowledge_api_id: str
+    external_knowledge_id: str
+    name: str = Field(..., min_length=1, max_length=100)
+    description: str | None = Field(None, max_length=400)
+    external_retrieval_model: dict[str, object] | None = Field(default=None)

api/services/external_knowledge_service.py

@@ -1,3 +1,4 @@
+from controllers.console.datasets.external import ExternalDatasetCreatePayload
 import json
 from copy import deepcopy
 from typing import Any, cast
@@ -274,16 +275,16 @@ def get_external_knowledge_api_settings(settings: dict[str, Any]) -> ExternalKno
         return ExternalKnowledgeApiSetting.model_validate(settings)
 
     @staticmethod
-    def create_external_dataset(tenant_id: str, user_id: str, args: dict[str, Any], *, session: Session) -> Dataset:
+    def create_external_dataset(tenant_id: str, user_id: str, args: ExternalDatasetCreatePayload, session: Session) -> Dataset:
         # check if dataset name already exists
         if session.scalar(
-            select(Dataset).where(Dataset.name == args.get("name"), Dataset.tenant_id == tenant_id).limit(1)
+            select(Dataset).where(Dataset.name == args.name, Dataset.tenant_id == tenant_id).limit(1)
         ):
-            raise DatasetNameDuplicateError(f"Dataset with name {args.get('name')} already exists.")
+            raise DatasetNameDuplicateError(f"Dataset with name {args.name} already exists.")
         external_knowledge_api = session.scalar(
             select(ExternalKnowledgeApis)
             .where(
-                ExternalKnowledgeApis.id == args.get("external_knowledge_api_id"),
+                ExternalKnowledgeApis.id == args.external_knowledge_api_id,
                 ExternalKnowledgeApis.tenant_id == tenant_id,
             )
             .limit(1)
@@ -294,26 +295,22 @@ def create_external_dataset(tenant_id: str, user_id: str, args: dict[str, Any],
 
         dataset = Dataset(
             tenant_id=tenant_id,
-            name=args.get("name"),
-            description=args.get("description", ""),
+            name=args.name,
+            description=args.description or "",
             provider="external",
-            retrieval_model=args.get("external_retrieval_model"),
+            retrieval_model=args.external_retrieval_model,
             created_by=user_id,
             maintainer=user_id,
         )
 
         session.add(dataset)
         session.flush()
-        if args.get("external_knowledge_id") is None:
-            raise ValueError("external_knowledge_id is required")
-        if args.get("external_knowledge_api_id") is None:
-            raise ValueError("external_knowledge_api_id is required")
 
         external_knowledge_binding = ExternalKnowledgeBindings(
             tenant_id=tenant_id,
             dataset_id=dataset.id,
-            external_knowledge_api_id=args.get("external_knowledge_api_id") or "",
-            external_knowledge_id=args.get("external_knowledge_id") or "",
+            external_knowledge_api_id=args.external_knowledge_api_id or "",
+            external_knowledge_id=args.external_knowledge_id or "",
             created_by=user_id,
         )
         session.add(external_knowledge_binding)

api/tests/unit_tests/controllers/console/datasets/test_external_dataset_payload.py

@@ -11,7 +11,7 @@
 import pytest
 from pydantic import ValidationError
 
-from controllers.console.datasets.external import ExternalDatasetCreatePayload
+from services.entities.external_knowledge_entities.external_knowledge_entities import ExternalDatasetCreatePayload
 
 
 def test_external_dataset_create_payload_allows_name_length_100() -> None: