"""Actionables."""
from __future__ import annotations
-from dataclasses import dataclass
from typing import Any, Set
from sqlite3 import Row
+from plomtask.misc import DictableNode
from plomtask.db import DatabaseConnection, BaseModel
from plomtask.processes import Process, ProcessStepsNode
from plomtask.versioned_attributes import VersionedAttribute
from plomtask.dating import valid_date
-@dataclass
-class TodoNode:
+class TodoNode(DictableNode):
"""Collects what's useful to know for Todo/Condition tree display."""
+ # pylint: disable=too-few-public-methods
todo: Todo
seen: bool
children: list[TodoNode]
+ _to_dict = ['todo', 'seen', 'children']
+
+
+class TodoOrProcStepNode(DictableNode):
+ """Collect what's useful for Todo-or-ProcessStep tree display."""
+ # pylint: disable=too-few-public-methods
+ node_id: int
+ todo: Todo | None
+ process: Process | None
+ children: list[TodoOrProcStepNode] # pylint: disable=undefined-variable
+ fillable: bool = False
+ _to_dict = ['node_id', 'todo', 'process', 'children', 'fillable']
class Todo(BaseModel[int], ConditionsRelations):
# pylint: disable=too-many-instance-attributes
# pylint: disable=too-many-public-methods
table_name = 'todos'
- to_save = ['process_id', 'is_done', 'date', 'comment', 'effort',
- 'calendarize']
+ to_save_simples = ['process_id', 'is_done', 'date', 'comment', 'effort',
+ 'calendarize']
to_save_relations = [('todo_conditions', 'todo', 'conditions', 0),
('todo_blockers', 'todo', 'blockers', 0),
('todo_enables', 'todo', 'enables', 0),
days_to_update: Set[str] = set()
children: list[Todo]
parents: list[Todo]
+ sorters = {'doneness': lambda t: t.is_done,
+ 'title': lambda t: t.title_then,
+ 'comment': lambda t: t.comment,
+ 'date': lambda t: t.date}
# pylint: disable=too-many-arguments
def __init__(self, id_: int | None,
todos, _, _ = cls.by_date_range_with_limits(db_conn, date_range)
return todos
- @classmethod
- def create_with_children(cls, db_conn: DatabaseConnection,
- process_id: int, date: str) -> Todo:
- """Create Todo of process for date, ensure children."""
-
- def key_order_func(n: ProcessStepsNode) -> int:
- assert isinstance(n.process.id_, int)
- return n.process.id_
+ def ensure_children(self, db_conn: DatabaseConnection) -> None:
+ """Ensure Todo children (create or adopt) demanded by Process chain."""
def walk_steps(parent: Todo, step_node: ProcessStepsNode) -> Todo:
- adoptables = [t for t in cls.by_date(db_conn, date)
+ adoptables = [t for t in Todo.by_date(db_conn, parent.date)
if (t not in parent.children)
and (t != parent)
- and step_node.process == t.process]
+ and step_node.process.id_ == t.process_id]
satisfier = None
for adoptable in adoptables:
satisfier = adoptable
break
if not satisfier:
- satisfier = cls(None, step_node.process, False, date)
+ satisfier = Todo(None, step_node.process, False, parent.date)
satisfier.save(db_conn)
- sub_step_nodes = list(step_node.steps.values())
- sub_step_nodes.sort(key=key_order_func)
+ sub_step_nodes = sorted(
+ step_node.steps,
+ key=lambda s: s.process.id_ if s.process.id_ else 0)
for sub_node in sub_step_nodes:
if sub_node.is_suppressed:
continue
n_slots = len([n for n in sub_step_nodes
if n.process == sub_node.process])
filled_slots = len([t for t in satisfier.children
- if t.process == sub_node.process])
+ if t.process.id_ == sub_node.process.id_])
# if we did not newly create satisfier, it may already fill
# some step dependencies, so only fill what remains open
if n_slots - filled_slots > 0:
satisfier.save(db_conn)
return satisfier
- process = Process.by_id(db_conn, process_id)
- todo = cls(None, process, False, date)
- todo.save(db_conn)
+ process = Process.by_id(db_conn, self.process_id)
steps_tree = process.get_steps(db_conn)
- for step_node in steps_tree.values():
+ for step_node in steps_tree:
if step_node.is_suppressed:
continue
- todo.add_child(walk_steps(todo, step_node))
- todo.save(db_conn)
- return todo
+ self.add_child(walk_steps(self, step_node))
+ self.save(db_conn)
@classmethod
def from_table_row(cls, db_conn: DatabaseConnection,
return 0
@property
- def process_id(self) -> int | str | None:
+ def process_id(self) -> int:
"""Needed for super().save to save Processes as attributes."""
+ assert isinstance(self.process.id_, int)
return self.process.id_
@property
@property
def title(self) -> VersionedAttribute:
"""Shortcut to .process.title."""
+ assert isinstance(self.process.title, VersionedAttribute)
return self.process.title
@property
self.children.remove(child)
child.parents.remove(self)
+ def update_attrs(self, **kwargs: Any) -> None:
+ """Update self's attributes listed in kwargs."""
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+
def save(self, db_conn: DatabaseConnection) -> None:
"""On save calls, also check if auto-deletion by effort < 0."""
if self.effort and self.effort < 0 and self.is_deletable:
projects / plomtask / blobdiff