wip

Signed-off-by: Yves Bastide <[email protected]>

Author: ybastide

simpleflow/__init__.py

@@ -2,7 +2,7 @@
 
 from .activity import Activity  # NOQA
 from .runtime import logger  # NOQA
-from .signal import WaitForSignal  # NOQA
+from .simpleflow_signal import WaitForSignal  # NOQA
 from .workflow import Workflow  # NOQA
 
 __version__ = "0.34.0"

simpleflow/command.py

@@ -453,6 +453,19 @@ def workflow_history(
                 "external_workflows_signaling": history.external_workflows_signaling,
                 "signaled_workflows": history.signaled_workflows,
             }
+        elif mode == "cooked2":
+            history.parse()
+            events = {
+                "workflow": [t for t in history.tasks if t.type == "child_workflow"],
+                "activities": [t for t in history.tasks if t.type == "activity"],
+                "child_workflows": history.child_workflows,
+                "markers": history.markers,
+                "timers": history.timers,
+                "signals": [t for t in history.tasks if t.type == "signal"],
+                "signal_lists": history.signal_lists,
+                "external_workflows_signaling": history.external_workflows_signaling,
+                "signaled_workflows": history.signaled_workflows,
+            }
         else:
             raise NotImplementedError
     print(json.dumps(events, separators=(",", ":"), default=serialize_complex_object))

simpleflow/history.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import collections
-from typing import TYPE_CHECKING, Callable, ClassVar
+from typing import TYPE_CHECKING, Callable, ClassVar, cast
 
 import simpleflow.swf.mapper.models.history
 from simpleflow import logger
@@ -143,6 +143,48 @@ def tasks(self):
     def events(self) -> list[Event]:
         return self._history.events
 
+    def get_activities_history(self) -> dict[str, dict[str, Any]]:
+        activities: dict[str, dict[str, Any]] = {}
+        scheduled_to_activity_id: dict[int, str] = {}
+        event: ActivityTaskEvent | Any
+        for event in self.events:
+            if event.type != "ActivityTask":
+                continue
+            cast(ActivityTaskEvent, event)
+            activity_id = getattr(event, "activity_id", None)
+            if event.state == "scheduled" and activity_id not in activities:
+                activities[activity_id] = {
+                    "id": activity_id,
+                    "name": event.activity_type["name"],
+                    "version": event.activity_type["version"],
+                    "states": [event.state],
+                    "scheduled_ids": [event.id],
+                    "scheduled_timestamps": [event.timestamp],
+                    "inputs": [event.input],
+                    "task_lists": [event.task_list["name"]],
+                }
+                scheduled_to_activity_id[event.id] = activity_id
+            else:
+                if event.state != "scheduled":
+                    scheduled_event = self.events[event.scheduled_event_id - 1]
+                    scheduled_id = scheduled_event.id
+                    activity = activities[scheduled_to_activity_id[scheduled_id]]
+                    activity["task_lists"].append(event.task_list["name"])
+                else:
+                    activity = activities[event.activity_id]
+                activity.setdefault("states", []).append(event.state)
+                activity.setdefault(f"{event.state}_ids", []).append(event.id)
+                activity.setdefault(f"{event.state}_timestamp", []).append(event.timestamp)
+                for attr in ("identity", "result", "reason", "details"):
+                    if hasattr(event, attr):
+                        activity.setdefault(attr, []).append(getattr(event, attr))
+                if event.state == "timed_out":
+                    activity.setdefault("timeout_types", []).append(event.timeout_type)
+                    activity.setdefault(f"{event.timeout_type}_timeouts", []).append(
+                        getattr(event, f"{event.timeout_type}_timeout")
+                    )
+                    activity.setdefault(f"{event.timeout_values}", []).append()
+
     def parse_activity_event(self, events: list[Event | ActivityTaskEvent], event: ActivityTaskEvent):
         """
         Aggregate all the attributes of an activity in a single entry.

simpleflow/local/executor.py

@@ -10,7 +10,7 @@
 from simpleflow.base import Submittable
 from simpleflow.history import History
 from simpleflow.marker import Marker
-from simpleflow.signal import WaitForSignal
+from simpleflow.simpleflow_signal import WaitForSignal
 from simpleflow.swf.mapper.models.history import builder
 from simpleflow.task import ActivityTask, MarkerTask, SignalTask, TaskFailureContext, WorkflowTask
 from simpleflow.utils import format_exc, format_exc_type, issubclass_, json_dumps

simpleflow/signal.py simpleflow/simpleflow_signal.py

@@ -19,7 +19,7 @@
 from simpleflow.base import Submittable
 from simpleflow.history import History
 from simpleflow.marker import Marker
-from simpleflow.signal import WaitForSignal
+from simpleflow.simpleflow_signal import WaitForSignal
 from simpleflow.swf import constants
 from simpleflow.swf.helpers import swf_identity
 from simpleflow.swf.mapper.core import ConnectedSWFObject

simpleflow/swf/executor.py

@@ -36,22 +36,23 @@ class ActivityTaskEvent(Event):
 
 class ActivityTaskEventDict(TypedDict):
     type: str
-    id: int
+    id: str
     name: str
     version: str
     state: str
     scheduled_id: int
     scheduled_timestamp: datetime
+    schedule_failed_timestamp: datetime | None
     started_id: int
     started_timestamp: datetime
     completed_id: int
-    completed_timestamp: datetime
+    completed_timestamp: datetime | None
     failed_id: int
-    failed_timestamp: datetime
+    failed_timestamp: datetime | None
     timeout_type: str
     timeout_value: int
     timed_out_id: int
-    timed_out_timestamp: datetime
+    timed_out_timestamp: datetime | None
     identity: Any
     input: dict
     task_list: str

simpleflow/swf/mapper/models/event/task.py

@@ -3,7 +3,7 @@
 from typing import TYPE_CHECKING
 
 from simpleflow.base import Submittable, SubmittableContainer
-from simpleflow.signal import WaitForSignal
+from simpleflow.simpleflow_signal import WaitForSignal
 from simpleflow.task import CancelTimerTask, TaskFailureContext, TimerTask
 
 from . import canvas, task