typing: fix beets.dbcore.db

This commit is contained in:
Šarūnas Nejus
2026-07-07 20:40:43 +01:00
parent 8d03803fcf
commit 7a553aad7e
3 changed files with 27 additions and 20 deletions

View File

@@ -242,13 +242,15 @@ class Model(ABC, Generic[D]):
flags are used to track which fields need to be stored.
"""
id: int | None
# Abstract components (to be provided by subclasses).
_table: str
_table: ClassVar[str]
"""The main SQLite table name.
"""
_flex_table: str
_flex_table: ClassVar[str]
"""The flex field SQLite table name.
"""
@@ -257,12 +259,12 @@ class Model(ABC, Generic[D]):
keys are field names and the values are `Type` objects.
"""
_search_fields: Sequence[str] = ()
_search_fields: ClassVar[Sequence[str]] = ()
"""The fields that should be queried by default by unqualified query
terms.
"""
_indices: Sequence[Index] = ()
_indices: ClassVar[Sequence[Index]] = ()
"""A sequence of `Index` objects that describe the indices to be
created for this table.
"""
@@ -298,7 +300,7 @@ class Model(ABC, Generic[D]):
@cached_classproperty
def _relation(cls) -> type[Model]:
"""The model that this model is closely related to."""
return cls
return cls # type: ignore[return-value]
@cached_classproperty
def relation_join(cls) -> str:
@@ -332,7 +334,8 @@ class Model(ABC, Generic[D]):
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):
if self.id is not None and (obj := self.db._get(model_cls, self.id)):
return obj
raise NotFoundError(f"No matching {model_cls.__name__} found") from None
@@ -400,7 +403,7 @@ class Model(ABC, Generic[D]):
return self._db
def copy(self) -> Model:
def copy(self) -> Self:
"""Create a copy of the model object.
The field values and other state is duplicated, but the new copy
@@ -1032,13 +1035,13 @@ class Migration(ABC):
self.db._connection().row_factory = original_factory
def migrate_model(
self, model_cls: type[Model], *args: Any, **kwargs: Any
self, model_cls: type[Model], current_fields: set[str]
) -> None:
"""Run this migration once for a model's backing table."""
table = model_cls._table
if not self.db.migration_exists(self.name, table):
self._before_migration_backup(table)
self._migrate_data(model_cls, *args, **kwargs)
self._migrate_data(model_cls, current_fields)
self.db.record_migration(self.name, table)
def _before_migration_backup(self, table: str) -> None:
@@ -1194,8 +1197,8 @@ class Database:
0,
):
# If possible, disable double-quoted strings
conn.setconfig(sqlite3.SQLITE_DBCONFIG_DQS_DDL, 0)
conn.setconfig(sqlite3.SQLITE_DBCONFIG_DQS_DML, 0)
conn.setconfig(sqlite3.SQLITE_DBCONFIG_DQS_DDL, False)
conn.setconfig(sqlite3.SQLITE_DBCONFIG_DQS_DML, False)
self.add_functions(conn)

View File

@@ -352,10 +352,11 @@ class Album(LibModel):
def _getters(cls) -> dict[str, Callable[[Self], object]]:
# In addition to plugin-provided computed fields, also expose
# the album's directory as `path`.
getters = plugins.album_field_getters()
getters["path"] = Album.item_dir
getters["albumtotal"] = Album._albumtotal
return getters
return {
**plugins.album_field_getters(),
"path": Album.item_dir,
"albumtotal": Album._albumtotal,
}
def items(self):
"""Return an iterable over the items associated with this
@@ -766,11 +767,12 @@ class Item(LibModel):
@classmethod
def _getters(cls) -> dict[str, Callable[[Self], object]]:
getters = plugins.item_field_getters()
getters["singleton"] = lambda i: i.album_id is None
getters["filesize"] = Item.try_filesize # In bytes.
getters["has_cover_art"] = Item.has_cover_art
return getters
return {
**plugins.item_field_getters(),
"singleton": lambda i: i.album_id is None,
"filesize": Item.try_filesize, # In bytes.
"has_cover_art": Item.has_cover_art,
}
def duplicates_query(self, fields: list[str]) -> dbcore.AndQuery:
"""Return a query for entities with same values in the given fields."""

View File

@@ -43,6 +43,8 @@ def libtree(lib: Library) -> Node:
"""
root = Node({}, {})
for item in lib.items():
if item.id is None:
continue
dest = item.destination(relative_to_libdir=True)
parts = util.components(util.as_string(dest))
_insert(root, parts, item.id)