1 """Non-doable elements of ProcessStep/Todo chains."""
2 from __future__ import annotations
4 from sqlite3 import Row
5 from plomtask.db import DatabaseConnection, BaseModel
6 from plomtask.versioned_attributes import VersionedAttribute
7 from plomtask.exceptions import HandledException
10 class Condition(BaseModel[int]):
11 """Non-Process dependency for ProcessSteps and Todos."""
12 table_name = 'conditions'
13 to_save = ['is_active']
15 def __init__(self, id_: int | None, is_active: bool = False) -> None:
17 self.is_active = is_active
18 self.title = VersionedAttribute(self, 'condition_titles', 'UNNAMED')
19 self.description = VersionedAttribute(self, 'condition_descriptions',
22 def __lt__(self, other: Condition) -> bool:
23 assert isinstance(self.id_, int)
24 assert isinstance(other.id_, int)
25 return self.id_ < other.id_
28 def from_table_row(cls, db_conn: DatabaseConnection,
29 row: Row | list[Any]) -> Condition:
30 """Build condition from row, including VersionedAttributes."""
31 condition = super().from_table_row(db_conn, row)
32 for name in ('title', 'description'):
33 table_name = f'condition_{name}s'
34 for row_ in db_conn.row_where(table_name, 'parent', row[0]):
35 getattr(condition, name).history_from_row(row_)
38 def save(self, db_conn: DatabaseConnection) -> None:
39 """Save self and its VersionedAttributes to DB and cache."""
40 self.save_core(db_conn)
41 self.title.save(db_conn)
42 self.description.save(db_conn)
44 def remove(self, db_conn: DatabaseConnection) -> None:
45 """Remove from DB, with VersionedAttributes.
47 Checks for Todos and Processes that depend on Condition, prohibits
51 raise HandledException('cannot remove unsaved item')
52 for item in ('process', 'todo'):
53 for attr in ('conditions', 'enables', 'disables'):
54 table_name = f'{item}_{attr}'
55 for _ in db_conn.row_where(table_name, 'condition', self.id_):
56 raise HandledException('cannot remove Condition in use')
57 db_conn.delete_where('condition_titles', 'parent', self.id_)
58 db_conn.delete_where('condition_descriptions', 'parent', self.id_)
59 super().remove(db_conn)
62 class ConditionsRelations:
63 """Methods for handling relations to Conditions, for Todo and Process."""
65 def set_conditions(self, db_conn: DatabaseConnection, ids: list[int],
66 target: str = 'conditions') -> None:
67 """Set self.[target] to Conditions identified by ids."""
68 target_list = getattr(self, target)
69 while len(target_list) > 0:
72 target_list += [Condition.by_id(db_conn, id_)]
74 def set_enables(self, db_conn: DatabaseConnection,
75 ids: list[int]) -> None:
76 """Set self.enables to Conditions identified by ids."""
77 self.set_conditions(db_conn, ids, 'enables')
79 def set_disables(self, db_conn: DatabaseConnection,
80 ids: list[int]) -> None:
81 """Set self.disables to Conditions identified by ids."""
82 self.set_conditions(db_conn, ids, 'disables')