1 """Non-doable elements of ProcessStep/Todo chains."""
2 from __future__ import annotations
3 from sqlite3 import Row
4 from plomtask.db import DatabaseConnection
5 from plomtask.misc import VersionedAttribute
6 from plomtask.exceptions import BadFormatException, NotFoundException
10 """Non Process-dependency for ProcessSteps and Todos."""
12 def __init__(self, id_: int | None, is_active: bool = False) -> None:
13 if (id_ is not None) and id_ < 1:
14 msg = f'illegal Condition ID, must be >=1: {id_}'
15 raise BadFormatException(msg)
17 self.is_active = is_active
18 self.title = VersionedAttribute(self, 'condition_titles', 'UNNAMED')
19 self.description = VersionedAttribute(self, 'condition_descriptions',
23 def from_table_row(cls, db_conn: DatabaseConnection,
24 row: Row) -> Condition:
25 """Build condition from row, including VersionedAttributes."""
26 condition = cls(row[0], row[1])
27 for title_row in db_conn.exec('SELECT * FROM condition_titles '
28 'WHERE parent_id = ?', (row[0],)):
29 condition.title.history[title_row[1]] = title_row[2]
30 for desc_row in db_conn.exec('SELECT * FROM condition_descriptions '
31 'WHERE parent_id = ?', (row[0],)):
32 condition.description.history[desc_row[1]] = desc_row[2]
36 def all(cls, db_conn: DatabaseConnection) -> list[Condition]:
37 """Collect all Conditions and their VersionedAttributes."""
39 for id_, condition in db_conn.cached_conditions.items():
40 conditions[id_] = condition
41 already_recorded = conditions.keys()
42 for row in db_conn.exec('SELECT id FROM conditions'):
43 if row[0] not in already_recorded:
44 condition = cls.by_id(db_conn, row[0])
45 conditions[condition.id_] = condition
46 return list(conditions.values())
49 def by_id(cls, db_conn: DatabaseConnection, id_: int | None,
50 create: bool = False) -> Condition:
51 """Collect (or create) Condition and its VersionedAttributes."""
53 if id_ in db_conn.cached_conditions.keys():
54 condition = db_conn.cached_conditions[id_]
56 for row in db_conn.exec('SELECT * FROM conditions WHERE id = ?',
58 condition = cls.from_table_row(db_conn, row)
62 raise NotFoundException(f'Condition not found of id: {id_}')
63 condition = cls(id_, False)
66 def save(self, db_conn: DatabaseConnection) -> None:
67 """Save self and its VersionedAttributes to DB and cache."""
68 cursor = db_conn.exec('REPLACE INTO conditions VALUES (?, ?)',
69 (self.id_, self.is_active))
70 self.id_ = cursor.lastrowid
71 self.title.save(db_conn)
72 self.description.save(db_conn)
73 assert self.id_ is not None
74 db_conn.cached_conditions[self.id_] = self