From: Christian Heller Date: Thu, 25 Apr 2024 05:12:41 +0000 (+0200) Subject: Re-order BaseModel methods for more meaningful grouping. X-Git-Url: https://plomlompom.com/repos/%7B%7Bdb.prefix%7D%7D/static/%7B%7B%20web_path%20%7D%7D/%7B%7Bprefix%7D%7D/foo.html?a=commitdiff_plain;h=4815fe5c7be508e67ceec144968a81bdd6a923d4;p=plomtask Re-order BaseModel methods for more meaningful grouping. --- diff --git a/plomtask/db.py b/plomtask/db.py index 1753da4..fe67e5c 100644 --- a/plomtask/db.py +++ b/plomtask/db.py @@ -123,6 +123,45 @@ class BaseModel(Generic[BaseModelId]): id_: None | BaseModelId cache_: dict[BaseModelId, Self] + @classmethod + def get_cached(cls: type[BaseModelInstance], + id_: BaseModelId) -> BaseModelInstance | None: + """Get object of id_ from class's cache, or None if not found.""" + # pylint: disable=consider-iterating-dictionary + cache = cls.get_cache() + if id_ in cache.keys(): + obj = cache[id_] + assert isinstance(obj, cls) + return obj + return None + + @classmethod + def empty_cache(cls) -> None: + """Empty class's cache.""" + cls.cache_ = {} + + @classmethod + def get_cache(cls: type[BaseModelInstance]) -> dict[Any, BaseModel[Any]]: + """Get cache dictionary, create it if not yet existing.""" + if not hasattr(cls, 'cache_'): + d: dict[Any, BaseModel[Any]] = {} + cls.cache_ = d + return cls.cache_ + + def cache(self) -> None: + """Update object in class's cache.""" + if self.id_ is None: + raise HandledException('Cannot cache object without ID.') + cache = self.__class__.get_cache() + cache[self.id_] = self + + def uncache(self) -> None: + """Remove self from cache.""" + if self.id_ is None: + raise HandledException('Cannot un-cache object without ID.') + cache = self.__class__.get_cache() + del cache[self.id_] + @classmethod def from_table_row(cls: type[BaseModelInstance], # pylint: disable=unused-argument @@ -149,6 +188,22 @@ class BaseModel(Generic[BaseModelId]): break return obj, from_cache + @classmethod + def all(cls: type[BaseModelInstance], + db_conn: DatabaseConnection) -> list[BaseModelInstance]: + """Collect all objects of class.""" + items: dict[BaseModelId, BaseModelInstance] = {} + for k, v in cls.get_cache().items(): + assert isinstance(v, cls) + items[k] = v + already_recorded = items.keys() + for id_ in db_conn.column_all(cls.table_name, 'id'): + if id_ not in already_recorded: + # pylint: disable=no-member + item = cls.by_id(db_conn, id_) # type: ignore[attr-defined] + items[item.id_] = item + return list(items.values()) + def set_int_id(self, id_: int | None) -> None: """Set id_ if >= 1 or None, else fail.""" if (id_ is not None) and id_ < 1: @@ -168,58 +223,3 @@ class BaseModel(Generic[BaseModelId]): if update_with_lastrowid: self.id_ = cursor.lastrowid # type: ignore[assignment] self.cache() - - @classmethod - def get_cached(cls: type[BaseModelInstance], - id_: BaseModelId) -> BaseModelInstance | None: - """Get object of id_ from class's cache, or None if not found.""" - # pylint: disable=consider-iterating-dictionary - cache = cls.get_cache() - if id_ in cache.keys(): - obj = cache[id_] - assert isinstance(obj, cls) - return obj - return None - - def cache(self) -> None: - """Update object in class's cache.""" - if self.id_ is None: - raise HandledException('Cannot cache object without ID.') - cache = self.__class__.get_cache() - cache[self.id_] = self - - def uncache(self) -> None: - """Remove self from cache.""" - if self.id_ is None: - raise HandledException('Cannot un-cache object without ID.') - cache = self.__class__.get_cache() - del cache[self.id_] - - @classmethod - def empty_cache(cls) -> None: - """Empty class's cache.""" - cls.cache_ = {} - - @classmethod - def all(cls: type[BaseModelInstance], - db_conn: DatabaseConnection) -> list[BaseModelInstance]: - """Collect all objects of class.""" - items: dict[BaseModelId, BaseModelInstance] = {} - for k, v in cls.get_cache().items(): - assert isinstance(v, cls) - items[k] = v - already_recorded = items.keys() - for id_ in db_conn.column_all(cls.table_name, 'id'): - if id_ not in already_recorded: - # pylint: disable=no-member - item = cls.by_id(db_conn, id_) # type: ignore[attr-defined] - items[item.id_] = item - return list(items.values()) - - @classmethod - def get_cache(cls: type[BaseModelInstance]) -> dict[Any, BaseModel[Any]]: - """Get cache dictionary, create it if not yet existing.""" - if not hasattr(cls, 'cache_'): - d: dict[Any, BaseModel[Any]] = {} - cls.cache_ = d - return cls.cache_