parent_id: int | None
is_explicit: bool
steps: dict[int, ProcessStepsNode]
- seen: bool
+ seen: bool = False
+ is_suppressed: bool = False
class Process(BaseModel[int], ConditionsRelations):
to_save_relations = [('process_conditions', 'process', 'conditions', 0),
('process_blockers', 'process', 'blockers', 0),
('process_enables', 'process', 'enables', 0),
- ('process_disables', 'process', 'disables', 0)]
+ ('process_disables', 'process', 'disables', 0),
+ ('process_step_suppressions', 'process',
+ 'suppressed_steps', 0)]
to_search = ['title.newest', 'description.newest']
def __init__(self, id_: int | None, calendarize: bool = False) -> None:
self.description = VersionedAttribute(self, 'process_descriptions', '')
self.effort = VersionedAttribute(self, 'process_efforts', 1.0)
self.explicit_steps: list[ProcessStep] = []
+ self.suppressed_steps: list[ProcessStep] = []
self.calendarize = calendarize
@classmethod
process.id_):
step = ProcessStep.from_table_row(db_conn, row_)
process.explicit_steps += [step] # pylint: disable=no-member
+ for row_ in db_conn.row_where('process_step_suppressions', 'process',
+ process.id_):
+ step = ProcessStep.by_id(db_conn, row_[1])
+ process.suppressed_steps += [step] # pylint: disable=no-member
for name in ('conditions', 'blockers', 'enables', 'disables'):
table = f'process_{name}'
assert isinstance(process.id_, int)
Process | None = None) -> dict[int, ProcessStepsNode]:
"""Return tree of depended-on explicit and implicit ProcessSteps."""
- def make_node(step: ProcessStep) -> ProcessStepsNode:
+ def make_node(step: ProcessStep, suppressed: bool) -> 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)
+ step_steps = {}
+ if not suppressed:
+ step_steps = process.get_steps(db_conn, external_owner)
return ProcessStepsNode(process, step.parent_step_id,
- is_explicit, step_steps, False)
+ is_explicit, step_steps, False, suppressed)
def walk_steps(node_id: int, node: ProcessStepsNode) -> None:
+ node.seen = node_id in seen_step_ids
+ seen_step_ids.add(node_id)
+ if node.is_suppressed:
+ return
explicit_children = [s for s in self.explicit_steps
if s.parent_step_id == node_id]
for child in explicit_children:
assert isinstance(child.id_, int)
- node.steps[child.id_] = make_node(child)
- # # ensure that one (!) explicit step of process replaces
- # # one (!) implicit step of same process
- # for i in [i for i, s in node.steps.items()
- # if not s.process_step.owner_id == child.id_
- # and s.process.id_ == child.step_process_id]:
- # del node.steps[i]
- # break
- node.seen = node_id in seen_step_ids
- seen_step_ids.add(node_id)
+ node.steps[child.id_] = make_node(child, False)
for id_, step in node.steps.items():
walk_steps(id_, step)
for step in [s for s in self.explicit_steps
if s.parent_step_id is None]:
assert isinstance(step.id_, int)
- steps[step.id_] = make_node(step)
+ new_node = make_node(step, step in external_owner.suppressed_steps)
+ steps[step.id_] = new_node
for step_id, step_node in steps.items():
walk_steps(step_id, step_node)
return 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_steps(self, db_conn: DatabaseConnection,
steps: list[ProcessStep]) -> None:
"""Set self.explicit_steps in bulk.