瀏覽代碼

refactor(iteration_node): use Sequence and Mapping in parameters (#11483)

Signed-off-by: -LAN- <laipz8200@outlook.com>
tags/0.13.2
-LAN- 10 月之前
父節點
當前提交
537068cfde
No account linked to committer's email address
共有 2 個文件被更改,包括 25 次插入23 次删除
  1. 9
    8
      api/core/workflow/graph_engine/entities/event.py
  2. 16
    15
      api/core/workflow/nodes/iteration/iteration_node.py

+ 9
- 8
api/core/workflow/graph_engine/entities/event.py 查看文件

from collections.abc import Mapping
from datetime import datetime from datetime import datetime
from typing import Any, Optional from typing import Any, Optional




class IterationRunStartedEvent(BaseIterationEvent): class IterationRunStartedEvent(BaseIterationEvent):
start_at: datetime = Field(..., description="start at") start_at: datetime = Field(..., description="start at")
inputs: Optional[dict[str, Any]] = None
metadata: Optional[dict[str, Any]] = None
inputs: Optional[Mapping[str, Any]] = None
metadata: Optional[Mapping[str, Any]] = None
predecessor_node_id: Optional[str] = None predecessor_node_id: Optional[str] = None






class IterationRunSucceededEvent(BaseIterationEvent): class IterationRunSucceededEvent(BaseIterationEvent):
start_at: datetime = Field(..., description="start at") start_at: datetime = Field(..., description="start at")
inputs: Optional[dict[str, Any]] = None
outputs: Optional[dict[str, Any]] = None
metadata: Optional[dict[str, Any]] = None
inputs: Optional[Mapping[str, Any]] = None
outputs: Optional[Mapping[str, Any]] = None
metadata: Optional[Mapping[str, Any]] = None
steps: int = 0 steps: int = 0
iteration_duration_map: Optional[dict[str, float]] = None iteration_duration_map: Optional[dict[str, float]] = None




class IterationRunFailedEvent(BaseIterationEvent): class IterationRunFailedEvent(BaseIterationEvent):
start_at: datetime = Field(..., description="start at") start_at: datetime = Field(..., description="start at")
inputs: Optional[dict[str, Any]] = None
outputs: Optional[dict[str, Any]] = None
metadata: Optional[dict[str, Any]] = None
inputs: Optional[Mapping[str, Any]] = None
outputs: Optional[Mapping[str, Any]] = None
metadata: Optional[Mapping[str, Any]] = None
steps: int = 0 steps: int = 0
error: str = Field(..., description="failed reason") error: str = Field(..., description="failed reason")



+ 16
- 15
api/core/workflow/nodes/iteration/iteration_node.py 查看文件

for index, item in enumerate(iterator_list_value): for index, item in enumerate(iterator_list_value):
future: Future = thread_pool.submit( future: Future = thread_pool.submit(
self._run_single_iter_parallel, self._run_single_iter_parallel,
current_app._get_current_object(), # type: ignore
q,
iterator_list_value,
inputs,
outputs,
start_at,
graph_engine,
iteration_graph,
index,
item,
iter_run_map,
flask_app=current_app._get_current_object(), # type: ignore
q=q,
iterator_list_value=iterator_list_value,
inputs=inputs,
outputs=outputs,
start_at=start_at,
graph_engine=graph_engine,
iteration_graph=iteration_graph,
index=index,
item=item,
iter_run_map=iter_run_map,
) )
future.add_done_callback(thread_pool.task_done_callback) future.add_done_callback(thread_pool.task_done_callback)
futures.append(future) futures.append(future)
def _run_single_iter( def _run_single_iter(
self, self,
*, *,
iterator_list_value: list[str],
iterator_list_value: Sequence[str],
variable_pool: VariablePool, variable_pool: VariablePool,
inputs: dict[str, list],
inputs: Mapping[str, list],
outputs: list, outputs: list,
start_at: datetime, start_at: datetime,
graph_engine: "GraphEngine", graph_engine: "GraphEngine",


def _run_single_iter_parallel( def _run_single_iter_parallel(
self, self,
*,
flask_app: Flask, flask_app: Flask,
q: Queue, q: Queue,
iterator_list_value: list[str],
inputs: dict[str, list],
iterator_list_value: Sequence[str],
inputs: Mapping[str, list],
outputs: list, outputs: list,
start_at: datetime, start_at: datetime,
graph_engine: "GraphEngine", graph_engine: "GraphEngine",

Loading…
取消
儲存