"""Collecting Processes and Process-related items."""
from __future__ import annotations
+from dataclasses import dataclass
+from typing import Set, Any
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:
- """Template for, and metadata for, Todos, and their arrangements."""
+@dataclass
+class ProcessStepsNode:
+ """Collects what's useful to know for ProcessSteps tree display."""
+ process: Process
+ parent_id: int | None
+ is_explicit: bool
+ steps: dict[int, ProcessStepsNode]
+ seen: bool
- 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.explicit_steps: list[ProcessStep] = []
- def __eq__(self, other: object) -> bool:
- return isinstance(other, self.__class__) and self.id_ == other.id_
+class Process(BaseModel[int], ConditionsRelations):
+ """Template for, and metadata for, Todos, and their arrangements."""
+ table_name = 'processes'
- @classmethod
- def from_table_row(cls, row: Row) -> Process:
- """Make Process from database row, with empty VersionedAttributes."""
- return cls(row[0])
+ # pylint: disable=too-many-instance-attributes
- @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]
- return list(processes.values())
+ def __init__(self, id_: int | None) -> None:
+ super().__init__(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] = []
+ self.conditions: list[Condition] = []
+ self.enables: list[Condition] = []
+ self.disables: list[Condition] = []
@classmethod
- def by_id(cls, db_conn: DatabaseConnection, id_: int | None,
- 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 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)]
+ def from_table_row(cls, db_conn: DatabaseConnection,
+ row: Row | list[Any]) -> Process:
+ """Make from DB row, with dependencies."""
+ process = super().from_table_row(db_conn, row)
+ assert 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] # pylint: disable=no-member
+ for name in ('conditions', 'enables', 'disables'):
+ table = f'process_{name}'
+ assert isinstance(process.id_, int)
+ for c_id in db_conn.column_where(table, 'condition',
+ 'process', process.id_):
+ target = getattr(process, name)
+ target += [Condition.by_id(db_conn, c_id)]
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]]:
+ Process | None = None) -> dict[int, ProcessStepsNode]:
"""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)
+ def make_node(step: ProcessStep) -> ProcessStepsNode:
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,
- 'is_explicit': is_explicit, 'steps': step_steps}
+ process = self.__class__.by_id(db_conn, step.step_process_id)
+ step_steps = process.get_steps(db_conn, external_owner)
+ return ProcessStepsNode(process, step.parent_step_id,
+ is_explicit, step_steps, False)
- def walk_steps(node_id: int, node: dict[str, Any]) -> None:
+ def walk_steps(node_id: int, node: ProcessStepsNode) -> None:
explicit_children = [s for s in self.explicit_steps
if s.parent_step_id == node_id]
for child in explicit_children:
- node['steps'][child.id_] = make_node(child)
- node['seen'] = node_id in seen_step_ids
+ assert isinstance(child.id_, int)
+ node.steps[child.id_] = make_node(child)
+ node.seen = node_id in seen_step_ids
seen_step_ids.add(node_id)
- for id_, step in node['steps'].items():
+ for id_, step in node.steps.items():
walk_steps(id_, step)
- steps: dict[int, dict[str, object]] = {}
+ steps: dict[int, ProcessStepsNode] = {}
seen_step_ids: Set[int] = set()
if external_owner is 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
owned by the current Process.
"""
+
def walk_steps(node: ProcessStep) -> None:
if node.step_process_id == self.id_:
raise BadFormatException('bad step selection causes recursion')
step_process = self.by_id(db_conn, node.step_process_id)
for step in step_process.explicit_steps:
walk_steps(step)
+
if parent_step_id is not None:
try:
parent_step = ProcessStep.by_id(db_conn, parent_step_id)
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:
+ step.uncache()
+ 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)
-class ProcessStep:
+class ProcessStep(BaseModel[int]):
"""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_
+ super().__init__(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)
- 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