"""Collecting Processes and Process-related items."""
from __future__ import annotations
-from sqlite3 import Row
-from datetime import datetime
from typing import Any, Set
-from plomtask.db import DatabaseConnection
+from plomtask.db import DatabaseConnection, BaseModel
+from plomtask.misc import VersionedAttribute
+from plomtask.conditions import Condition
from plomtask.exceptions import NotFoundException, BadFormatException
-class Process:
+class Process(BaseModel):
"""Template for, and metadata for, Todos, and their arrangements."""
+ table_name = 'processes'
+
+ # pylint: disable=too-many-instance-attributes
def __init__(self, id_: int | None) -> None:
- if (id_ is not None) and id_ < 1:
- raise BadFormatException(f'illegal Process ID, must be >=1: {id_}')
- self.id_ = id_
- self.title = VersionedAttribute(self, 'title', 'UNNAMED')
- self.description = VersionedAttribute(self, 'description', '')
- self.effort = VersionedAttribute(self, 'effort', 1.0)
+ self.set_int_id(id_)
+ self.title = VersionedAttribute(self, 'process_titles', 'UNNAMED')
+ self.description = VersionedAttribute(self, 'process_descriptions', '')
+ self.effort = VersionedAttribute(self, 'process_efforts', 1.0)
self.explicit_steps: list[ProcessStep] = []
-
- @classmethod
- def from_table_row(cls, db_conn: DatabaseConnection, row: Row) -> Process:
- """Make Process from database row, with empty VersionedAttributes."""
- process = cls(row[0])
- assert process.id_ is not None
- db_conn.cached_processes[process.id_] = process
- return process
+ self.conditions: list[Condition] = []
+ self.fulfills: list[Condition] = []
+ self.undoes: list[Condition] = []
@classmethod
def all(cls, db_conn: DatabaseConnection) -> list[Process]:
for id_, process in db_conn.cached_processes.items():
processes[id_] = process
already_recorded = processes.keys()
- for row in db_conn.exec('SELECT id FROM processes'):
- if row[0] not in already_recorded:
- process = cls.by_id(db_conn, row[0])
+ for id_ in db_conn.column_all('processes', 'id'):
+ if id_ not in already_recorded:
+ process = cls.by_id(db_conn, id_)
processes[process.id_] = process
return list(processes.values())
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_}')
process = Process(id_)
- if process:
- for row in db_conn.exec('SELECT * FROM process_titles '
- 'WHERE process_id = ?', (process.id_,)):
- process.title.history[row[1]] = row[2]
- for row in db_conn.exec('SELECT * FROM process_descriptions '
- 'WHERE process_id = ?', (process.id_,)):
- process.description.history[row[1]] = row[2]
- for row in db_conn.exec('SELECT * FROM process_efforts '
- 'WHERE process_id = ?', (process.id_,)):
- process.effort.history[row[1]] = row[2]
- for row in db_conn.exec('SELECT * FROM process_steps '
- 'WHERE owner_id = ?', (process.id_,)):
- process.explicit_steps += [ProcessStep.from_table_row(db_conn,
- row)]
+ if isinstance(process.id_, int):
+ for name in ('title', 'description', 'effort'):
+ table = f'process_{name}s'
+ for row in db_conn.row_where(table, 'parent', process.id_):
+ getattr(process, name).history_from_row(row)
+ for row in db_conn.row_where('process_steps', 'owner',
+ process.id_):
+ step = ProcessStep.from_table_row(db_conn, row)
+ process.explicit_steps += [step]
+ for name in ('conditions', 'fulfills', 'undoes'):
+ table = f'process_{name}'
+ for cond_id in db_conn.column_where(table, 'condition',
+ 'process', process.id_):
+ target = getattr(process, name)
+ target += [Condition.by_id(db_conn, cond_id)]
assert isinstance(process, Process)
return process
def used_as_step_by(self, db_conn: DatabaseConnection) -> list[Process]:
"""Return Processes using self for a ProcessStep."""
+ if not self.id_:
+ return []
owner_ids = set()
- for owner_id in db_conn.exec('SELECT owner_id FROM process_steps WHERE'
- ' step_process_id = ?', (self.id_,)):
- owner_ids.add(owner_id[0])
+ for id_ in db_conn.column_where('process_steps', 'owner',
+ 'step_process', self.id_):
+ owner_ids.add(id_)
return [self.__class__.by_id(db_conn, id_) for id_ in owner_ids]
def get_steps(self, db_conn: DatabaseConnection, external_owner:
"""Return tree of depended-on explicit and implicit ProcessSteps."""
def make_node(step: ProcessStep) -> dict[str, object]:
- step_process = self.__class__.by_id(db_conn, step.step_process_id)
is_explicit = False
if external_owner is not None:
is_explicit = step.owner_id == external_owner.id_
- step_steps = step_process.get_steps(db_conn, external_owner)
- return {'process': step_process, 'parent_id': step.parent_step_id,
+ process = self.__class__.by_id(db_conn, step.step_process_id)
+ step_steps = process.get_steps(db_conn, external_owner)
+ return {'process': process, 'parent_id': step.parent_step_id,
'is_explicit': is_explicit, 'steps': step_steps}
def walk_steps(node_id: int, node: dict[str, Any]) -> None:
external_owner = self
for step in [s for s in self.explicit_steps
if s.parent_step_id is None]:
- assert step.id_ is not None # for mypy
+ assert isinstance(step.id_, int)
steps[step.id_] = make_node(step)
for step_id, step_node in steps.items():
walk_steps(step_id, step_node)
return steps
- def add_step(self, db_conn: DatabaseConnection, id_: int | None,
- step_process_id: int,
- parent_step_id: int | None) -> ProcessStep:
+ 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.
Also checks against step recursion.
+
The new step's parent_step_id will fall back to None either if no
matching ProcessStep is found (which can be assumed in case it was
just deleted under its feet), or if the parent step would not be
parent_step_id = None
except NotFoundException:
parent_step_id = None
- assert self.id_ is not None
+ assert isinstance(self.id_, int)
step = ProcessStep(id_, self.id_, step_process_id, parent_step_id)
walk_steps(step)
self.explicit_steps += [step]
step.save(db_conn) # NB: This ensures a non-None step.id_.
return step
- def save_without_steps(self, db_conn: DatabaseConnection) -> None:
- """Add (or re-write) self and connected VersionedAttributes to DB."""
- cursor = db_conn.exec('REPLACE INTO processes VALUES (?)', (self.id_,))
- self.id_ = cursor.lastrowid
+ def set_steps(self, db_conn: DatabaseConnection,
+ steps: list[tuple[int | None, int, int | None]]) -> None:
+ """Set self.explicit_steps in bulk."""
+ assert isinstance(self.id_, int)
+ for step in self.explicit_steps:
+ assert isinstance(step.id_, int)
+ del db_conn.cached_process_steps[step.id_]
+ self.explicit_steps = []
+ db_conn.delete_where('process_steps', 'owner', self.id_)
+ for step_tuple in steps:
+ self._add_step(db_conn, step_tuple[0],
+ step_tuple[1], step_tuple[2])
+
+ def save(self, db_conn: DatabaseConnection) -> None:
+ """Add (or re-write) self and connected items to DB."""
+ self.save_core(db_conn)
+ assert isinstance(self.id_, int)
self.title.save(db_conn)
self.description.save(db_conn)
self.effort.save(db_conn)
- assert self.id_ is not None
- db_conn.cached_processes[self.id_] = self
-
- def fix_steps(self, db_conn: DatabaseConnection) -> None:
- """Rewrite ProcessSteps from self.explicit_steps.
-
- This also fixes illegal Step.parent_step_id values, i.e. those pointing
- to steps now absent, or owned by a different Process, fall back into
- .parent_step_id=None
- """
- db_conn.exec('DELETE FROM process_steps WHERE owner_id = ?',
- (self.id_,))
+ db_conn.rewrite_relations('process_conditions', 'process', self.id_,
+ [[c.id_] for c in self.conditions])
+ db_conn.rewrite_relations('process_fulfills', 'process', self.id_,
+ [[c.id_] for c in self.fulfills])
+ db_conn.rewrite_relations('process_undoes', 'process', self.id_,
+ [[c.id_] for c in self.undoes])
+ db_conn.delete_where('process_steps', 'owner', self.id_)
for step in self.explicit_steps:
- if step.parent_step_id is not None:
- try:
- parent_step = ProcessStep.by_id(db_conn,
- step.parent_step_id)
- if parent_step.owner_id != self.id_:
- step.parent_step_id = None
- except NotFoundException:
- step.parent_step_id = None
step.save(db_conn)
+ db_conn.cached_processes[self.id_] = self
-class ProcessStep:
+class ProcessStep(BaseModel):
"""Sub-unit of Processes."""
+ table_name = 'process_steps'
+ to_save = ['owner_id', 'step_process_id', 'parent_step_id']
def __init__(self, id_: int | None, owner_id: int, step_process_id: int,
parent_step_id: int | None) -> None:
- self.id_ = id_
+ self.set_int_id(id_)
self.owner_id = owner_id
self.step_process_id = step_process_id
self.parent_step_id = parent_step_id
- @classmethod
- def from_table_row(cls, db_conn: DatabaseConnection,
- row: Row) -> ProcessStep:
- """Make ProcessStep from database row, store in DB cache."""
- step = cls(row[0], row[1], row[2], row[3])
- assert step.id_ is not None
- db_conn.cached_process_steps[step.id_] = step
- return step
-
@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_,)):
- return cls.from_table_row(db_conn, row)
raise NotFoundException(f'found no ProcessStep of ID {id_}')
def save(self, db_conn: DatabaseConnection) -> None:
- """Save to database and cache."""
- cursor = db_conn.exec('REPLACE INTO process_steps VALUES (?, ?, ?, ?)',
- (self.id_, self.owner_id, self.step_process_id,
- self.parent_step_id))
- self.id_ = cursor.lastrowid
- assert self.id_ is not None
- db_conn.cached_process_steps[self.id_] = self
-
-
-class VersionedAttribute:
- """Attributes whose values are recorded as a timestamped history."""
-
- def __init__(self,
- parent: Process, name: str, default: str | float) -> None:
- self.parent = parent
- self.name = name
- self.default = default
- self.history: dict[str, str | float] = {}
-
- @property
- def _newest_timestamp(self) -> str:
- """Return most recent timestamp."""
- return sorted(self.history.keys())[-1]
-
- @property
- def newest(self) -> str | float:
- """Return most recent value, or self.default if self.history empty."""
- if 0 == len(self.history):
- return self.default
- return self.history[self._newest_timestamp]
-
- def set(self, value: str | float) -> None:
- """Add to self.history if and only if not same value as newest one."""
- if 0 == len(self.history) \
- or value != self.history[self._newest_timestamp]:
- self.history[datetime.now().strftime('%Y-%m-%d %H:%M:%S')] = value
-
- def at(self, queried_time: str) -> str | float:
- """Retrieve value of timestamp nearest queried_time from the past."""
- sorted_timestamps = sorted(self.history.keys())
- if 0 == len(sorted_timestamps):
- return self.default
- selected_timestamp = sorted_timestamps[0]
- for timestamp in sorted_timestamps[1:]:
- if timestamp > queried_time:
- break
- selected_timestamp = timestamp
- return self.history[selected_timestamp]
-
- def save(self, db_conn: DatabaseConnection) -> None:
- """Save as self.history entries, but first wipe old ones."""
- db_conn.exec(f'DELETE FROM process_{self.name}s WHERE process_id = ?',
- (self.parent.id_,))
- for timestamp, value in self.history.items():
- db_conn.exec(f'INSERT INTO process_{self.name}s VALUES (?, ?, ?)',
- (self.parent.id_, timestamp, value))
+ """Default to simply calling self.save_core for simple cases."""
+ self.save_core(db_conn)
projects / plomtask / blobdiff