from __future__ import annotations
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
-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']
- def __init__(self,
- todo: Todo,
- seen: bool,
- children: list[TodoNode]) -> None:
- self.todo = todo
- self.seen = seen
- self.children = children
- @property
- def as_dict(self) -> dict[str, object]:
- """Return self as (json.dumps-coompatible) dict."""
- return {'todo': self.todo.id_,
- 'seen': self.seen,
- 'children': [c.as_dict for c in self.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):
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
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: