refactor: replace bare dict with dict[str, Any] in services grab-bag (#35112)

api/services/app_dsl_service.py

@@ -3,7 +3,7 @@ import hashlib
 import logging
 import uuid
 from collections.abc import Mapping
-from typing import cast
+from typing import Any, cast
 from urllib.parse import urlparse
 from uuid import uuid4
 
@@ -400,7 +400,7 @@ class AppDslService:
         self,
         *,
         app: App | None,
-        data: dict,
+        data: dict[str, Any],
         account: Account,
         name: str | None = None,
         description: str | None = None,
@@ -567,7 +567,7 @@ class AppDslService:
 
     @classmethod
     def _append_workflow_export_data(
-        cls, *, export_data: dict, app_model: App, include_secret: bool, workflow_id: str | None = None
+        cls, *, export_data: dict[str, Any], app_model: App, include_secret: bool, workflow_id: str | None = None
     ):
         """
         Append workflow export data
@@ -620,7 +620,7 @@ class AppDslService:
         ]
 
     @classmethod
-    def _append_model_config_export_data(cls, export_data: dict, app_model: App):
+    def _append_model_config_export_data(cls, export_data: dict[str, Any], app_model: App):
         """
         Append model config export data
         :param export_data: export data

api/services/app_service.py

@@ -32,7 +32,7 @@ logger = logging.getLogger(__name__)
 
 
 class AppService:
-    def get_paginate_apps(self, user_id: str, tenant_id: str, args: dict) -> Pagination | None:
+    def get_paginate_apps(self, user_id: str, tenant_id: str, args: dict[str, Any]) -> Pagination | None:
         """
         Get app list with pagination
         :param user_id: user id
@@ -78,7 +78,7 @@ class AppService:
 
         return app_models
 
-    def create_app(self, tenant_id: str, args: dict, account: Account) -> App:
+    def create_app(self, tenant_id: str, args: dict[str, Any], account: Account) -> App:
         """
         Create app
         :param tenant_id: tenant id
@@ -389,7 +389,7 @@ class AppService:
         """
         app_mode = AppMode.value_of(app_model.mode)
 
-        meta: dict = {"tool_icons": {}}
+        meta: dict[str, Any] = {"tool_icons": {}}
 
         if app_mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
             workflow = app_model.workflow

api/services/billing_service.py

@@ -2,7 +2,7 @@ import json
 import logging
 import os
 from collections.abc import Sequence
-from typing import Literal, NotRequired, TypedDict
+from typing import Any, Literal, NotRequired, TypedDict
 
 import httpx
 from pydantic import TypeAdapter
@@ -541,7 +541,7 @@ class BillingService:
         start_time / end_time: RFC3339 strings (e.g. "2026-03-01T00:00:00Z"), optional.
         Returns {"notification_id": str}.
         """
-        payload: dict = {
+        payload: dict[str, Any] = {
             "contents": contents,
             "frequency": frequency,
             "status": status,

api/services/hit_testing_service.py

@@ -45,7 +45,7 @@ class HitTestingService:
         query: str,
         account: Account,
         retrieval_model: dict[str, Any] | None,
-        external_retrieval_model: dict,
+        external_retrieval_model: dict[str, Any],
         attachment_ids: list | None = None,
         limit: int = 10,
     ):
@@ -125,8 +125,8 @@ class HitTestingService:
         dataset: Dataset,
         query: str,
         account: Account,
-        external_retrieval_model: dict | None = None,
+        external_retrieval_model: dict[str, Any] | None = None,
-        metadata_filtering_conditions: dict | None = None,
+        metadata_filtering_conditions: dict[str, Any] | None = None,
     ):
         if dataset.provider != "external":
             return {

api/services/model_load_balancing_service.py

@@ -502,7 +502,7 @@ class ModelLoadBalancingService:
         provider: str,
         model: str,
         model_type: str,
-        credentials: dict,
+        credentials: dict[str, Any],
         config_id: str | None = None,
     ):
         """
@@ -561,7 +561,7 @@ class ModelLoadBalancingService:
         provider_configuration: ProviderConfiguration,
         model_type: ModelType,
         model: str,
-        credentials: dict,
+        credentials: dict[str, Any],
         load_balancing_model_config: LoadBalancingModelConfig | None = None,
         model_provider_factory: ModelProviderFactory | None = None,
         validate: bool = True,

api/services/recommend_app/buildin/buildin_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,7 +14,7 @@ class BuildInRecommendAppRetrieval(RecommendAppRetrievalBase):
     Retrieval recommended app from buildin, the location  is constants/recommended_apps.json
     """
 
-    builtin_data: dict | None = None
+    builtin_data: dict[str, Any] | None = None
 
     def get_type(self) -> str:
         return RecommendAppType.BUILDIN
@@ -53,7 +54,7 @@ class BuildInRecommendAppRetrieval(RecommendAppRetrievalBase):
         return builtin_data.get("recommended_apps", {}).get(language, {})
 
     @classmethod
-    def fetch_recommended_app_detail_from_builtin(cls, app_id: str) -> dict | None:
+    def fetch_recommended_app_detail_from_builtin(cls, app_id: str) -> dict[str, Any] | None:
         """
         Fetch recommended app detail from builtin.
         :param app_id: App ID

api/services/recommend_app/remote/remote_retrieval.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any
 
 import httpx
 
@@ -35,7 +36,7 @@ class RemoteRecommendAppRetrieval(RecommendAppRetrievalBase):
         return RecommendAppType.REMOTE
 
     @classmethod
-    def fetch_recommended_app_detail_from_dify_official(cls, app_id: str) -> dict | None:
+    def fetch_recommended_app_detail_from_dify_official(cls, app_id: str) -> dict[str, Any] | None:
         """
         Fetch recommended app detail from dify official.
         :param app_id: App ID
@@ -46,7 +47,7 @@ class RemoteRecommendAppRetrieval(RecommendAppRetrievalBase):
         response = httpx.get(url, timeout=httpx.Timeout(10.0, connect=3.0))
         if response.status_code != 200:
             return None
-        data: dict = response.json()
+        data: dict[str, Any] = response.json()
         return data
 
     @classmethod
@@ -62,7 +63,7 @@ class RemoteRecommendAppRetrieval(RecommendAppRetrievalBase):
         if response.status_code != 200:
             raise ValueError(f"fetch recommended apps failed, status code: {response.status_code}")
 
-        result: dict = response.json()
+        result: dict[str, Any] = response.json()
 
         if "categories" in result:
             result["categories"] = sorted(result["categories"])

api/services/recommended_app_service.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from sqlalchemy import select
 
 from configs import dify_config
@@ -37,7 +39,7 @@ class RecommendedAppService:
         return result
 
     @classmethod
-    def get_recommend_app_detail(cls, app_id: str) -> dict | None:
+    def get_recommend_app_detail(cls, app_id: str) -> dict[str, Any] | None:
         """
         Get recommend app detail.
         :param app_id: app id
@@ -45,7 +47,7 @@ class RecommendedAppService:
         """
         mode = dify_config.HOSTED_FETCH_APP_TEMPLATES_MODE
         retrieval_instance = RecommendAppRetrievalFactory.get_recommend_app_factory(mode)()
-        result: dict = retrieval_instance.get_recommend_app_detail(app_id)
+        result: dict[str, Any] = retrieval_instance.get_recommend_app_detail(app_id)
         if FeatureService.get_system_features().enable_trial_app:
             app_id = result["id"]
             trial_app_model = db.session.scalar(select(TrialApp).where(TrialApp.app_id == app_id).limit(1))