Minor fixes.

[plomtask] / plomtask / processes.py

diff --git a/plomtask/processes.py b/plomtask/processes.py
index 76ed30df049b5d801b283ab904364da27f05f240..32eee4d350db5ed78f4501f7016688f5c174c7cb 100644 (file)
--- a/plomtask/processes.py
+++ b/plomtask/processes.py
@@ -1,37 +1,38 @@
 """Collecting Processes and Process-related items."""
 from __future__ import annotations
 """Collecting Processes and Process-related items."""
 from __future__ import annotations

-from sqlite3 import Row
-from typing import Any, Set
-from plomtask.db import DatabaseConnection
+from dataclasses import dataclass
+from typing import Set
+from plomtask.db import DatabaseConnection, BaseModel

 from plomtask.misc import VersionedAttribute
 from plomtask.misc import VersionedAttribute

-from plomtask.conditions import Condition
+from plomtask.conditions import Condition, ConditionsRelations

 from plomtask.exceptions import NotFoundException, BadFormatException
 
 
 from plomtask.exceptions import NotFoundException, BadFormatException
 
 

-class Process:
+@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
+
+
+class Process(BaseModel, ConditionsRelations):

     """Template for, and metadata for, Todos, and their arrangements."""
     """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:
 
     # 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.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] = []
         self.conditions: list[Condition] = []
         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.fulfills: list[Condition] = []
-        self.undoes: list[Condition] = []
-
-    @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.enables: list[Condition] = []
+        self.disables: list[Condition] = []

 
     @classmethod
     def all(cls, db_conn: DatabaseConnection) -> list[Process]:
 
     @classmethod
     def all(cls, db_conn: DatabaseConnection) -> list[Process]:


@@ -40,9 +41,9 @@ class Process:
         for id_, process in db_conn.cached_processes.items():
             processes[id_] = process
         already_recorded = processes.keys()
         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())
 
                 processes[process.id_] = process
         return list(processes.values())
 


@@ -50,105 +51,77 @@ class Process:
     def by_id(cls, db_conn: DatabaseConnection, id_: int | None,
               create: bool = False) -> Process:
         """Collect Process, its VersionedAttributes, and its child IDs."""
     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
         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 not process:
             if not create:
                 raise NotFoundException(f'Process not found of id: {id_}')
             process = Process(id_)

-        for row in db_conn.exec('SELECT * FROM process_titles '
-                                'WHERE parent_id = ?', (process.id_,)):
-            process.title.history[row[1]] = row[2]
-        for row in db_conn.exec('SELECT * FROM process_descriptions '
-                                'WHERE parent_id = ?', (process.id_,)):
-            process.description.history[row[1]] = row[2]
-        for row in db_conn.exec('SELECT * FROM process_efforts '
-                                'WHERE parent_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)]
-        for row in db_conn.exec('SELECT condition FROM process_conditions '
-                                'WHERE process = ?', (process.id_,)):
-            process.conditions += [Condition.by_id(db_conn, row[0])]
-        for row in db_conn.exec('SELECT condition FROM process_fulfills '
-                                'WHERE process = ?', (process.id_,)):
-            process.fulfills += [Condition.by_id(db_conn, row[0])]
-        for row in db_conn.exec('SELECT condition FROM process_undoes '
-                                'WHERE process = ?', (process.id_,)):
-            process.undoes += [Condition.by_id(db_conn, row[0])]
+        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."""
         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()
         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 [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."""
 
         """Return tree of depended-on explicit and implicit ProcessSteps."""
 

-        def make_node(step: ProcessStep) -> dict[str, object]:
+        def make_node(step: ProcessStep) -> ProcessStepsNode:

             is_explicit = False
             if external_owner is not None:
                 is_explicit = step.owner_id == external_owner.id_
             process = self.__class__.by_id(db_conn, step.step_process_id)
             step_steps = process.get_steps(db_conn, external_owner)
             is_explicit = False
             if external_owner is not None:
                 is_explicit = step.owner_id == external_owner.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}
+            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:
             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)
             seen_step_ids.add(node_id)

-            for id_, step in node['steps'].items():
+            for id_, step in node.steps.items():

                 walk_steps(id_, step)
 
                 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]:
         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
 
             steps[step.id_] = make_node(step)
         for step_id, step_node in steps.items():
             walk_steps(step_id, step_node)
         return steps
 

-    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,
     def _add_step(self,
                   db_conn: DatabaseConnection,
                   id_: int | None,


@@ -163,12 +136,14 @@ class Process:
         just deleted under its feet), or if the parent step would not be
         owned by the current Process.
         """
         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)
         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)
         if parent_step_id is not None:
             try:
                 parent_step = ProcessStep.by_id(db_conn, parent_step_id)


@@ -176,7 +151,7 @@ class Process:
                     parent_step_id = None
             except NotFoundException:
                 parent_step_id = None
                     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 = ProcessStep(id_, self.id_, step_process_id, parent_step_id)
         walk_steps(step)
         self.explicit_steps += [step]


@@ -186,86 +161,56 @@ class Process:
     def set_steps(self, db_conn: DatabaseConnection,
                   steps: list[tuple[int | None, int, int | None]]) -> None:
         """Set self.explicit_steps in bulk."""
     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:
         for step in self.explicit_steps:

-            assert step.id_ is not None
+            assert isinstance(step.id_, int)

             del db_conn.cached_process_steps[step.id_]
         self.explicit_steps = []
             del db_conn.cached_process_steps[step.id_]
         self.explicit_steps = []

-        db_conn.exec('DELETE FROM process_steps WHERE owner_id = ?',
-                     (self.id_,))
+        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])
 
         for step_tuple in steps:
             self._add_step(db_conn, step_tuple[0],
                            step_tuple[1], step_tuple[2])
 

-    def save_id(self, db_conn: DatabaseConnection) -> None:
-        """Write bare-bones self (sans connected items), ensuring self.id_."""
-        cursor = db_conn.exec('REPLACE INTO processes VALUES (?)', (self.id_,))
-        self.id_ = cursor.lastrowid
-

     def save(self, db_conn: DatabaseConnection) -> None:
         """Add (or re-write) self and connected items to DB."""
     def save(self, db_conn: DatabaseConnection) -> None:
         """Add (or re-write) self and connected items to DB."""

-        self.save_id(db_conn)
+        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)
         self.title.save(db_conn)
         self.description.save(db_conn)
         self.effort.save(db_conn)

-        db_conn.exec('DELETE FROM process_conditions WHERE process = ?',
-                     (self.id_,))
-        for condition in self.conditions:
-            db_conn.exec('INSERT INTO process_conditions VALUES (?,?)',
-                         (self.id_, condition.id_))
-        db_conn.exec('DELETE FROM process_fulfills WHERE process = ?',
-                     (self.id_,))
-        for condition in self.fulfills:
-            db_conn.exec('INSERT INTO process_fulfills VALUES (?,?)',
-                         (self.id_, condition.id_))
-        db_conn.exec('DELETE FROM process_undoes WHERE process = ?',
-                     (self.id_,))
-        for condition in self.undoes:
-            db_conn.exec('INSERT INTO process_undoes VALUES (?,?)',
-                         (self.id_, condition.id_))
-        assert self.id_ is not 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:
             step.save(db_conn)
         db_conn.cached_processes[self.id_] = self
 
 
         for step in self.explicit_steps:
             step.save(db_conn)
         db_conn.cached_processes[self.id_] = self
 
 

-class ProcessStep:
+class ProcessStep(BaseModel):

     """Sub-unit of Processes."""
     """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:
 
     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
 
         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."""
     @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
             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:
         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
+        """Default to simply calling self.save_core for simple cases."""
+        self.save_core(db_conn)