refactor: cleanup duplicate code (#36173)

api/controllers/common/human_input.py

@@ -1,6 +1,21 @@
+import json
+
 from pydantic import BaseModel, JsonValue
 
 
 class HumanInputFormSubmitPayload(BaseModel):
     inputs: dict[str, JsonValue]
     action: str
+
+
+def stringify_form_default_values(values: dict[str, object]) -> dict[str, str]:
+    """Serialize default values into strings expected by human-input form clients."""
+    result: dict[str, str] = {}
+    for key, value in values.items():
+        if value is None:
+            result[key] = ""
+        elif isinstance(value, (dict, list)):
+            result[key] = json.dumps(value, ensure_ascii=False)
+        else:
+            result[key] = str(value)
+    return result

api/controllers/console/apikey.py

@@ -11,6 +11,7 @@ from werkzeug.exceptions import Forbidden
 from controllers.common.schema import register_schema_models
 from extensions.ext_database import db
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
 from models.enums import ApiTokenType
@@ -21,12 +22,6 @@ from . import console_ns
 from .wraps import account_initialization_required, edit_permission_required, setup_required
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class ApiKeyItem(ResponseModel):
     id: str
     type: str
@@ -37,7 +32,7 @@ class ApiKeyItem(ResponseModel):
     @field_validator("last_used_at", "created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ApiKeyList(ResponseModel):

api/controllers/console/app/app.py

@@ -34,7 +34,7 @@ from core.trigger.constants import TRIGGER_NODE_TYPES
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.enums import WorkflowExecutionStatus
-from libs.helper import build_icon_url
+from libs.helper import build_icon_url, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType
@@ -178,12 +178,6 @@ class AppTracePayload(BaseModel):
 type JSONValue = Any
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class Tag(ResponseModel):
     id: str
     name: str
@@ -200,7 +194,7 @@ class WorkflowPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ModelConfigPartial(ResponseModel):
@@ -214,7 +208,7 @@ class ModelConfigPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ModelConfig(ResponseModel):
@@ -275,7 +269,7 @@ class ModelConfig(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class Site(ResponseModel):
@@ -318,7 +312,7 @@ class Site(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DeletedTool(ResponseModel):
@@ -361,7 +355,7 @@ class AppPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AppDetail(ResponseModel):
@@ -391,7 +385,7 @@ class AppDetail(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AppDetailWithSite(AppDetail):

api/controllers/console/app/conversation_variables.py

@@ -16,6 +16,7 @@ from controllers.console.wraps import account_initialization_required, setup_req
 from extensions.ext_database import db
 from fields._value_type_serializer import serialize_value_type
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import login_required
 from models import ConversationVariable
 from models.model import AppMode
@@ -25,12 +26,6 @@ class ConversationVariablesQuery(BaseModel):
     conversation_id: str = Field(..., description="Conversation ID to filter variables")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class ConversationVariableResponse(ResponseModel):
     id: str
     name: str
@@ -65,7 +60,7 @@ class ConversationVariableResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class PaginatedConversationVariableResponse(ResponseModel):

api/controllers/console/app/mcp_server.py

@@ -13,6 +13,7 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from extensions.ext_database import db
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.enums import AppMCPServerStatus
 from models.model import AppMCPServer
@@ -30,12 +31,6 @@ class MCPServerUpdatePayload(BaseModel):
     status: str | None = Field(default=None, description="Server status")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class AppMCPServerResponse(ResponseModel):
     id: str
     name: str
@@ -59,7 +54,7 @@ class AppMCPServerResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 register_schema_models(console_ns, MCPServerCreatePayload, MCPServerUpdatePayload, AppMCPServerResponse)

api/controllers/console/app/message.py

@@ -37,10 +37,9 @@ from fields.conversation_fields import (
     JSONValue,
     MessageFile,
     format_files_contained,
-    to_timestamp,
 )
 from graphon.model_runtime.errors.invoke import InvokeError
-from libs.helper import uuid_value
+from libs.helper import to_timestamp, uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required
 from models.enums import FeedbackFromSource, FeedbackRating
@@ -144,9 +143,7 @@ class MessageDetailResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class MessageInfiniteScrollPaginationResponse(ResponseModel):

api/controllers/console/app/workflow_app_log.py

@@ -16,6 +16,7 @@ from fields.base import ResponseModel
 from fields.end_user_fields import SimpleEndUser
 from fields.member_fields import SimpleAccount
 from graphon.enums import WorkflowExecutionStatus
+from libs.helper import to_timestamp
 from libs.login import login_required
 from models import App
 from models.model import AppMode
@@ -82,9 +83,7 @@ class WorkflowRunForLogResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowRunForArchivedLogResponse(ResponseModel):
@@ -117,9 +116,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowArchivedLogPartialResponse(ResponseModel):
@@ -133,9 +130,7 @@ class WorkflowArchivedLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/controllers/console/datasets/datasets_document.py

@@ -39,6 +39,7 @@ from fields.document_fields import (
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from libs.datetime_utils import naive_utc_now
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import DatasetProcessRule, Document, DocumentSegment, UploadFile
 from models.dataset import DocumentPipelineExecutionLog
@@ -71,12 +72,6 @@ from ..wraps import (
 logger = logging.getLogger(__name__)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 def _normalize_enum(value: Any) -> Any:
     if isinstance(value, str) or value is None:
         return value
@@ -101,7 +96,7 @@ class DatasetResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DocumentMetadataResponse(ResponseModel):
@@ -152,7 +147,7 @@ class DocumentResponse(ResponseModel):
     @field_validator("created_at", "disabled_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DocumentWithSegmentsResponse(DocumentResponse):

api/controllers/console/datasets/hit_testing.py

@@ -8,6 +8,7 @@ from pydantic import Field, field_validator
 
 from controllers.common.schema import register_schema_models
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import login_required
 
 from .. import console_ns
@@ -19,12 +20,6 @@ from ..wraps import (
 )
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class HitTestingDocument(ResponseModel):
     id: str | None = None
     data_source_type: str | None = None
@@ -61,7 +56,7 @@ class HitTestingSegment(ResponseModel):
     @field_validator("disabled_at", "created_at", "indexing_at", "completed_at", "stopped_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class HitTestingChildChunk(ResponseModel):

api/controllers/console/explore/installed_app.py

@@ -16,6 +16,7 @@ from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.file import helpers as file_helpers
 from libs.datetime_utils import naive_utc_now
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import App, InstalledApp, RecommendedApp
 from models.model import IconType
@@ -105,9 +106,7 @@ class InstalledAppResponse(ResponseModel):
     @field_validator("last_used_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class InstalledAppListResponse(ResponseModel):

api/controllers/console/extension.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field, TypeAdapter, field_validator
 
 from constants import HIDDEN_VALUE
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.api_based_extension import APIBasedExtension
 from services.api_based_extension_service import APIBasedExtensionService
@@ -40,12 +41,6 @@ def _mask_api_key(api_key: str) -> str:
     return api_key[:3] + "******" + api_key[-3:]
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class APIBasedExtensionResponse(ResponseModel):
     id: str
     name: str
@@ -61,7 +56,7 @@ class APIBasedExtensionResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 register_schema_models(console_ns, APIBasedExtensionPayload, CodeBasedExtensionResponse, APIBasedExtensionResponse)

api/controllers/console/workspace/account.py

@@ -42,7 +42,7 @@ from fields.base import ResponseModel
 from fields.member_fields import Account as AccountResponse
 from graphon.file import helpers as file_helpers
 from libs.datetime_utils import naive_utc_now
-from libs.helper import EmailStr, extract_remote_ip, timezone
+from libs.helper import EmailStr, extract_remote_ip, timezone, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import AccountIntegrate, InvitationCode
 from models.account import AccountStatus, InvitationCodeStatus
@@ -185,12 +185,6 @@ def _serialize_account(account) -> dict[str, Any]:
     return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class AccountIntegrateResponse(ResponseModel):
     provider: str
     created_at: int | None = None
@@ -200,7 +194,7 @@ class AccountIntegrateResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AccountIntegrateListResponse(ResponseModel):
@@ -220,7 +214,7 @@ class EducationStatusResponse(ResponseModel):
     @field_validator("expire_at", mode="before")
     @classmethod
     def _normalize_expire_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class EducationAutocompleteResponse(ResponseModel):

api/controllers/console/workspace/workspace.py

@@ -29,7 +29,7 @@ from controllers.console.wraps import (
 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from libs.helper import TimestampField
+from libs.helper import TimestampField, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.account import Tenant, TenantCustomConfigDict, TenantStatus
 from services.account_service import TenantService
@@ -86,9 +86,7 @@ class TenantInfoResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None):
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 register_schema_models(

api/controllers/service_api/app/conversation.py

@@ -22,7 +22,7 @@ from fields.conversation_fields import (
     SimpleConversation,
 )
 from graphon.variables.types import SegmentType
-from libs.helper import UUIDStrOrEmpty
+from libs.helper import UUIDStrOrEmpty, to_timestamp
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService
 
@@ -115,9 +115,7 @@ class ConversationVariableResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class ConversationVariableInfiniteScrollPaginationResponse(ResponseModel):

api/controllers/service_api/app/human_input_form.py

@@ -7,18 +7,18 @@ paused human input forms in workflow/chatflow runs.
 
 import json
 import logging
-from datetime import datetime
 
 from flask import Response
 from flask_restx import Resource
 from werkzeug.exceptions import BadRequest, NotFound
 
-from controllers.common.human_input import HumanInputFormSubmitPayload
+from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
 from controllers.common.schema import register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from core.workflow.human_input_policy import HumanInputSurface, is_recipient_type_allowed_for_surface
 from extensions.ext_database import db
+from libs.helper import to_timestamp
 from models.model import App, EndUser
 from services.human_input_service import Form, FormNotFoundError, HumanInputService
 
@@ -28,30 +28,14 @@ logger = logging.getLogger(__name__)
 register_schema_models(service_api_ns, HumanInputFormSubmitPayload)
 
 
-def _stringify_default_values(values: dict[str, object]) -> dict[str, str]:
-    result: dict[str, str] = {}
-    for key, value in values.items():
-        if value is None:
-            result[key] = ""
-        elif isinstance(value, (dict, list)):
-            result[key] = json.dumps(value, ensure_ascii=False)
-        else:
-            result[key] = str(value)
-    return result
-
-
-def _to_timestamp(value: datetime) -> int:
-    return int(value.timestamp())
-
-
 def _jsonify_form_definition(form: Form) -> Response:
     definition_payload = form.get_definition().model_dump()
     payload = {
         "form_content": definition_payload["rendered_content"],
         "inputs": definition_payload["inputs"],
-        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
+        "resolved_default_values": stringify_form_default_values(definition_payload["default_values"]),
         "user_actions": definition_payload["user_actions"],
-        "expiration_time": _to_timestamp(form.expiration_time),
+        "expiration_time": to_timestamp(form.expiration_time),
     }
     return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
 

api/controllers/service_api/app/workflow.py

@@ -39,6 +39,7 @@ from graphon.enums import WorkflowExecutionStatus
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
+from libs.helper import to_timestamp
 from models.model import App, AppMode, EndUser
 from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
@@ -68,12 +69,6 @@ class WorkflowLogQuery(BaseModel):
 register_schema_models(service_api_ns, WorkflowRunPayload, WorkflowLogQuery)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 def _enum_value(value):
     return getattr(value, "value", value)
 
@@ -109,7 +104,7 @@ class WorkflowRunResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowRunForLogResponse(ResponseModel):
@@ -133,7 +128,7 @@ class WorkflowRunForLogResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPartialResponse(ResponseModel):
@@ -154,7 +149,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/controllers/web/human_input_form.py

@@ -4,7 +4,6 @@ Web App Human Input Form APIs.
 
 import json
 import logging
-from datetime import datetime
 from typing import Any, NotRequired, TypedDict
 
 from flask import Response, request
@@ -13,12 +12,12 @@ from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.common.human_input import HumanInputFormSubmitPayload
+from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
 from controllers.web import web_ns
 from controllers.web.error import NotFoundError, WebFormRateLimitExceededError
 from controllers.web.site import serialize_app_site_payload
 from extensions.ext_database import db
-from libs.helper import RateLimiter, extract_remote_ip
+from libs.helper import RateLimiter, extract_remote_ip, to_timestamp
 from models.account import TenantStatus
 from models.model import App, Site
 from services.human_input_service import Form, FormNotFoundError, HumanInputService
@@ -38,22 +37,6 @@ _FORM_ACCESS_RATE_LIMITER = RateLimiter(
 )
 
 
-def _stringify_default_values(values: dict[str, object]) -> dict[str, str]:
-    result: dict[str, str] = {}
-    for key, value in values.items():
-        if value is None:
-            result[key] = ""
-        elif isinstance(value, (dict, list)):
-            result[key] = json.dumps(value, ensure_ascii=False)
-        else:
-            result[key] = str(value)
-    return result
-
-
-def _to_timestamp(value: datetime) -> int:
-    return int(value.timestamp())
-
-
 class FormDefinitionPayload(TypedDict):
     form_content: Any
     inputs: Any
@@ -69,9 +52,9 @@ def _jsonify_form_definition(form: Form, site_payload: dict | None = None) -> Re
     payload: FormDefinitionPayload = {
         "form_content": definition_payload["rendered_content"],
         "inputs": definition_payload["inputs"],
-        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
+        "resolved_default_values": stringify_form_default_values(definition_payload["default_values"]),
         "user_actions": definition_payload["user_actions"],
-        "expiration_time": _to_timestamp(form.expiration_time),
+        "expiration_time": to_timestamp(form.expiration_time),
     }
     if site_payload is not None:
         payload["site"] = site_payload

api/fields/annotation_fields.py

@@ -5,12 +5,7 @@ from datetime import datetime
 from pydantic import Field, field_validator
 
 from fields.base import ResponseModel
-
-
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
+from libs.helper import to_timestamp
 
 
 class Annotation(ResponseModel):
@@ -23,7 +18,7 @@ class Annotation(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AnnotationList(ResponseModel):
@@ -50,7 +45,7 @@ class AnnotationHitHistory(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AnnotationHitHistoryList(ResponseModel):

api/fields/conversation_fields.py

@@ -7,6 +7,7 @@ from pydantic import Field, field_validator, model_validator
 
 from fields.base import ResponseModel
 from graphon.file import File
+from libs.helper import to_timestamp
 
 type JSONValue = Any
 
@@ -47,9 +48,7 @@ class SimpleConversation(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationInfiniteScrollPagination(ResponseModel):
@@ -90,9 +89,7 @@ class ConversationAnnotation(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationAnnotationHitHistory(ResponseModel):
@@ -103,9 +100,7 @@ class ConversationAnnotationHitHistory(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class AgentThought(ResponseModel):
@@ -125,9 +120,7 @@ class AgentThought(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
     @model_validator(mode="after")
     def _fallback_chain_id(self):
@@ -169,9 +162,7 @@ class MessageDetail(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class FeedbackStat(ResponseModel):
@@ -237,9 +228,7 @@ class Conversation(ResponseModel):
     @field_validator("read_at", "created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationPagination(ResponseModel):
@@ -263,9 +252,7 @@ class ConversationMessageDetail(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationWithSummary(ResponseModel):
@@ -291,9 +278,7 @@ class ConversationWithSummary(ResponseModel):
     @field_validator("read_at", "created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationWithSummaryPagination(ResponseModel):
@@ -322,15 +307,7 @@ class ConversationDetail(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
-
-
-def to_timestamp(value: datetime | None) -> int | None:
-    if value is None:
-        return None
-    return int(value.timestamp())
+        return to_timestamp(value)
 
 
 def format_files_contained(value: JSONValue) -> JSONValue:

api/fields/conversation_variable_fields.py

@@ -8,7 +8,7 @@ from pydantic import field_validator
 
 from fields.base import ResponseModel
 from graphon.variables.types import SegmentType
-from libs.helper import TimestampField
+from libs.helper import TimestampField, to_timestamp
 
 from ._value_type_serializer import serialize_value_type
 
@@ -37,12 +37,6 @@ conversation_variable_infinite_scroll_pagination_fields = {
 }
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class ConversationVariableResponse(ResponseModel):
     id: str
     name: str
@@ -88,7 +82,7 @@ class ConversationVariableResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class PaginatedConversationVariableResponse(ResponseModel):

api/fields/file_fields.py

@@ -5,12 +5,7 @@ from datetime import datetime
 from pydantic import field_validator
 
 from fields.base import ResponseModel
-
-
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
+from libs.helper import to_timestamp
 
 
 class UploadConfig(ResponseModel):
@@ -45,7 +40,7 @@ class FileResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class RemoteFileInfo(ResponseModel):
@@ -66,7 +61,7 @@ class FileWithSignedUrl(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 __all__ = [

api/fields/member_fields.py

@@ -7,6 +7,7 @@ from pydantic import computed_field, field_validator
 
 from fields.base import ResponseModel
 from graphon.file import helpers as file_helpers
+from libs.helper import to_timestamp
 
 simple_account_fields = {
     "id": fields.String,
@@ -15,12 +16,6 @@ simple_account_fields = {
 }
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 def _build_avatar_url(avatar: str | None) -> str | None:
     if avatar is None:
         return None
@@ -59,7 +54,7 @@ class Account(_AccountAvatar):
     @field_validator("last_login_at", "created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AccountWithRole(_AccountAvatar):
@@ -75,7 +70,7 @@ class AccountWithRole(_AccountAvatar):
     @field_validator("last_login_at", "last_active_at", "created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AccountWithRoleList(ResponseModel):

api/fields/message_fields.py

@@ -9,6 +9,7 @@ from core.entities.execution_extra_content import ExecutionExtraContentDomainMod
 from fields.base import ResponseModel
 from fields.conversation_fields import AgentThought, JSONValue, MessageFile
 from graphon.file import File
+from libs.helper import to_timestamp
 
 type JSONValueType = JSONValue
 
@@ -39,9 +40,7 @@ class RetrieverResource(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class MessageListItem(ResponseModel):
@@ -68,9 +67,7 @@ class MessageListItem(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class WebMessageListItem(MessageListItem):
@@ -106,9 +103,7 @@ class SavedMessageItem(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class SavedMessageInfiniteScrollPagination(ResponseModel):
@@ -121,12 +116,6 @@ class SuggestedQuestionsResponse(ResponseModel):
     data: list[str]
 
 
-def to_timestamp(value: datetime | None) -> int | None:
-    if value is None:
-        return None
-    return int(value.timestamp())
-
-
 def format_files_contained(value: JSONValueType) -> JSONValueType:
     if isinstance(value, File):
         # Response payloads must preserve legacy file keys like `related_id`/`url`

api/fields/workflow_app_log_fields.py

@@ -17,7 +17,7 @@ from fields.workflow_run_fields import (
     workflow_run_for_archived_log_fields,
     workflow_run_for_log_fields,
 )
-from libs.helper import TimestampField
+from libs.helper import TimestampField, to_timestamp
 
 workflow_app_log_partial_fields = {
     "id": fields.String,
@@ -96,12 +96,6 @@ def build_workflow_archived_log_pagination_model(api_or_ns: Namespace):
     return api_or_ns.model("WorkflowArchivedLogPagination", copied_fields)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class WorkflowAppLogPartialResponse(ResponseModel):
     id: str
     workflow_run: WorkflowRunForLogResponse | None = None
@@ -115,7 +109,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowArchivedLogPartialResponse(ResponseModel):
@@ -129,7 +123,7 @@ class WorkflowArchivedLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/fields/workflow_run_fields.py

@@ -16,7 +16,7 @@ from pydantic import AliasChoices, Field, field_validator
 from fields.base import ResponseModel
 from fields.end_user_fields import SimpleEndUser
 from fields.member_fields import SimpleAccount
-from libs.helper import TimestampField
+from libs.helper import TimestampField, to_timestamp
 
 workflow_run_for_log_fields = {
     "id": fields.String,
@@ -50,12 +50,6 @@ def build_workflow_run_for_archived_log_model(api_or_ns: Namespace):
     return api_or_ns.model("WorkflowRunForArchivedLog", workflow_run_for_archived_log_fields)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class WorkflowRunForLogResponse(ResponseModel):
     id: str
     version: str | None = None
@@ -79,7 +73,7 @@ class WorkflowRunForLogResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowRunForArchivedLogResponse(ResponseModel):
@@ -120,7 +114,7 @@ class WorkflowRunForListResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AdvancedChatWorkflowRunForListResponse(WorkflowRunForListResponse):
@@ -180,7 +174,7 @@ class WorkflowRunDetailResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowRunNodeExecutionResponse(ResponseModel):
@@ -217,7 +211,7 @@ class WorkflowRunNodeExecutionResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowRunNodeExecutionListResponse(ResponseModel):

api/libs/helper.py

@@ -10,7 +10,7 @@ import uuid
 from collections.abc import Callable, Generator, Mapping
 from datetime import datetime
 from hashlib import sha256
-from typing import TYPE_CHECKING, Annotated, Any, Protocol, cast
+from typing import TYPE_CHECKING, Annotated, Any, Protocol, cast, overload
 from uuid import UUID
 from zoneinfo import available_timezones
 
@@ -162,6 +162,30 @@ class OptionalTimestampField(fields.Raw):
         return int(value.timestamp())
 
 
+@overload
+def to_timestamp(value: datetime) -> int: ...
+
+
+@overload
+def to_timestamp(value: int) -> int: ...
+
+
+@overload
+def to_timestamp(value: None) -> None: ...
+
+
+def to_timestamp(value: datetime | int | None) -> int | None:
+    """Normalize API response timestamp values to epoch seconds."""
+    if isinstance(value, datetime):
+        return int(value.timestamp())
+    return value
+
+
+def current_timestamp() -> int:
+    """Return the current Unix timestamp in seconds."""
+    return int(time.time())
+
+
 def email(email):
     # Define a regex pattern for email addresses
     pattern = r"^[\w\.!#$%&'*+\-/=?^_`{|}~]+@([\w-]+\.)+[\w-]{2,}$"

api/schedule/trigger_provider_refresh_task.py

@@ -1,6 +1,5 @@
 import logging
 import math
-import time
 from collections.abc import Iterable, Sequence
 
 from celery import group
@@ -13,16 +12,13 @@ from configs import dify_config
 from core.trigger.utils.locks import build_trigger_refresh_lock_keys
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
+from libs.helper import current_timestamp
 from models.trigger import TriggerSubscription
 from tasks.trigger_subscription_refresh_tasks import trigger_subscription_refresh
 
 logger = logging.getLogger(__name__)
 
 
-def _now_ts() -> int:
-    return int(time.time())
-
-
 def _build_due_filter(now_ts: int):
     """Build SQLAlchemy filter for due credential or subscription refresh."""
     credential_due: ColumnElement[bool] = and_(
@@ -54,7 +50,7 @@ def trigger_provider_refresh() -> None:
     """
     Scan due trigger subscriptions and enqueue refresh tasks with in-flight locks.
     """
-    now: int = _now_ts()
+    now: int = current_timestamp()
 
     batch_size: int = int(dify_config.TRIGGER_PROVIDER_REFRESH_BATCH_SIZE)
     lock_ttl: int = max(300, int(dify_config.TRIGGER_PROVIDER_SUBSCRIPTION_THRESHOLD_SECONDS))

api/services/app_dsl_service.py

@@ -10,7 +10,6 @@ from uuid import uuid4
 import yaml
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import pad, unpad
-from packaging import version
 from packaging.version import parse as parse_version
 from pydantic import BaseModel
 from sqlalchemy import select
@@ -40,6 +39,7 @@ from libs.datetime_utils import naive_utc_now
 from models import Account, App, AppMode
 from models.model import AppModelConfig, AppModelConfigDict, IconType
 from models.workflow import Workflow
+from services.dsl_version import check_version_compatibility
 from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus
 from services.plugin.dependencies_analysis import DependenciesAnalysisService
 from services.workflow_draft_variable_service import WorkflowDraftVariableService
@@ -64,30 +64,6 @@ class Import(BaseModel):
     error: str = ""
 
 
-def _check_version_compatibility(imported_version: str) -> ImportStatus:
-    """Determine import status based on version comparison"""
-    try:
-        current_ver = version.parse(CURRENT_DSL_VERSION)
-        imported_ver = version.parse(imported_version)
-    except version.InvalidVersion:
-        return ImportStatus.FAILED
-
-    # If imported version is newer than current, always return PENDING
-    if imported_ver > current_ver:
-        return ImportStatus.PENDING
-
-    # If imported version is older than current's major, return PENDING
-    if imported_ver.major < current_ver.major:
-        return ImportStatus.PENDING
-
-    # If imported version is older than current's minor, return COMPLETED_WITH_WARNINGS
-    if imported_ver.minor < current_ver.minor:
-        return ImportStatus.COMPLETED_WITH_WARNINGS
-
-    # If imported version equals or is older than current's micro, return COMPLETED
-    return ImportStatus.COMPLETED
-
-
 class PendingData(BaseModel):
     import_mode: str
     yaml_content: str
@@ -203,7 +179,7 @@ class AppDslService:
             # check if imported_version is a float-like string
             if not isinstance(imported_version, str):
                 raise ValueError(f"Invalid version type, expected str, got {type(imported_version)}")
-            status = _check_version_compatibility(imported_version)
+            status = check_version_compatibility(imported_version, CURRENT_DSL_VERSION)
 
             # Extract app data
             app_data = data.get("app")

api/services/dsl_version.py

@@ -0,0 +1,20 @@
+from packaging import version
+
+from services.entities.dsl_entities import ImportStatus
+
+
+def check_version_compatibility(imported_version: str, current_version: str) -> ImportStatus:
+    """Determine DSL import status based on imported and current versions."""
+    try:
+        current_ver = version.parse(current_version)
+        imported_ver = version.parse(imported_version)
+    except version.InvalidVersion:
+        return ImportStatus.FAILED
+
+    if imported_ver > current_ver:
+        return ImportStatus.PENDING
+    if imported_ver.major < current_ver.major:
+        return ImportStatus.PENDING
+    if imported_ver.minor < current_ver.minor:
+        return ImportStatus.COMPLETED_WITH_WARNINGS
+    return ImportStatus.COMPLETED

api/services/rag_pipeline/rag_pipeline_dsl_service.py

@@ -13,7 +13,6 @@ import yaml  # type: ignore
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import pad, unpad
 from flask_login import current_user
-from packaging import version
 from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -37,6 +36,7 @@ from models import Account
 from models.dataset import Dataset, DatasetCollectionBinding, Pipeline
 from models.enums import CollectionBindingType, DatasetRuntimeMode
 from models.workflow import Workflow, WorkflowType
+from services.dsl_version import check_version_compatibility
 from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus
 from services.entities.knowledge_entities.rag_pipeline_entities import (
     IconInfo,
@@ -64,30 +64,6 @@ class RagPipelineImportInfo(BaseModel):
     dataset_id: str | None = None
 
 
-def _check_version_compatibility(imported_version: str) -> ImportStatus:
-    """Determine import status based on version comparison"""
-    try:
-        current_ver = version.parse(CURRENT_DSL_VERSION)
-        imported_ver = version.parse(imported_version)
-    except version.InvalidVersion:
-        return ImportStatus.FAILED
-
-    # If imported version is newer than current, always return PENDING
-    if imported_ver > current_ver:
-        return ImportStatus.PENDING
-
-    # If imported version is older than current's major, return PENDING
-    if imported_ver.major < current_ver.major:
-        return ImportStatus.PENDING
-
-    # If imported version is older than current's minor, return COMPLETED_WITH_WARNINGS
-    if imported_ver.minor < current_ver.minor:
-        return ImportStatus.COMPLETED_WITH_WARNINGS
-
-    # If imported version equals or is older than current's micro, return COMPLETED
-    return ImportStatus.COMPLETED
-
-
 class RagPipelinePendingData(BaseModel):
     import_mode: str
     yaml_content: str
@@ -202,7 +178,7 @@ class RagPipelineDslService:
             # check if imported_version is a float-like string
             if not isinstance(imported_version, str):
                 raise ValueError(f"Invalid version type, expected str, got {type(imported_version)}")
-            status = _check_version_compatibility(imported_version)
+            status = check_version_compatibility(imported_version, CURRENT_DSL_VERSION)
 
             # Extract app data
             pipeline_data = data.get("rag_pipeline")

api/tasks/trigger_subscription_refresh_tasks.py

@@ -1,5 +1,4 @@
 import logging
-import time
 from collections.abc import Mapping
 from typing import Any
 
@@ -12,16 +11,13 @@ from core.db.session_factory import session_factory
 from core.plugin.entities.plugin_daemon import CredentialType
 from core.trigger.utils.locks import build_trigger_refresh_lock_key
 from extensions.ext_redis import redis_client
+from libs.helper import current_timestamp
 from models.trigger import TriggerSubscription
 from services.trigger.trigger_provider_service import TriggerProviderService
 
 logger = logging.getLogger(__name__)
 
 
-def _now_ts() -> int:
-    return int(time.time())
-
-
 def _load_subscription(session: Session, tenant_id: str, subscription_id: str) -> TriggerSubscription | None:
     return session.scalar(
         select(TriggerSubscription)
@@ -96,7 +92,7 @@ def trigger_subscription_refresh(tenant_id: str, subscription_id: str) -> None:
 
     logger.info("Begin subscription refresh: tenant=%s id=%s", tenant_id, subscription_id)
     try:
-        now: int = _now_ts()
+        now: int = current_timestamp()
         with session_factory.create_session() as session:
             subscription: TriggerSubscription | None = _load_subscription(session, tenant_id, subscription_id)
 

api/tests/test_containers_integration_tests/services/test_app_dsl_service.py

@@ -35,9 +35,9 @@ from services.app_dsl_service import (
     ImportMode,
     ImportStatus,
     PendingData,
-    _check_version_compatibility,
 )
 from services.app_service import AppService, CreateAppParams
+from services.dsl_version import check_version_compatibility
 from tests.test_containers_integration_tests.helpers import generate_valid_password
 
 _DEFAULT_TENANT_ID = "00000000-0000-0000-0000-000000000001"
@@ -193,22 +193,25 @@ class TestAppDslService:
     # ── Version Compatibility ─────────────────────────────────────────
 
     def test_check_version_compatibility_invalid_version_returns_failed(self):
-        assert _check_version_compatibility("not-a-version") == ImportStatus.FAILED
+        assert check_version_compatibility("not-a-version", app_dsl_service.CURRENT_DSL_VERSION) == ImportStatus.FAILED
 
     def test_check_version_compatibility_newer_version_returns_pending(self):
-        assert _check_version_compatibility("99.0.0") == ImportStatus.PENDING
+        assert check_version_compatibility("99.0.0", app_dsl_service.CURRENT_DSL_VERSION) == ImportStatus.PENDING
 
     def test_check_version_compatibility_major_older_returns_pending(self, monkeypatch: pytest.MonkeyPatch):
         monkeypatch.setattr(app_dsl_service, "CURRENT_DSL_VERSION", "1.0.0")
-        assert _check_version_compatibility("0.9.9") == ImportStatus.PENDING
+        assert check_version_compatibility("0.9.9", app_dsl_service.CURRENT_DSL_VERSION) == ImportStatus.PENDING
 
     def test_check_version_compatibility_minor_older_returns_completed_with_warnings(
         self,
     ):
-        assert _check_version_compatibility("0.5.0") == ImportStatus.COMPLETED_WITH_WARNINGS
+        assert (
+            check_version_compatibility("0.5.0", app_dsl_service.CURRENT_DSL_VERSION)
+            == ImportStatus.COMPLETED_WITH_WARNINGS
+        )
 
     def test_check_version_compatibility_equal_returns_completed(self):
-        assert _check_version_compatibility(CURRENT_DSL_VERSION) == ImportStatus.COMPLETED
+        assert check_version_compatibility(CURRENT_DSL_VERSION, CURRENT_DSL_VERSION) == ImportStatus.COMPLETED
 
     # ── Import: Validation ────────────────────────────────────────────
 

api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py

@@ -8,11 +8,12 @@ from sqlalchemy.orm import Session
 
 from core.workflow.nodes.knowledge_index import KNOWLEDGE_INDEX_NODE_TYPE
 from graphon.enums import BuiltinNodeTypes
+from services.dsl_version import check_version_compatibility
 from services.entities.knowledge_entities.rag_pipeline_entities import IconInfo, RagPipelineDatasetCreateEntity
+from services.rag_pipeline import rag_pipeline_dsl_service
 from services.rag_pipeline.rag_pipeline_dsl_service import (
     ImportStatus,
     RagPipelineDslService,
-    _check_version_compatibility,
 )
 
 
@@ -26,7 +27,9 @@ from services.rag_pipeline.rag_pipeline_dsl_service import (
     ],
 )
 def test_check_version_compatibility(imported_version: str, expected_status: ImportStatus) -> None:
-    assert _check_version_compatibility(imported_version) == expected_status
+    assert (
+        check_version_compatibility(imported_version, rag_pipeline_dsl_service.CURRENT_DSL_VERSION) == expected_status
+    )
 
 
 def test_encrypt_decrypt_dataset_id_roundtrip() -> None:
@@ -1101,7 +1104,7 @@ def test_extract_dependencies_from_model_config_includes_dataset_reranking_and_t
 def test_check_version_compatibility_hits_major_older_branch(mocker) -> None:
     mocker.patch("services.rag_pipeline.rag_pipeline_dsl_service.CURRENT_DSL_VERSION", "1.0.0")
 
-    status = _check_version_compatibility("0.9.0")
+    status = check_version_compatibility("0.9.0", rag_pipeline_dsl_service.CURRENT_DSL_VERSION)
 
     assert status == ImportStatus.PENDING