refactor(api): replace dict/Mapping with TypedDict in core/app (#33601)

api/core/app/app_config/common/parameters_mapping/__init__.py

@@ -1,13 +1,36 @@
 from collections.abc import Mapping
-from typing import Any
+from typing import Any, TypedDict
 
 from configs import dify_config
 from constants import DEFAULT_FILE_NUMBER_LIMITS
 
 
+class SystemParametersDict(TypedDict):
+    image_file_size_limit: int
+    video_file_size_limit: int
+    audio_file_size_limit: int
+    file_size_limit: int
+    workflow_file_upload_limit: int
+
+
+class AppParametersDict(TypedDict):
+    opening_statement: str | None
+    suggested_questions: list[str]
+    suggested_questions_after_answer: dict[str, Any]
+    speech_to_text: dict[str, Any]
+    text_to_speech: dict[str, Any]
+    retriever_resource: dict[str, Any]
+    annotation_reply: dict[str, Any]
+    more_like_this: dict[str, Any]
+    user_input_form: list[dict[str, Any]]
+    sensitive_word_avoidance: dict[str, Any]
+    file_upload: dict[str, Any]
+    system_parameters: SystemParametersDict
+
+
 def get_parameters_from_feature_dict(
     *, features_dict: Mapping[str, Any], user_input_form: list[dict[str, Any]]
-) -> Mapping[str, Any]:
+) -> AppParametersDict:
     """
     Mapping from feature dict to webapp parameters
     """

api/core/app/apps/common/workflow_response_converter.py

@@ -3,7 +3,7 @@ import time
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, NewType, Union
+from typing import Any, NewType, TypedDict, Union
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -76,6 +76,20 @@ NodeExecutionId = NewType("NodeExecutionId", str)
 logger = logging.getLogger(__name__)
 
 
+class AccountCreatedByDict(TypedDict):
+    id: str
+    name: str
+    email: str
+
+
+class EndUserCreatedByDict(TypedDict):
+    id: str
+    user: str
+
+
+CreatedByDict = AccountCreatedByDict | EndUserCreatedByDict
+
+
 @dataclass(slots=True)
 class _NodeSnapshot:
     """In-memory cache for node metadata between start and completion events."""
@@ -249,19 +263,19 @@ class WorkflowResponseConverter:
         outputs_mapping = graph_runtime_state.outputs or {}
         encoded_outputs = WorkflowRuntimeTypeConverter().to_json_encodable(outputs_mapping)
 
-        created_by: Mapping[str, object] | None
+        created_by: CreatedByDict | dict[str, object] = {}
         user = self._user
         if isinstance(user, Account):
-            created_by = {
+            created_by = AccountCreatedByDict(
-                "id": user.id,
+                id=user.id,
-                "name": user.name,
+                name=user.name,
-                "email": user.email,
+                email=user.email,
-            }
+            )
-        else:
+        elif isinstance(user, EndUser):
-            created_by = {
+            created_by = EndUserCreatedByDict(
-                "id": user.id,
+                id=user.id,
-                "user": user.session_id,
+                user=user.session_id,
-            }
+            )
 
         return WorkflowFinishStreamResponse(
             task_id=task_id,

api/core/app/task_pipeline/message_file_utils.py

@@ -1,3 +1,5 @@
+from typing import TypedDict
+
 from core.tools.signature import sign_tool_file
 from dify_graph.file import helpers as file_helpers
 from dify_graph.file.enums import FileTransferMethod
@@ -6,7 +8,20 @@ from models.model import MessageFile, UploadFile
 MAX_TOOL_FILE_EXTENSION_LENGTH = 10
 
 
-def prepare_file_dict(message_file: MessageFile, upload_files_map: dict[str, UploadFile]) -> dict:
+class MessageFileInfoDict(TypedDict):
+    related_id: str
+    extension: str
+    filename: str
+    size: int
+    mime_type: str
+    transfer_method: str
+    type: str
+    url: str
+    upload_file_id: str
+    remote_url: str | None
+
+
+def prepare_file_dict(message_file: MessageFile, upload_files_map: dict[str, UploadFile]) -> MessageFileInfoDict:
     """
     Prepare file dictionary for message end stream response.