"""Non-Process dependency for ProcessSteps and Todos."""
table_name = 'conditions'
to_save = ['is_active']
+ to_save_versioned = ['title', 'description']
def __init__(self, id_: int | None, is_active: bool = False) -> None:
super().__init__(id_)
getattr(condition, name).history_from_row(row_)
return condition
- def save(self, db_conn: DatabaseConnection) -> None:
- """Save self and its VersionedAttributes to DB and cache."""
- self.save_core(db_conn)
- self.title.save(db_conn)
- self.description.save(db_conn)
-
def remove(self, db_conn: DatabaseConnection) -> None:
"""Remove from DB, with VersionedAttributes.
table_name = f'{item}_{attr}'
for _ in db_conn.row_where(table_name, 'condition', self.id_):
raise HandledException('cannot remove Condition in use')
- db_conn.delete_where('condition_titles', 'parent', self.id_)
- db_conn.delete_where('condition_descriptions', 'parent', self.id_)
super().remove(db_conn)
"""Return date succeeding date of this Day."""
next_datetime = self.datetime + timedelta(days=1)
return next_datetime.strftime(DATE_FORMAT)
-
- def save(self, db_conn: DatabaseConnection) -> None:
- """Add (or re-write) self to DB and cache."""
- self.save_core(db_conn)
"""Close DB connection."""
self.conn.close()
- def rewrite_relations(self, table_name: str, key: str, target: int,
+ def rewrite_relations(self, table_name: str, key: str, target: int | str,
rows: list[list[Any]]) -> None:
"""Rewrite relations in table_name to target, with rows values."""
self.delete_where(table_name, key, target)
"""Template for most of the models we use/derive from the DB."""
table_name = ''
to_save: list[str] = []
+ to_save_versioned: list[str] = []
+ to_save_relations: list[tuple[str, str, str]] = []
id_: None | BaseModelId
cache_: dict[BaseModelId, Self]
items[item.id_] = item
return list(items.values())
- def save_core(self, db_conn: DatabaseConnection) -> None:
- """Write bare-bones self (sans connected items), ensuring self.id_.
+ def save(self, db_conn: DatabaseConnection) -> None:
+ """Write self to DB and cache and ensure .id_.
Write both to DB, and to cache. To DB, write .id_ and attributes
- listed in cls.to_save.
+ listed in cls.to_save[_versioned|_relations].
Ensure self.id_ by setting it to what the DB command returns as the
last saved row's ID (cursor.lastrowid), EXCEPT if self.id_ already
if not isinstance(self.id_, str):
self.id_ = cursor.lastrowid # type: ignore[assignment]
self.cache()
+ for attr_name in self.to_save_versioned:
+ getattr(self, attr_name).save(db_conn)
+ for table, column, attr_name in self.to_save_relations:
+ assert isinstance(self.id_, (int, str))
+ db_conn.rewrite_relations(table, column, self.id_,
+ [[i.id_] for i
+ in getattr(self, attr_name)])
def remove(self, db_conn: DatabaseConnection) -> None:
- """Remove from DB and cache."""
+ """Remove from DB and cache, including dependencies."""
if self.id_ is None or self.__class__.get_cached(self.id_) is None:
raise HandledException('cannot remove unsaved item')
+ for attr_name in self.to_save_versioned:
+ getattr(self, attr_name).remove(db_conn)
+ for table, column, attr_name in self.to_save_relations:
+ db_conn.delete_where(table, column, self.id_)
self.uncache()
db_conn.delete_where(self.table_name, 'id', self.id_)
self.form_data.get_all_int('condition'))
process.set_enables(self.conn, self.form_data.get_all_int('enables'))
process.set_disables(self.conn, self.form_data.get_all_int('disables'))
- process.save_core(self.conn)
+ process.save(self.conn)
process.explicit_steps = []
steps: list[tuple[int | None, int, int | None]] = []
for step_id in self.form_data.get_all_int('steps'):
class Process(BaseModel[int], ConditionsRelations):
"""Template for, and metadata for, Todos, and their arrangements."""
table_name = 'processes'
+ to_save_versioned = ['title', 'description', 'effort']
+ to_save_relations = [('process_conditions', 'process', 'conditions'),
+ ('process_enables', 'process', 'enables'),
+ ('process_disables', 'process', 'disables')]
# pylint: disable=too-many-instance-attributes
def save(self, db_conn: DatabaseConnection) -> None:
"""Add (or re-write) self and connected items to DB."""
- self.save_core(db_conn)
+ super().save(db_conn)
assert isinstance(self.id_, int)
- self.title.save(db_conn)
- self.description.save(db_conn)
- self.effort.save(db_conn)
- db_conn.rewrite_relations('process_conditions', 'process', self.id_,
- [[c.id_] for c in self.conditions])
- db_conn.rewrite_relations('process_enables', 'process', self.id_,
- [[c.id_] for c in self.enables])
- db_conn.rewrite_relations('process_disables', 'process', self.id_,
- [[c.id_] for c in self.disables])
db_conn.delete_where('process_steps', 'owner', self.id_)
for step in self.explicit_steps:
step.save(db_conn)
raise HandledException('cannot remove Process in use')
for _ in db_conn.row_where('todos', 'process', self.id_):
raise HandledException('cannot remove Process in use')
- db_conn.delete_where('process_conditions', 'process', self.id_)
- db_conn.delete_where('process_enables', 'process', self.id_)
- db_conn.delete_where('process_disables', 'process', self.id_)
for step in self.explicit_steps:
step.remove(db_conn)
- db_conn.delete_where('process_titles', 'parent', self.id_)
- db_conn.delete_where('process_descriptions', 'parent', self.id_)
- db_conn.delete_where('process_efforts', 'parent', self.id_)
super().remove(db_conn)
self.step_process_id = step_process_id
self.parent_step_id = parent_step_id
- def save(self, db_conn: DatabaseConnection) -> None:
- """Default to simply calling self.save_core for simple cases."""
- self.save_core(db_conn)
-
def remove(self, db_conn: DatabaseConnection) -> None:
"""Remove from DB, and owner's .explicit_steps."""
owner = Process.by_id(db_conn, self.owner_id)
# pylint: disable=too-many-instance-attributes
table_name = 'todos'
to_save = ['process_id', 'is_done', 'date']
+ to_save_relations = [('todo_conditions', 'todo', 'conditions'),
+ ('todo_enables', 'todo', 'enables'),
+ ('todo_disables', 'todo', 'disables'),
+ ('todo_children', 'parent', 'children'),
+ ('todo_children', 'child', 'parents')]
def __init__(self, id_: int | None, process: Process,
is_done: bool, date: str) -> None:
super().__init__(id_)
+ if process.id_ is None:
+ raise NotFoundException('Process of Todo without ID (not saved?)')
self.process = process
self._is_done = is_done
self.date = date
self.children.remove(child)
child.parents.remove(self)
- def save(self, db_conn: DatabaseConnection) -> None:
- """Write self and children to DB and its cache."""
- if self.process.id_ is None:
- raise NotFoundException('Process of Todo without ID (not saved?)')
- self.save_core(db_conn)
- assert isinstance(self.id_, int)
- db_conn.rewrite_relations('todo_children', 'child', self.id_,
- [[p.id_] for p in self.parents])
- db_conn.rewrite_relations('todo_children', 'parent', self.id_,
- [[c.id_] for c in self.children])
- db_conn.rewrite_relations('todo_conditions', 'todo', self.id_,
- [[c.id_] for c in self.conditions])
- db_conn.rewrite_relations('todo_enables', 'todo', self.id_,
- [[c.id_] for c in self.enables])
- db_conn.rewrite_relations('todo_disables', 'todo', self.id_,
- [[c.id_] for c in self.disables])
-
def remove(self, db_conn: DatabaseConnection) -> None:
"""Remove from DB, including relations."""
- assert isinstance(self.id_, int)
children_to_remove = self.children[:]
parents_to_remove = self.parents[:]
for child in children_to_remove:
self.remove_child(child)
for parent in parents_to_remove:
parent.remove_child(self)
- db_conn.delete_where('todo_children', 'parent', self.id_)
- db_conn.delete_where('todo_children', 'child', self.id_)
- db_conn.delete_where('todo_conditions', 'todo', self.id_)
- db_conn.delete_where('todo_enables', 'todo', self.id_)
- db_conn.delete_where('todo_disables', 'todo', self.id_)
super().remove(db_conn)
db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_,
[[item[0], item[1]]
for item in self.history.items()])
+
+ def remove(self, db_conn: DatabaseConnection) -> None:
+ """Remove from DB."""
+ db_conn.delete_where(self.table_name, 'parent', self.parent.id_)
#!/bin/sh
set -e
-# for dir in $(echo '.' 'plomtask' 'tests'); do
-for dir in $(echo 'tests'); do
+for dir in $(echo '.' 'plomtask' 'tests'); do
echo "Running mypy on ${dir}/ …."
python3 -m mypy --strict ${dir}/*.py
echo "Running flake8 on ${dir}/ …"
self.check_remove()
c = Condition(None)
proc = Process(None)
+ proc.save(self.db_conn)
todo = Todo(None, proc, False, '2024-01-01')
for depender in (proc, todo):
assert hasattr(depender, 'save')
def test_Todo_by_id(self) -> None:
"""Test creation and findability of Todos."""
process_unsaved = Process(None)
- todo = Todo(None, process_unsaved, False, self.date1)
with self.assertRaises(NotFoundException):
- todo.save(self.db_conn)
+ todo = Todo(None, process_unsaved, False, self.date1)
process_unsaved.save(self.db_conn)
+ todo = Todo(None, process_unsaved, False, self.date1)
todo.save(self.db_conn)
self.assertEqual(Todo.by_id(self.db_conn, 1), todo)
with self.assertRaises(NotFoundException):
projects / plomtask / commitdiff