from __future__ import annotations
from sqlite3 import Row
from datetime import datetime
-from typing import Any
+from typing import Any, Set
from plomtask.db import DatabaseConnection
from plomtask.exceptions import NotFoundException, BadFormatException
process.explicit_steps += [ProcessStep.from_table_row(row)]
return process
+ def used_as_step_by(self, db_conn: DatabaseConnection) -> list[Process]:
+ """Return Processes using self for a ProcessStep."""
+ owner_ids = set()
+ for owner_id in db_conn.exec('SELECT owner_id FROM process_steps WHERE'
+ ' step_process_id = ?', (self.id_,)):
+ owner_ids.add(owner_id[0])
+ 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]]:
"""Return tree of depended-on explicit and implicit ProcessSteps."""
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
+ seen_step_ids.add(node_id)
for id_, step in node['steps'].items():
walk_steps(id_, step)
steps: dict[int, dict[str, object]] = {}
+ seen_step_ids: Set[int] = set()
if external_owner is None:
external_owner = self
for step in [s for s in self.explicit_steps