Hide (almost all) remaining SQL code in DB module.

[plomtask] / plomtask / todos.py

diff --git a/plomtask/todos.py b/plomtask/todos.py
index 348dbddcbb96338ba1f1e10da50e094e36b92f63..fd72af6bf8c0842f2a2727185c2b4c3f707a20d4 100644 (file)
--- a/plomtask/todos.py
+++ b/plomtask/todos.py
@@ -47,30 +47,24 @@ class Todo(BaseModel):
         return todo
 
     @classmethod
-    def by_id(cls, db_conn: DatabaseConnection, id_: int | None) -> Todo:
+    def by_id(cls, db_conn: DatabaseConnection, id_: int) -> Todo:
         """Get Todo of .id_=id_ and children (from DB cache if possible)."""
-        if id_:
-            todo, from_cache = super()._by_id(db_conn, id_)
-        else:
-            todo, from_cache = None, False
+        todo, from_cache = super()._by_id(db_conn, id_)
         if todo is None:
             raise NotFoundException(f'Todo of ID not found: {id_}')
         if not from_cache:
-            for row in db_conn.exec('SELECT child FROM todo_children '
-                                    'WHERE parent = ?', (id_,)):
-                todo.children += [cls.by_id(db_conn, row[0])]
-            for row in db_conn.exec('SELECT parent FROM todo_children '
-                                    'WHERE child = ?', (id_,)):
-                todo.parents += [cls.by_id(db_conn, row[0])]
-            for row in db_conn.exec('SELECT condition FROM todo_conditions '
-                                    'WHERE todo = ?', (id_,)):
-                todo.conditions += [Condition.by_id(db_conn, row[0])]
-            for row in db_conn.exec('SELECT condition FROM todo_fulfills '
-                                    'WHERE todo = ?', (id_,)):
-                todo.fulfills += [Condition.by_id(db_conn, row[0])]
-            for row in db_conn.exec('SELECT condition FROM todo_undoes '
-                                    'WHERE todo = ?', (id_,)):
-                todo.undoes += [Condition.by_id(db_conn, row[0])]
+            for t_id in db_conn.column_where('todo_children', 'child',
+                                             'parent', id_):
+                todo.children += [cls.by_id(db_conn, t_id)]
+            for t_id in db_conn.column_where('todo_children', 'parent',
+                                             'child', id_):
+                todo.parents += [cls.by_id(db_conn, t_id)]
+            for name in ('conditions', 'fulfills', 'undoes'):
+                table = f'todo_{name}'
+                for cond_id in db_conn.column_where(table, 'condition',
+                                                    'todo', todo.id_):
+                    target = getattr(todo, name)
+                    target += [Condition.by_id(db_conn, cond_id)]
         assert isinstance(todo, Todo)
         return todo
 
@@ -78,18 +72,19 @@ class Todo(BaseModel):
     def by_date(cls, db_conn: DatabaseConnection, date: str) -> list[Todo]:
         """Collect all Todos for Day of date."""
         todos = []
-        for row in db_conn.exec('SELECT id FROM todos WHERE day = ?', (date,)):
-            todos += [cls.by_id(db_conn, row[0])]
+        for id_ in db_conn.column_where('todos', 'id', 'day', date):
+            todos += [cls.by_id(db_conn, id_)]
         return todos
 
     @classmethod
     def enablers_for_at(cls, db_conn: DatabaseConnection, condition: Condition,
                         date: str) -> list[Todo]:
         """Collect all Todos of day that enable condition."""
+        assert isinstance(condition.id_, int)
         enablers = []
-        for row in db_conn.exec('SELECT todo FROM todo_fulfills '
-                                'WHERE condition = ?', (condition.id_,)):
-            todo = cls.by_id(db_conn, row[0])
+        for id_ in db_conn.column_where('todo_fulfills', 'todo', 'condition',
+                                        condition.id_):
+            todo = cls.by_id(db_conn, id_)
             if todo.date == date:
                 enablers += [todo]
         return enablers
@@ -98,10 +93,11 @@ class Todo(BaseModel):
     def disablers_for_at(cls, db_conn: DatabaseConnection,
                          condition: Condition, date: str) -> list[Todo]:
         """Collect all Todos of day that disable condition."""
+        assert isinstance(condition.id_, int)
         disablers = []
-        for row in db_conn.exec('SELECT todo FROM todo_undoes '
-                                'WHERE condition = ?', (condition.id_,)):
-            todo = cls.by_id(db_conn, row[0])
+        for id_ in db_conn.column_where('todo_undoes', 'todo', 'condition',
+                                        condition.id_):
+            todo = cls.by_id(db_conn, id_)
             if todo.date == date:
                 disablers += [todo]
         return disablers