Add NotFoundError and Model.get_fresh_from_db; tidy DB getters

Introduce NotFoundError and a Model.get_fresh_from_db helper that reloads
an object from the database and raises when missing. Use it to simplify
Model.load and UI change detection.
This commit is contained in:
Šarūnas Nejus 2025-12-26 17:21:30 +00:00
parent 8ccb33e4bc
commit e1e0d945f8
No known key found for this signature in database
4 changed files with 31 additions and 27 deletions

View file

@ -38,7 +38,10 @@ from functools import cached_property
from sqlite3 import Connection, sqlite_version_info
from typing import TYPE_CHECKING, Any, AnyStr, Generic
from typing_extensions import TypeVar # default value support
from typing_extensions import (
Self,
TypeVar, # default value support
)
from unidecode import unidecode
import beets
@ -84,6 +87,10 @@ class DBCustomFunctionError(Exception):
)
class NotFoundError(LookupError):
pass
class FormattedMapping(Mapping[str, str]):
"""A `dict`-like formatted view of a model.
@ -369,6 +376,14 @@ class Model(ABC, Generic[D]):
"""
return self._check_db()
def get_fresh_from_db(self) -> Self:
"""Load this object from the database."""
model_cls = self.__class__
if obj := self.db._get(model_cls, self.id):
return obj
raise NotFoundError(f"No matching {model_cls.__name__} found") from None
@classmethod
def _getters(cls: type[Model]):
"""Return a mapping from field names to getter functions."""
@ -656,11 +671,8 @@ class Model(ABC, Generic[D]):
if not self._dirty and self.db.revision == self._revision:
# Exit early
return
stored_obj = self.db._get(type(self), self.id)
assert stored_obj is not None, f"object {self.id} not in DB"
self._values_fixed = LazyConvertDict(self)
self._values_flex = LazyConvertDict(self)
self.update(dict(stored_obj))
self.__dict__.update(self.get_fresh_from_db().__dict__)
self.clear_dirty()
def remove(self):
@ -1309,12 +1321,6 @@ class Database:
sort if sort.is_slow() else None, # Slow sort component.
)
def _get(
self,
model_cls: type[AnyModel],
id,
) -> AnyModel | None:
"""Get a Model object by its id or None if the id does not
exist.
"""
return self._fetch(model_cls, MatchQuery("id", id)).get()
def _get(self, model_cls: type[AnyModel], id_: int) -> AnyModel | None:
"""Get a Model object by its id or None if the id does not exist."""
return self._fetch(model_cls, MatchQuery("id", id_)).get()

View file

@ -125,24 +125,20 @@ class Library(dbcore.Database):
return self._fetch(Item, query, sort or self.get_default_item_sort())
# Convenience accessors.
def get_item(self, id):
def get_item(self, id_: int) -> Item | None:
"""Fetch a :class:`Item` by its ID.
Return `None` if no match is found.
"""
return self._get(Item, id)
return self._get(Item, id_)
def get_album(self, item_or_id):
def get_album(self, item_or_id: Item | int) -> Album | None:
"""Given an album ID or an item associated with an album, return
a :class:`Album` object for the album.
If no such album exists, return `None`.
"""
if isinstance(item_or_id, int):
album_id = item_or_id
else:
album_id = item_or_id.album_id
if album_id is None:
return None
return self._get(Album, album_id)
album_id = (
item_or_id if isinstance(item_or_id, int) else item_or_id.album_id
)
return self._get(Album, album_id) if album_id else None

View file

@ -620,6 +620,8 @@ class Album(LibModel):
class Item(LibModel):
"""Represent a song or track."""
album_id: int | None
_table = "items"
_flex_table = "item_attributes"
_fields = {

View file

@ -1073,7 +1073,7 @@ def show_model_changes(
restrict the detection to. `always` indicates whether the object is
always identified, regardless of whether any changes are present.
"""
old = old or new._db._get(type(new), new.id)
old = old or new.get_fresh_from_db()
# Keep the formatted views around instead of re-creating them in each
# iteration step