X-Git-Url: https://plomlompom.com/repos/%7B%7B%20web_path%20%7D%7D/decks/%7B%7Bdeck_id%7D%7D/cards/%7B%7B%20card_id%20%7D%7D/static/git-favicon.png?a=blobdiff_plain;f=plomtask%2Fdb.py;h=13cdaef5b9c7d3e992f8c92730a9979b9eee2d73;hb=HEAD;hp=b3f1db00986b1142f5f31be34060864840ab5bdc;hpb=21df71ef1fde304b158da5989692c01f463515b5;p=plomtask diff --git a/plomtask/db.py b/plomtask/db.py index b3f1db0..67a7fc7 100644 --- a/plomtask/db.py +++ b/plomtask/db.py @@ -4,7 +4,7 @@ from os import listdir from os.path import isfile from difflib import Differ from sqlite3 import connect as sql_connect, Cursor, Row -from typing import Any, Self, TypeVar, Generic +from typing import Any, Self, TypeVar, Generic, Callable from plomtask.exceptions import HandledException, NotFoundException from plomtask.dating import valid_date @@ -232,14 +232,16 @@ BaseModelInstance = TypeVar('BaseModelInstance', bound='BaseModel[Any]') class BaseModel(Generic[BaseModelId]): """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_simples: list[str] = [] to_save_relations: list[tuple[str, str, str, int]] = [] + versioned_defaults: dict[str, str | float] = {} + add_to_dict: list[str] = [] id_: None | BaseModelId cache_: dict[BaseModelId, Self] to_search: list[str] = [] can_create_by_id = False _exists = True + sorters: dict[str, Callable[..., Any]] = {} def __init__(self, id_: BaseModelId | None) -> None: if isinstance(id_, int) and id_ < 1: @@ -251,11 +253,12 @@ class BaseModel(Generic[BaseModelId]): self.id_ = id_ def __hash__(self) -> int: - hashable = [self.id_] + [getattr(self, name) for name in self.to_save] + hashable = [self.id_] + [getattr(self, name) + for name in self.to_save_simples] for definition in self.to_save_relations: attr = getattr(self, definition[2]) hashable += [tuple(rel.id_ for rel in attr)] - for name in self.to_save_versioned: + for name in self.to_save_versioned(): hashable += [hash(getattr(self, name))] return hash(tuple(hashable)) @@ -272,59 +275,61 @@ class BaseModel(Generic[BaseModelId]): assert isinstance(other.id_, int) return self.id_ < other.id_ + @classmethod + def to_save_versioned(cls) -> list[str]: + """Return keys of cls.versioned_defaults assuming we wanna save 'em.""" + return list(cls.versioned_defaults.keys()) + @property - def as_dict(self) -> dict[str, object]: - """Return self as (json.dumps-compatible) dict.""" - library: dict[str, dict[str | int, object]] = {} - d: dict[str, object] = {'id': self.id_, '_library': library} - for to_save in self.to_save: - attr = getattr(self, to_save) - if hasattr(attr, 'as_dict_into_reference'): - d[to_save] = attr.as_dict_into_reference(library) - else: - d[to_save] = attr - if len(self.to_save_versioned) > 0: + def as_dict_and_refs(self) -> tuple[dict[str, object], + list[BaseModel[int] | BaseModel[str]]]: + """Return self as json.dumps-ready dict, list of referenced objects.""" + d: dict[str, object] = {'id': self.id_} + refs: list[BaseModel[int] | BaseModel[str]] = [] + for to_save in self.to_save_simples: + d[to_save] = getattr(self, to_save) + if len(self.to_save_versioned()) > 0: d['_versioned'] = {} - for k in self.to_save_versioned: + for k in self.to_save_versioned(): attr = getattr(self, k) assert isinstance(d['_versioned'], dict) d['_versioned'][k] = attr.history - for r in self.to_save_relations: - attr_name = r[2] - l: list[int | str] = [] - for rel in getattr(self, attr_name): - l += [rel.as_dict_into_reference(library)] - d[attr_name] = l - return d - - def as_dict_into_reference(self, - library: dict[str, dict[str | int, object]] - ) -> int | str: - """Return self.id_ while writing .as_dict into library.""" - def into_library(library: dict[str, dict[str | int, object]], - cls_name: str, - id_: str | int, - d: dict[str, object] - ) -> None: - if cls_name not in library: - library[cls_name] = {} - if id_ in library[cls_name]: - if library[cls_name][id_] != d: - msg = 'Unexpected inequality of entries for ' +\ - f'_library at: {cls_name}/{id_}' - raise HandledException(msg) - else: - library[cls_name][id_] = d - as_dict = self.as_dict - assert isinstance(as_dict['_library'], dict) - for cls_name, dict_of_objs in as_dict['_library'].items(): - for id_, obj in dict_of_objs.items(): - into_library(library, cls_name, id_, obj) - del as_dict['_library'] - assert self.id_ is not None - into_library(library, self.__class__.__name__, self.id_, as_dict) - assert isinstance(as_dict['id'], (int, str)) - return as_dict['id'] + rels_to_collect = [rel[2] for rel in self.to_save_relations] + rels_to_collect += self.add_to_dict + for attr_name in rels_to_collect: + rel_list = [] + for item in getattr(self, attr_name): + rel_list += [item.id_] + if item not in refs: + refs += [item] + d[attr_name] = rel_list + return d, refs + + @classmethod + def name_lowercase(cls) -> str: + """Convenience method to return cls' name in lowercase.""" + return cls.__name__.lower() + + @classmethod + def sort_by(cls, seq: list[Any], sort_key: str, default: str = 'title' + ) -> str: + """Sort cls list by cls.sorters[sort_key] (reverse if '-'-prefixed). + + Before cls.sorters[sort_key] is applied, seq is sorted by .id_, to + ensure predictability where parts of seq are of same sort value. + """ + reverse = False + if len(sort_key) > 1 and '-' == sort_key[0]: + sort_key = sort_key[1:] + reverse = True + if sort_key not in cls.sorters: + sort_key = default + seq.sort(key=lambda x: x.id_, reverse=reverse) + sorter: Callable[..., Any] = cls.sorters[sort_key] + seq.sort(key=sorter, reverse=reverse) + if reverse: + sort_key = f'-{sort_key}' + return sort_key # cache management # (we primarily use the cache to ensure we work on the same object in @@ -412,7 +417,7 @@ class BaseModel(Generic[BaseModelId]): """Make from DB row (sans relations), update DB cache with it.""" obj = cls(*row) assert obj.id_ is not None - for attr_name in cls.to_save_versioned: + for attr_name in cls.to_save_versioned(): attr = getattr(obj, attr_name) table_name = attr.table_name for row_ in db_conn.row_where(table_name, 'parent', obj.id_): @@ -481,7 +486,7 @@ class BaseModel(Generic[BaseModelId]): date_col: str = 'day' ) -> tuple[list[BaseModelInstance], str, str]: - """Return list of items in database within (open) date_range interval. + """Return list of items in DB within (closed) date_range interval. If no range values provided, defaults them to 'yesterday' and 'tomorrow'. Knows to properly interpret these and 'today' as value. @@ -523,7 +528,7 @@ class BaseModel(Generic[BaseModelId]): """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[_versioned|_relations]. + listed in cls.to_save_[simples|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 @@ -531,14 +536,14 @@ class BaseModel(Generic[BaseModelId]): only the case with the Day class, where it's to be a date string. """ values = tuple([self.id_] + [getattr(self, key) - for key in self.to_save]) + for key in self.to_save_simples]) table_name = self.table_name cursor = db_conn.exec_on_vals(f'REPLACE INTO {table_name} VALUES', values) if not isinstance(self.id_, str): self.id_ = cursor.lastrowid # type: ignore[assignment] self.cache() - for attr_name in self.to_save_versioned: + for attr_name in self.to_save_versioned(): getattr(self, attr_name).save(db_conn) for table, column, attr_name, key_index in self.to_save_relations: assert isinstance(self.id_, (int, str)) @@ -550,7 +555,7 @@ class BaseModel(Generic[BaseModelId]): """Remove from DB and cache, including dependencies.""" if self.id_ is None or self._get_cached(self.id_) is None: raise HandledException('cannot remove unsaved item') - for attr_name in self.to_save_versioned: + 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_)