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

   1 """Actionables."""
   2 from __future__ import annotations
   3 from sqlite3 import Row
   4 from plomtask.db import DatabaseConnection
   5 from plomtask.days import Day
   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:
  13     """Individual actionable."""
  14
  15     # pylint: disable=too-many-instance-attributes
  16
  17     def __init__(self, id_: int | None, process: Process,
  18                  is_done: bool, day: Day) -> None:
  19         self.id_ = id_
  20         self.process = process
  21         self._is_done = is_done
  22         self.day = day
  23         self.children: list[Todo] = []
  24         self.parents: list[Todo] = []
  25         self.conditions: list[Condition] = []
  26         self.fulfills: list[Condition] = []
  27         self.undoes: list[Condition] = []
  28         if not self.id_:
  29             self.conditions = process.conditions[:]
  30             self.fulfills = process.fulfills[:]
  31             self.undoes = process.undoes[:]
  32
  33     @classmethod
  34     def from_table_row(cls, db_conn: DatabaseConnection, row: Row) -> Todo:
  35         """Make Todo from database row, write to DB cache."""
  36         todo = cls(id_=row[0],
  37                    process=Process.by_id(db_conn, row[1]),
  38                    is_done=bool(row[2]),
  39                    day=Day.by_date(db_conn, row[3]))
  40         assert todo.id_ is not None
  41         db_conn.cached_todos[todo.id_] = todo
  42         return todo
  43
  44     @classmethod
  45     def by_id(cls, db_conn: DatabaseConnection, id_: int | None) -> Todo:
  46         """Get Todo of .id_=id_ and children (from DB cache if possible)."""
  47         if id_ in db_conn.cached_todos.keys():
  48             todo = db_conn.cached_todos[id_]
  49         else:
  50             todo = None
  51             for row in db_conn.exec('SELECT * FROM todos WHERE id = ?',
  52                                     (id_,)):
  53                 todo = cls.from_table_row(db_conn, row)
  54                 break
  55             if todo is None:
  56                 raise NotFoundException(f'Todo of ID not found: {id_}')
  57             for row in db_conn.exec('SELECT child FROM todo_children '
  58                                     'WHERE parent = ?', (id_,)):
  59                 todo.children += [cls.by_id(db_conn, row[0])]
  60             for row in db_conn.exec('SELECT parent FROM todo_children '
  61                                     'WHERE child = ?', (id_,)):
  62                 todo.parents += [cls.by_id(db_conn, row[0])]
  63             for row in db_conn.exec('SELECT condition FROM todo_conditions '
  64                                     'WHERE todo = ?', (id_,)):
  65                 todo.conditions += [Condition.by_id(db_conn, row[0])]
  66             for row in db_conn.exec('SELECT condition FROM todo_fulfills '
  67                                     'WHERE todo = ?', (id_,)):
  68                 todo.fulfills += [Condition.by_id(db_conn, row[0])]
  69             for row in db_conn.exec('SELECT condition FROM todo_undoes '
  70                                     'WHERE todo = ?', (id_,)):
  71                 todo.undoes += [Condition.by_id(db_conn, row[0])]
  72         assert isinstance(todo, Todo)
  73         return todo
  74
  75     @classmethod
  76     def by_date(cls, db_conn: DatabaseConnection, date: str) -> list[Todo]:
  77         """Collect all Todos for Day of date."""
  78         todos = []
  79         for row in db_conn.exec('SELECT id FROM todos WHERE day = ?', (date,)):
  80             todos += [cls.by_id(db_conn, row[0])]
  81         return todos
  82
  83     @classmethod
  84     def enablers_for_at(cls, db_conn: DatabaseConnection, condition: Condition,
  85                         date: str) -> list[Todo]:
  86         """Collect all Todos of day that enable condition."""
  87         enablers = []
  88         for row in db_conn.exec('SELECT todo FROM todo_fulfills '
  89                                 'WHERE condition = ?', (condition.id_,)):
  90             todo = cls.by_id(db_conn, row[0])
  91             if todo.day.date == date:
  92                 enablers += [todo]
  93         return enablers
  94
  95     @classmethod
  96     def disablers_for_at(cls, db_conn: DatabaseConnection,
  97                          condition: Condition, date: str) -> list[Todo]:
  98         """Collect all Todos of day that disable condition."""
  99         disablers = []
 100         for row in db_conn.exec('SELECT todo FROM todo_undoes '
 101                                 'WHERE condition = ?', (condition.id_,)):
 102             todo = cls.by_id(db_conn, row[0])
 103             if todo.day.date == date:
 104                 disablers += [todo]
 105         return disablers
 106
 107     @property
 108     def is_doable(self) -> bool:
 109         """Decide whether .is_done settable based on children, Conditions."""
 110         for child in self.children:
 111             if not child.is_done:
 112                 return False
 113         for condition in self.conditions:
 114             if not condition.is_active:
 115                 return False
 116         return True
 117
 118     @property
 119     def is_done(self) -> bool:
 120         """Wrapper around self._is_done so we can control its setter."""
 121         return self._is_done
 122
 123     @is_done.setter
 124     def is_done(self, value: bool) -> None:
 125         if value != self.is_done and not self.is_doable:
 126             raise BadFormatException('cannot change doneness of undoable Todo')
 127         if self._is_done != value:
 128             self._is_done = value
 129             if value is True:
 130                 for condition in self.fulfills:
 131                     condition.is_active = True
 132                 for condition in self.undoes:
 133                     condition.is_active = False
 134
 135     def set_undoes(self, db_conn: DatabaseConnection, ids: list[int]) -> None:
 136         """Set self.undoes to Conditions identified by ids."""
 137         self.set_conditions(db_conn, ids, 'undoes')
 138
 139     def set_fulfills(self, db_conn: DatabaseConnection,
 140                      ids: list[int]) -> None:
 141         """Set self.fulfills to Conditions identified by ids."""
 142         self.set_conditions(db_conn, ids, 'fulfills')
 143
 144     def set_conditions(self, db_conn: DatabaseConnection, ids: list[int],
 145                        target: str = 'conditions') -> None:
 146         """Set self.[target] to Conditions identified by ids."""
 147         target_list = getattr(self, target)
 148         while len(target_list) > 0:
 149             target_list.pop()
 150         for id_ in ids:
 151             target_list += [Condition.by_id(db_conn, id_)]
 152
 153     def add_child(self, child: Todo) -> None:
 154         """Add child to self.children, guard against recursion"""
 155         def walk_steps(node: Todo) -> None:
 156             if node.id_ == self.id_:
 157                 raise BadFormatException('bad child choice causes recursion')
 158             for child in node.children:
 159                 walk_steps(child)
 160         if self.id_ is None:
 161             raise HandledException('Can only add children to saved Todos.')
 162         if child.id_ is None:
 163             raise HandledException('Can only add saved children to Todos.')
 164         if child in self.children:
 165             raise BadFormatException('cannot adopt same child twice')
 166         walk_steps(child)
 167         self.children += [child]
 168         child.parents += [self]
 169
 170     def save(self, db_conn: DatabaseConnection) -> None:
 171         """Write self and children to DB and its cache."""
 172         if self.process.id_ is None:
 173             raise NotFoundException('Process of Todo without ID (not saved?)')
 174         cursor = db_conn.exec('REPLACE INTO todos VALUES (?,?,?,?)',
 175                               (self.id_, self.process.id_,
 176                                self.is_done, self.day.date))
 177         self.id_ = cursor.lastrowid
 178         assert self.id_ is not None
 179         db_conn.cached_todos[self.id_] = self
 180         db_conn.exec('DELETE FROM todo_children WHERE parent = ?',
 181                      (self.id_,))
 182         for child in self.children:
 183             db_conn.exec('INSERT INTO todo_children VALUES (?, ?)',
 184                          (self.id_, child.id_))
 185         db_conn.exec('DELETE FROM todo_fulfills WHERE todo = ?', (self.id_,))
 186         for condition in self.fulfills:
 187             if condition.id_ is None:
 188                 raise NotFoundException('Fulfilled Condition of Todo '
 189                                         'without ID (not saved?)')
 190             db_conn.exec('INSERT INTO todo_fulfills VALUES (?, ?)',
 191                          (self.id_, condition.id_))
 192         db_conn.exec('DELETE FROM todo_undoes WHERE todo = ?', (self.id_,))
 193         for condition in self.undoes:
 194             if condition.id_ is None:
 195                 raise NotFoundException('Undone Condition of Todo '
 196                                         'without ID (not saved?)')
 197             db_conn.exec('INSERT INTO todo_undoes VALUES (?, ?)',
 198                          (self.id_, condition.id_))
 199         db_conn.exec('DELETE FROM todo_conditions WHERE todo = ?', (self.id_,))
 200         for condition in self.conditions:
 201             if condition.id_ is None:
 202                 raise NotFoundException('Condition of Todo '
 203                                         'without ID (not saved?)')
 204             db_conn.exec('INSERT INTO todo_conditions VALUES (?, ?)',
 205                          (self.id_, condition.id_))