- steps[step.id_] = make_node(step)
- for step_id, step_node in steps.items():
- walk_steps(step_id, step_node)
- return steps
-
- def set_conditions(self, db_conn: DatabaseConnection, ids: list[int],
- trgt: str = 'conditions') -> None:
- """Set self.[target] to Conditions identified by ids."""
- trgt_list = getattr(self, trgt)
- while len(trgt_list) > 0:
- trgt_list.pop()
- for id_ in ids:
- trgt_list += [Condition.by_id(db_conn, id_)]
-
- def set_fulfills(self, db_conn: DatabaseConnection,
- ids: list[int]) -> None:
- """Set self.fulfills to Conditions identified by ids."""
- self.set_conditions(db_conn, ids, 'fulfills')
-
- def set_undoes(self, db_conn: DatabaseConnection, ids: list[int]) -> None:
- """Set self.undoes to Conditions identified by ids."""
- self.set_conditions(db_conn, ids, 'undoes')
-
- def _add_step(self,
- db_conn: DatabaseConnection,
- id_: int | None,
- step_process_id: int,
- parent_step_id: int | None) -> ProcessStep:
- """Create new ProcessStep, save and add it to self.explicit_steps.
+ new_node = make_node(step, step in external_owner.suppressed_steps)
+ step_nodes += [new_node]
+ for step_node in step_nodes:
+ walk_steps(step_node)
+ return step_nodes
+
+ def set_step_relations(self,
+ db_conn: DatabaseConnection,
+ owners: list[int],
+ suppressions: list[int],
+ owned_steps: list[ProcessStep]
+ ) -> None:
+ """Set step owners, suppressions, and owned steps."""
+ self._set_owners(db_conn, owners)
+ self._set_step_suppressions(db_conn, suppressions)
+ self.set_steps(db_conn, owned_steps)
+
+ def _set_step_suppressions(self,
+ db_conn: DatabaseConnection,
+ step_ids: list[int]
+ ) -> None:
+ """Set self.suppressed_steps from step_ids."""
+ assert isinstance(self.id_, int)
+ db_conn.delete_where('process_step_suppressions', 'process', self.id_)
+ self.suppressed_steps = [ProcessStep.by_id(db_conn, s)
+ for s in step_ids]
+
+ def _set_owners(self,
+ db_conn: DatabaseConnection,
+ owner_ids: list[int]
+ ) -> None:
+ """Re-set owners to those identified in owner_ids."""
+ owners_old = self.used_as_step_by(db_conn)
+ losers = [o for o in owners_old if o.id_ not in owner_ids]
+ owners_old_ids = [o.id_ for o in owners_old]
+ winners = [Process.by_id(db_conn, id_) for id_ in owner_ids
+ if id_ not in owners_old_ids]
+ steps_to_remove = []
+ for loser in losers:
+ steps_to_remove += [s for s in loser.explicit_steps
+ if s.step_process_id == self.id_]
+ for step in steps_to_remove:
+ step.remove(db_conn)
+ for winner in winners:
+ assert isinstance(winner.id_, int)
+ assert isinstance(self.id_, int)
+ new_step = ProcessStep(None, winner.id_, self.id_, None)
+ new_explicit_steps = winner.explicit_steps + [new_step]
+ winner.set_steps(db_conn, new_explicit_steps)