refactor: replace bare dict with dict[str, Any] in pipeline_template module (#35071)

api/services/rag_pipeline/pipeline_template/built_in/built_in_retrieval.py

@@ -1,6 +1,7 @@
 import json
 from os import path
 from pathlib import Path
+from typing import Any
 
 from flask import current_app
 
@@ -13,21 +14,21 @@ class BuiltInPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
     Retrieval pipeline template from built-in, the location  is constants/pipeline_templates.json
     """
 
-    builtin_data: dict | None = None
+    builtin_data: dict[str, Any] | None = None
 
     def get_type(self) -> str:
         return PipelineTemplateType.BUILTIN
 
-    def get_pipeline_templates(self, language: str) -> dict:
+    def get_pipeline_templates(self, language: str) -> dict[str, Any]:
         result = self.fetch_pipeline_templates_from_builtin(language)
         return result
 
-    def get_pipeline_template_detail(self, template_id: str):
+    def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
         result = self.fetch_pipeline_template_detail_from_builtin(template_id)
         return result
 
     @classmethod
-    def _get_builtin_data(cls) -> dict:
+    def _get_builtin_data(cls) -> dict[str, Any]:
         """
         Get builtin data.
         :return:
@@ -43,7 +44,7 @@ class BuiltInPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
         return cls.builtin_data or {}
 
     @classmethod
-    def fetch_pipeline_templates_from_builtin(cls, language: str) -> dict:
+    def fetch_pipeline_templates_from_builtin(cls, language: str) -> dict[str, Any]:
         """
         Fetch pipeline templates from builtin.
         :param language: language
@@ -53,7 +54,7 @@ class BuiltInPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
         return builtin_data.get("pipeline_templates", {}).get(language, {})
 
     @classmethod
-    def fetch_pipeline_template_detail_from_builtin(cls, template_id: str) -> dict | None:
+    def fetch_pipeline_template_detail_from_builtin(cls, template_id: str) -> dict[str, Any] | None:
         """
         Fetch pipeline template detail from builtin.
         :param template_id: Template ID

api/services/rag_pipeline/pipeline_template/customized/customized_retrieval.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 import yaml
 from sqlalchemy import select
 
@@ -13,12 +15,12 @@ class CustomizedPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
     Retrieval recommended app from database
     """
 
-    def get_pipeline_templates(self, language: str) -> dict:
+    def get_pipeline_templates(self, language: str) -> dict[str, Any]:
         _, current_tenant_id = current_account_with_tenant()
         result = self.fetch_pipeline_templates_from_customized(tenant_id=current_tenant_id, language=language)
         return result
 
-    def get_pipeline_template_detail(self, template_id: str):
+    def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
         result = self.fetch_pipeline_template_detail_from_db(template_id)
         return result
 
@@ -26,7 +28,7 @@ class CustomizedPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
         return PipelineTemplateType.CUSTOMIZED
 
     @classmethod
-    def fetch_pipeline_templates_from_customized(cls, tenant_id: str, language: str) -> dict:
+    def fetch_pipeline_templates_from_customized(cls, tenant_id: str, language: str) -> dict[str, Any]:
         """
         Fetch pipeline templates from db.
         :param tenant_id: tenant id
@@ -53,7 +55,7 @@ class CustomizedPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
         return {"pipeline_templates": recommended_pipelines_results}
 
     @classmethod
-    def fetch_pipeline_template_detail_from_db(cls, template_id: str) -> dict | None:
+    def fetch_pipeline_template_detail_from_db(cls, template_id: str) -> dict[str, Any] | None:
         """
         Fetch pipeline template detail from db.
         :param template_id: Template ID

api/services/rag_pipeline/pipeline_template/database/database_retrieval.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 import yaml
 from sqlalchemy import select
 
@@ -12,11 +14,11 @@ class DatabasePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
     Retrieval pipeline   template from database
     """
 
-    def get_pipeline_templates(self, language: str) -> dict:
+    def get_pipeline_templates(self, language: str) -> dict[str, Any]:
         result = self.fetch_pipeline_templates_from_db(language)
         return result
 
-    def get_pipeline_template_detail(self, template_id: str):
+    def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
         result = self.fetch_pipeline_template_detail_from_db(template_id)
         return result
 
@@ -24,7 +26,7 @@ class DatabasePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
         return PipelineTemplateType.DATABASE
 
     @classmethod
-    def fetch_pipeline_templates_from_db(cls, language: str) -> dict:
+    def fetch_pipeline_templates_from_db(cls, language: str) -> dict[str, Any]:
         """
         Fetch pipeline templates from db.
         :param language: language
@@ -54,7 +56,7 @@ class DatabasePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
         return {"pipeline_templates": recommended_pipelines_results}
 
     @classmethod
-    def fetch_pipeline_template_detail_from_db(cls, template_id: str) -> dict | None:
+    def fetch_pipeline_template_detail_from_db(cls, template_id: str) -> dict[str, Any] | None:
         """
         Fetch pipeline template detail from db.
         :param pipeline_id: Pipeline ID

api/services/rag_pipeline/pipeline_template/pipeline_template_base.py

@@ -1,15 +1,16 @@
 from abc import ABC, abstractmethod
+from typing import Any
 
 
 class PipelineTemplateRetrievalBase(ABC):
     """Interface for pipeline template retrieval."""
 
     @abstractmethod
-    def get_pipeline_templates(self, language: str) -> dict:
+    def get_pipeline_templates(self, language: str) -> dict[str, Any]:
         raise NotImplementedError
 
     @abstractmethod
-    def get_pipeline_template_detail(self, template_id: str) -> dict | None:
+    def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
         raise NotImplementedError
 
     @abstractmethod

api/services/rag_pipeline/pipeline_template/remote/remote_retrieval.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any
 
 import httpx
 
@@ -15,8 +16,8 @@ class RemotePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
     Retrieval recommended app from dify official
     """
 
-    def get_pipeline_template_detail(self, template_id: str) -> dict | None:
+    def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
-        result: dict | None
+        result: dict[str, Any] | None
         try:
             result = self.fetch_pipeline_template_detail_from_dify_official(template_id)
         except Exception as e:
@@ -24,7 +25,7 @@ class RemotePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
             result = DatabasePipelineTemplateRetrieval.fetch_pipeline_template_detail_from_db(template_id)
         return result
 
-    def get_pipeline_templates(self, language: str) -> dict:
+    def get_pipeline_templates(self, language: str) -> dict[str, Any]:
         try:
             result = self.fetch_pipeline_templates_from_dify_official(language)
         except Exception as e:
@@ -36,7 +37,7 @@ class RemotePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
         return PipelineTemplateType.REMOTE
 
     @classmethod
-    def fetch_pipeline_template_detail_from_dify_official(cls, template_id: str) -> dict:
+    def fetch_pipeline_template_detail_from_dify_official(cls, template_id: str) -> dict[str, Any]:
         """
         Fetch pipeline template detail from dify official.
 
@@ -53,11 +54,11 @@ class RemotePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
                 + f" status_code: {response.status_code},"
                 + f" response: {response.text[:1000]}"
             )
-        data: dict = response.json()
+        data: dict[str, Any] = response.json()
         return data
 
     @classmethod
-    def fetch_pipeline_templates_from_dify_official(cls, language: str) -> dict:
+    def fetch_pipeline_templates_from_dify_official(cls, language: str) -> dict[str, Any]:
         """
         Fetch pipeline templates from dify official.
         :param language: language
@@ -69,6 +70,6 @@ class RemotePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
         if response.status_code != 200:
             raise ValueError(f"fetch pipeline templates failed, status code: {response.status_code}")
 
-        result: dict = response.json()
+        result: dict[str, Any] = response.json()
 
         return result