"""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, ConditionsRelations
from plomtask.exceptions import NotFoundException, BadFormatException
-class Process:
+class Process(BaseModel, ConditionsRelations):
"""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] = []
-
- def __eq__(self, other: object) -> bool:
- return isinstance(other, self.__class__) and self.id_ == other.id_
-
- @classmethod
- def from_table_row(cls, row: Row) -> Process:
- """Make Process from database row, with empty VersionedAttributes."""
- return cls(row[0])
+ self.conditions: list[Condition] = []
+ self.enables: list[Condition] = []
+ self.disables: list[Condition] = []
@classmethod
def all(cls, db_conn: DatabaseConnection) -> list[Process]:
"""Collect all Processes and their connected VersionedAttributes."""
processes = {}
- for row in db_conn.exec('SELECT * FROM processes'):
- process = cls.from_table_row(row)
- processes[process.id_] = process
- for row in db_conn.exec('SELECT * FROM process_titles'):
- processes[row[0]].title.history[row[1]] = row[2]
- for row in db_conn.exec('SELECT * FROM process_descriptions'):
- processes[row[0]].description.history[row[1]] = row[2]
- for row in db_conn.exec('SELECT * FROM process_efforts'):
- processes[row[0]].effort.history[row[1]] = row[2]
+ for id_, process in db_conn.cached_processes.items():
+ processes[id_] = process
+ already_recorded = processes.keys()
+ 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())
@classmethod
create: bool = False) -> Process:
"""Collect Process, its VersionedAttributes, and its child IDs."""
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(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', 'enables', 'disables'):
+ 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 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:
Process | None = None) -> dict[int, dict[str, object]]:
"""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 _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)
-
- 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_enables', 'process', self.id_,
+ [[c.id_] for c in self.enables])
+ db_conn.rewrite_relations('process_disables', 'process', self.id_,
+ [[c.id_] for c in self.disables])
+ 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, row: Row) -> ProcessStep:
- """Make ProcessStep from database row."""
- return cls(row[0], row[1], row[2], row[3])
-
@classmethod
def by_id(cls, db_conn: DatabaseConnection, id_: int) -> ProcessStep:
"""Retrieve ProcessStep by id_, or throw NotFoundException."""
- for row in db_conn.exec('SELECT * FROM process_steps '
- 'WHERE step_id = ?', (id_,)):
- return cls.from_table_row(row)
+ step, _ = super()._by_id(db_conn, id_)
+ if step:
+ assert isinstance(step, ProcessStep)
+ return step
raise NotFoundException(f'found no ProcessStep of ID {id_}')
def save(self, db_conn: DatabaseConnection) -> None:
- """Save to database."""
- 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
-
-
-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