plomlompom.com Git - plomtask/blob - plomtask/todos.py

   1 """Actionables."""
   2 from __future__ import annotations
   3 from typing import Any
   4 from sqlite3 import Row
   5 from plomtask.db import DatabaseConnection, BaseModel
   6 from plomtask.processes import Process
   7 from plomtask.conditions import Condition
   8 from plomtask.exceptions import (NotFoundException, BadFormatException,
   9                                  HandledException)
  10
  11
  12 class Todo(BaseModel):
  13     """Individual actionable."""
  14
  15     # pylint: disable=too-many-instance-attributes
  16
  17     table_name = 'todos'
  18     to_save = ['process_id', 'is_done', 'date']
  19
  20     def __init__(self, id_: int | None, process: Process,
  21                  is_done: bool, date: str) -> None:
  22         self.set_int_id(id_)
  23         self.process = process
  24         self._is_done = is_done
  25         self.date = date
  26         self.children: list[Todo] = []
  27         self.parents: list[Todo] = []
  28         self.conditions: list[Condition] = []
  29         self.enables: list[Condition] = []
  30         self.disables: list[Condition] = []
  31         if not self.id_:
  32             self.conditions = process.conditions[:]
  33             self.enables = process.enables[:]
  34             self.disables = process.disables[:]
  35
  36     @classmethod
  37     def from_table_row(cls, db_conn: DatabaseConnection,
  38                        row: Row | list[Any]) -> Todo:
  39         """Make from DB row, write to DB cache."""
  40         if row[1] == 0:
  41             raise NotFoundException('calling Todo of '
  42                                     'unsaved Process')
  43         row_as_list = list(row)
  44         row_as_list[1] = Process.by_id(db_conn, row[1])
  45         todo = super().from_table_row(db_conn, row_as_list)
  46         assert isinstance(todo, Todo)
  47         return todo
  48
  49     @classmethod
  50     def by_id(cls, db_conn: DatabaseConnection, id_: int) -> Todo:
  51         """Get Todo of .id_=id_ and children (from DB cache if possible)."""
  52         todo, from_cache = super()._by_id(db_conn, id_)
  53         if todo is None:
  54             raise NotFoundException(f'Todo of ID not found: {id_}')
  55         if not from_cache:
  56             for t_id in db_conn.column_where('todo_children', 'child',
  57                                              'parent', id_):
  58                 todo.children += [cls.by_id(db_conn, t_id)]
  59             for t_id in db_conn.column_where('todo_children', 'parent',
  60                                              'child', id_):
  61                 todo.parents += [cls.by_id(db_conn, t_id)]
  62             for name in ('conditions', 'enables', 'disables'):
  63                 table = f'todo_{name}'
  64                 for cond_id in db_conn.column_where(table, 'condition',
  65                                                     'todo', todo.id_):
  66                     target = getattr(todo, name)
  67                     target += [Condition.by_id(db_conn, cond_id)]
  68         assert isinstance(todo, Todo)
  69         return todo
  70
  71     @classmethod
  72     def by_date(cls, db_conn: DatabaseConnection, date: str) -> list[Todo]:
  73         """Collect all Todos for Day of date."""
  74         todos = []
  75         for id_ in db_conn.column_where('todos', 'id', 'day', date):
  76             todos += [cls.by_id(db_conn, id_)]
  77         return todos
  78
  79     @staticmethod
  80     def _x_ablers_for_at(db_conn: DatabaseConnection, name: str,
  81                          cond: Condition, date: str) -> list[Todo]:
  82         """Collect all Todos of day that [name] condition."""
  83         assert isinstance(cond.id_, int)
  84         x_ablers = []
  85         table = f'todo_{name}'
  86         for id_ in db_conn.column_where(table, 'todo', 'condition', cond.id_):
  87             todo = Todo.by_id(db_conn, id_)
  88             if todo.date == date:
  89                 x_ablers += [todo]
  90         return x_ablers
  91
  92     @classmethod
  93     def enablers_for_at(cls, db_conn: DatabaseConnection,
  94                         condition: Condition, date: str) -> list[Todo]:
  95         """Collect all Todos of day that enable condition."""
  96         return cls._x_ablers_for_at(db_conn, 'enables', condition, date)
  97
  98     @classmethod
  99     def disablers_for_at(cls, db_conn: DatabaseConnection,
 100                          condition: Condition, date: str) -> list[Todo]:
 101         """Collect all Todos of day that disable condition."""
 102         return cls._x_ablers_for_at(db_conn, 'disables', condition, date)
 103
 104     @property
 105     def is_doable(self) -> bool:
 106         """Decide whether .is_done settable based on children, Conditions."""
 107         for child in self.children:
 108             if not child.is_done:
 109                 return False
 110         for condition in self.conditions:
 111             if not condition.is_active:
 112                 return False
 113         return True
 114
 115     @property
 116     def process_id(self) -> int | str | None:
 117         """Return ID of tasked Process."""
 118         return self.process.id_
 119
 120     @property
 121     def is_done(self) -> bool:
 122         """Wrapper around self._is_done so we can control its setter."""
 123         return self._is_done
 124
 125     @is_done.setter
 126     def is_done(self, value: bool) -> None:
 127         if value != self.is_done and not self.is_doable:
 128             raise BadFormatException('cannot change doneness of undoable Todo')
 129         if self._is_done != value:
 130             self._is_done = value
 131             if value is True:
 132                 for condition in self.enables:
 133                     condition.is_active = True
 134                 for condition in self.disables:
 135                     condition.is_active = False
 136
 137     def set_disables(self, db_conn: DatabaseConnection,
 138                      ids: list[int]) -> None:
 139         """Set self.disables to Conditions identified by ids."""
 140         self.set_conditions(db_conn, ids, 'disables')
 141
 142     def set_enables(self, db_conn: DatabaseConnection,
 143                     ids: list[int]) -> None:
 144         """Set self.enables to Conditions identified by ids."""
 145         self.set_conditions(db_conn, ids, 'enables')
 146
 147     def set_conditions(self, db_conn: DatabaseConnection, ids: list[int],
 148                        target: str = 'conditions') -> None:
 149         """Set self.[target] to Conditions identified by ids."""
 150         target_list = getattr(self, target)
 151         while len(target_list) > 0:
 152             target_list.pop()
 153         for id_ in ids:
 154             target_list += [Condition.by_id(db_conn, id_)]
 155
 156     def add_child(self, child: Todo) -> None:
 157         """Add child to self.children, guard against recursion"""
 158         def walk_steps(node: Todo) -> None:
 159             if node.id_ == self.id_:
 160                 raise BadFormatException('bad child choice causes recursion')
 161             for child in node.children:
 162                 walk_steps(child)
 163         if self.id_ is None:
 164             raise HandledException('Can only add children to saved Todos.')
 165         if child.id_ is None:
 166             raise HandledException('Can only add saved children to Todos.')
 167         if child in self.children:
 168             raise BadFormatException('cannot adopt same child twice')
 169         walk_steps(child)
 170         self.children += [child]
 171         child.parents += [self]
 172
 173     def save(self, db_conn: DatabaseConnection) -> None:
 174         """Write self and children to DB and its cache."""
 175         if self.process.id_ is None:
 176             raise NotFoundException('Process of Todo without ID (not saved?)')
 177         self.save_core(db_conn)
 178         assert isinstance(self.id_, int)
 179         db_conn.cached_todos[self.id_] = self
 180         db_conn.rewrite_relations('todo_children', 'parent', self.id_,
 181                                   [[c.id_] for c in self.children])
 182         db_conn.rewrite_relations('todo_conditions', 'todo', self.id_,
 183                                   [[c.id_] for c in self.conditions])
 184         db_conn.rewrite_relations('todo_enables', 'todo', self.id_,
 185                                   [[c.id_] for c in self.enables])
 186         db_conn.rewrite_relations('todo_disables', 'todo', self.id_,
 187                                   [[c.id_] for c in self.disables])