"""Collecting Processes and Process-related items."""
from __future__ import annotations
-from typing import Any, Set
+from dataclasses import dataclass
+from typing import Set
from plomtask.db import DatabaseConnection, BaseModel
from plomtask.misc import VersionedAttribute
from plomtask.conditions import Condition, ConditionsRelations
from plomtask.exceptions import NotFoundException, BadFormatException
+@dataclass
+class ProcessStepsNode:
+ """Collects what's useful to know for ProcessSteps tree display."""
+ process: Process
+ parent_id: int | None
+ is_explicit: bool
+ steps: dict[int, ProcessStepsNode]
+ seen: bool
+
+
class Process(BaseModel, ConditionsRelations):
"""Template for, and metadata for, Todos, and their arrangements."""
table_name = 'processes'
return [self.__class__.by_id(db_conn, id_) for id_ in owner_ids]
def get_steps(self, db_conn: DatabaseConnection, external_owner:
- Process | None = None) -> dict[int, dict[str, object]]:
+ Process | None = None) -> dict[int, ProcessStepsNode]:
"""Return tree of depended-on explicit and implicit ProcessSteps."""
- def make_node(step: ProcessStep) -> dict[str, object]:
+ def make_node(step: ProcessStep) -> ProcessStepsNode:
is_explicit = False
if external_owner is not None:
is_explicit = step.owner_id == external_owner.id_
process = self.__class__.by_id(db_conn, step.step_process_id)
step_steps = process.get_steps(db_conn, external_owner)
- return {'process': process, 'parent_id': step.parent_step_id,
- 'is_explicit': is_explicit, 'steps': step_steps}
+ return ProcessStepsNode(process, step.parent_step_id,
+ is_explicit, step_steps, False)
- def walk_steps(node_id: int, node: dict[str, Any]) -> None:
+ def walk_steps(node_id: int, node: ProcessStepsNode) -> None:
explicit_children = [s for s in self.explicit_steps
if s.parent_step_id == node_id]
for child in explicit_children:
- node['steps'][child.id_] = make_node(child)
- node['seen'] = node_id in seen_step_ids
+ assert isinstance(child.id_, int)
+ node.steps[child.id_] = make_node(child)
+ node.seen = node_id in seen_step_ids
seen_step_ids.add(node_id)
- for id_, step in node['steps'].items():
+ for id_, step in node.steps.items():
walk_steps(id_, step)
- steps: dict[int, dict[str, object]] = {}
+ steps: dict[int, ProcessStepsNode] = {}
seen_step_ids: Set[int] = set()
if external_owner is None:
external_owner = self
just deleted under its feet), or if the parent step would not be
owned by the current Process.
"""
+
def walk_steps(node: ProcessStep) -> None:
if node.step_process_id == self.id_:
raise BadFormatException('bad step selection causes recursion')
step_process = self.by_id(db_conn, node.step_process_id)
for step in step_process.explicit_steps:
walk_steps(step)
+
if parent_step_id is not None:
try:
parent_step = ProcessStep.by_id(db_conn, parent_step_id)