Implemented foundation of task-flow-engine

task_flow_engine/models.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class AssertionDef:
+    id: str
+    target: str
+    op: str
+    value: Any = None
+    description: str = ""
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "AssertionDef":
+        return cls(
+            id=str(data["id"]),
+            target=str(data.get("target", "")),
+            op=str(data["op"]),
+            value=data.get("value"),
+            description=str(data.get("description", "")),
+        )
+
+
+@dataclass(frozen=True)
+class WorkstationDef:
+    name: str
+    entry_assertions: list[AssertionDef] = field(default_factory=list)
+    exit_assertions: list[AssertionDef] = field(default_factory=list)
+    description: str = ""
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "WorkstationDef":
+        return cls(
+            name=str(data["name"]),
+            description=str(data.get("description", "")),
+            entry_assertions=[
+                AssertionDef.from_dict(item)
+                for item in data.get("entry_assertions", []) or []
+            ],
+            exit_assertions=[
+                AssertionDef.from_dict(item)
+                for item in data.get("exit_assertions", []) or []
+            ],
+        )
+
+
+@dataclass(frozen=True)
+class FlowDef:
+    id: str
+    entity_type: str
+    workstations: list[WorkstationDef]
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "FlowDef":
+        return cls(
+            id=str(data["id"]),
+            entity_type=str(data["entity_type"]),
+            workstations=[
+                WorkstationDef.from_dict(item)
+                for item in data.get("workstations", []) or []
+            ],
+        )
+
+    def workstation(self, name: str) -> WorkstationDef | None:
+        return next((item for item in self.workstations if item.name == name), None)
+
+
+@dataclass(frozen=True)
+class AssertionResult:
+    id: str
+    passed: bool
+    target: str
+    op: str
+    expected: Any = None
+    actual: list[Any] = field(default_factory=list)
+    description: str = ""
+    reason: str = ""
+
+
+@dataclass(frozen=True)
+class UnreachableWorkstation:
+    workstation: str
+    blocking: AssertionResult
+
+
+@dataclass(frozen=True)
+class FlowResult:
+    current_workstation: str
+    exit_blocked: bool
+    blocking_assertions: list[AssertionResult]
+    reachable: list[str]
+    unreachable: list[UnreachableWorkstation]