id_: None | BaseModelId
cache_: dict[BaseModelId, Self]
to_search: list[str] = []
+ can_create_by_id = False
_exists = True
def __init__(self, id_: BaseModelId | None) -> None:
assert isinstance(other.id_, int)
return self.id_ < other.id_
+ @property
+ def as_dict(self) -> dict[str, object]:
+ """Return self as (json.dumps-coompatible) dict."""
+ d: dict[str, object] = {'id': self.id_}
+ if len(self.to_save_versioned) > 0:
+ d['_versioned'] = {}
+ for k in self.to_save:
+ attr = getattr(self, k)
+ if hasattr(attr, 'as_dict'):
+ d[k] = attr.as_dict
+ d[k] = attr
+ 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]
+ d[attr_name] = [x.as_dict for x in getattr(self, attr_name)]
+ return d
+
# cache management
# (we primarily use the cache to ensure we work on the same object in
# memory no matter where and how we retrieve it, e.g. we don't want
@classmethod
def empty_cache(cls) -> None:
- """Empty class's cache."""
+ """Empty class's cache, and disappear all former inhabitants."""
+ # pylint: disable=protected-access
+ # (cause we remain within the class)
+ if hasattr(cls, 'cache_'):
+ to_disappear = list(cls.cache_.values())
+ for item in to_disappear:
+ item._disappear()
cls.cache_ = {}
@classmethod
return obj
return None
- def _cache(self) -> None:
+ def cache(self) -> None:
"""Update object in class's cache.
Also calls ._disappear if cache holds older reference to object of same
# pylint: disable=unused-argument
db_conn: DatabaseConnection,
row: Row | list[Any]) -> BaseModelInstance:
- """Make from DB row, update DB cache with it."""
+ """Make from DB row (sans relations), update DB cache with it."""
obj = cls(*row)
- obj._cache()
+ assert obj.id_ is not None
+ 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_):
+ attr.history_from_row(row_)
+ obj.cache()
return obj
@classmethod
def by_id(cls, db_conn: DatabaseConnection,
- id_: BaseModelId | None,
- # pylint: disable=unused-argument
- create: bool = False) -> Self:
+ id_: BaseModelId | None
+ ) -> Self:
"""Retrieve by id_, on failure throw NotFoundException.
First try to get from cls.cache_, only then check DB; if found,
put into cache.
-
- If create=True, make anew (but do not cache yet).
"""
obj = None
if id_ is not None:
break
if obj:
return obj
- if create:
+ raise NotFoundException(f'found no object of ID {id_}')
+
+ @classmethod
+ def by_id_or_create(cls, db_conn: DatabaseConnection,
+ id_: BaseModelId | None
+ ) -> Self:
+ """Wrapper around .by_id, creating (not caching/saving) if not find."""
+ if not cls.can_create_by_id:
+ raise HandledException('Class cannot .by_id_or_create.')
+ try:
+ return cls.by_id(db_conn, id_)
+ except NotFoundException:
obj = cls(id_)
return obj
- raise NotFoundException(f'found no object of ID {id_}')
@classmethod
def all(cls: type[BaseModelInstance],
values)
if not isinstance(self.id_, str):
self.id_ = cursor.lastrowid # type: ignore[assignment]
- self._cache()
+ self.cache()
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: