def by_id(cls, db_conn: DatabaseConnection, id_: int | None,
create: bool = False) -> Process:
"""Collect Process, its VersionedAttributes, and its child IDs."""
- if id_ in db_conn.cached_processes.keys():
- process = db_conn.cached_processes[id_]
- assert isinstance(process, Process)
- return process
process = None
- for row in db_conn.exec('SELECT * FROM processes '
- 'WHERE id = ?', (id_,)):
- process = cls(row[0])
- break
+ if id_:
+ process, _ = super()._by_id(db_conn, id_)
if not process:
if not create:
raise NotFoundException(f'Process not found of id: {id_}')
@classmethod
def by_id(cls, db_conn: DatabaseConnection, id_: int) -> ProcessStep:
"""Retrieve ProcessStep by id_, or throw NotFoundException."""
- if id_ in db_conn.cached_process_steps.keys():
- step = db_conn.cached_process_steps[id_]
+ step, _ = super()._by_id(db_conn, id_)
+ if step:
assert isinstance(step, ProcessStep)
return step
- for row in db_conn.exec('SELECT * FROM process_steps '
- 'WHERE step_id = ?', (id_,)):
- step = cls.from_table_row(db_conn, row)
- assert isinstance(step, ProcessStep)
raise NotFoundException(f'found no ProcessStep of ID {id_}')
def save(self, db_conn: DatabaseConnection) -> None: