refactor(api): type workflow service dicts with TypedDict (#33829)

api/services/workflow/workflow_converter.py

@@ -1,5 +1,7 @@
 import json
-from typing import Any, TypedDict
+from typing import Any
+
+from typing_extensions import TypedDict
 
 from core.app.app_config.entities import (
     DatasetEntity,
@@ -34,6 +36,17 @@ class _NodeType(TypedDict):
     data: dict[str, Any]
 
 
+class _EdgeType(TypedDict):
+    id: str
+    source: str
+    target: str
+
+
+class WorkflowGraph(TypedDict):
+    nodes: list[_NodeType]
+    edges: list[_EdgeType]
+
+
 class WorkflowConverter:
     """
     App Convert to Workflow Mode
@@ -107,7 +120,7 @@ class WorkflowConverter:
         app_config = self._convert_to_app_config(app_model=app_model, app_model_config=app_model_config)
 
         # init workflow graph
-        graph: dict[str, Any] = {"nodes": [], "edges": []}
+        graph: WorkflowGraph = {"nodes": [], "edges": []}
 
         # Convert list:
         # - variables -> start
@@ -385,7 +398,7 @@ class WorkflowConverter:
         self,
         original_app_mode: AppMode,
         new_app_mode: AppMode,
-        graph: dict,
+        graph: WorkflowGraph,
         model_config: ModelConfigEntity,
         prompt_template: PromptTemplateEntity,
         file_upload: FileUploadConfig | None = None,
@@ -595,7 +608,7 @@ class WorkflowConverter:
             "data": {"title": "ANSWER", "type": BuiltinNodeTypes.ANSWER, "answer": "{{#llm.text#}}"},
         }
 
-    def _create_edge(self, source: str, target: str):
+    def _create_edge(self, source: str, target: str) -> _EdgeType:
         """
         Create Edge
         :param source: source node id
@@ -604,7 +617,7 @@ class WorkflowConverter:
         """
         return {"id": f"{source}-{target}", "source": source, "target": target}
 
-    def _append_node(self, graph: dict[str, Any], node: _NodeType):
+    def _append_node(self, graph: WorkflowGraph, node: _NodeType):
         """
         Append Node to Graph
 

api/services/workflow_app_service.py

@@ -5,6 +5,7 @@ from typing import Any
 
 from sqlalchemy import and_, func, or_, select
 from sqlalchemy.orm import Session
+from typing_extensions import TypedDict
 
 from dify_graph.enums import WorkflowExecutionStatus
 from models import Account, App, EndUser, TenantAccountJoin, WorkflowAppLog, WorkflowArchiveLog, WorkflowRun
@@ -14,6 +15,10 @@ from services.plugin.plugin_service import PluginService
 from services.workflow.entities import TriggerMetadata
 
 
+class LogViewDetails(TypedDict):
+    trigger_metadata: dict[str, Any] | None
+
+
 # Since the workflow_app_log table has exceeded 100 million records, we use an additional details field to extend it
 class LogView:
     """Lightweight wrapper for WorkflowAppLog with computed details.
@@ -22,12 +27,12 @@ class LogView:
     - Proxies all other attributes to the underlying `WorkflowAppLog`
     """
 
-    def __init__(self, log: WorkflowAppLog, details: dict | None):
+    def __init__(self, log: WorkflowAppLog, details: LogViewDetails | None):
         self.log = log
         self.details_ = details
 
     @property
-    def details(self) -> dict | None:
+    def details(self) -> LogViewDetails | None:
         return self.details_
 
     def __getattr__(self, name):