From 1c2bbed405bac45ba4852145c9dbba3a03ea1303 Mon Sep 17 00:00:00 2001 From: wdeveloper16 Date: Tue, 14 Apr 2026 05:02:34 +0200 Subject: [PATCH] refactor: replace bare dict with dict[str, Any] in services grab-bag (#35112) --- api/services/app_dsl_service.py | 8 ++++---- api/services/app_service.py | 6 +++--- api/services/billing_service.py | 4 ++-- api/services/hit_testing_service.py | 6 +++--- api/services/model_load_balancing_service.py | 4 ++-- api/services/recommend_app/buildin/buildin_retrieval.py | 5 +++-- api/services/recommend_app/remote/remote_retrieval.py | 7 ++++--- api/services/recommended_app_service.py | 6 ++++-- 8 files changed, 25 insertions(+), 21 deletions(-) diff --git a/api/services/app_dsl_service.py b/api/services/app_dsl_service.py index 40e1e5f8ab8..87c28980ab0 100644 --- a/api/services/app_dsl_service.py +++ b/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 diff --git a/api/services/app_service.py b/api/services/app_service.py index 87d52a3159c..ef170c50ba0 100644 --- a/api/services/app_service.py +++ b/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 diff --git a/api/services/billing_service.py b/api/services/billing_service.py index 735b22aa4c7..a1362ccad6b 100644 --- a/api/services/billing_service.py +++ b/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, diff --git a/api/services/hit_testing_service.py b/api/services/hit_testing_service.py index 4a21e3c5bd8..43985e49cd0 100644 --- a/api/services/hit_testing_service.py +++ b/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, - metadata_filtering_conditions: dict | None = None, + external_retrieval_model: dict[str, Any] | None = None, + metadata_filtering_conditions: dict[str, Any] | None = None, ): if dataset.provider != "external": return { diff --git a/api/services/model_load_balancing_service.py b/api/services/model_load_balancing_service.py index 41b6b885b29..b652e049ce8 100644 --- a/api/services/model_load_balancing_service.py +++ b/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, diff --git a/api/services/recommend_app/buildin/buildin_retrieval.py b/api/services/recommend_app/buildin/buildin_retrieval.py index 64751d186ce..16dc66cd767 100644 --- a/api/services/recommend_app/buildin/buildin_retrieval.py +++ b/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 diff --git a/api/services/recommend_app/remote/remote_retrieval.py b/api/services/recommend_app/remote/remote_retrieval.py index b217c9026ae..5818be04805 100644 --- a/api/services/recommend_app/remote/remote_retrieval.py +++ b/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"]) diff --git a/api/services/recommended_app_service.py b/api/services/recommended_app_service.py index 9819822103b..134dd37a3ed 100644 --- a/api/services/recommended_app_service.py +++ b/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))