mirrors/beets
2
0
Fork 0
mirror of https://github.com/beetbox/beets.git synced 2026-07-16 14:07:40 -04:00
Code Issues Packages Projects Releases Wiki Activity

typing: type beets.library (#6825)

Browse Source 
- This PR finishes a broad typing pass across `beets.library`, mainly in
`beets/library/models.py` and `beets/library/library.py`, and tightens a
few related plugin and dbcore interfaces.

- Architecturally, the change makes the `Library`, `Item`, `Album`,
query parsing, and template helper layers agree on clearer contracts:
nullable values are explicit, collection and query return types are
defined, and cross-layer APIs now use more precise types.

- The main behavioral surface is still the same, but several weak spots
are now modeled directly in the code, such as optional albums on items,
optional art paths, library-backed template memoization, and query
objects flowing through `Library._fetch()`.

- The PR also cleans up how template uniqueness helpers work by giving
`_memotable` and related keys a stable shape, and by tightening the
logic around album-aware formatting and disambiguation.

- A small architectural rename in `dbcore` changes `Database._fetch()`
to `Database.get_results()`, which makes the lower-level database API
clearer and avoids confusion with `Library._fetch()`.

- High-level impact: better static type safety, clearer boundaries
between database, library, model, and templating code, and less
ambiguity for future refactors without intentionally changing
user-facing behavior.
This commit is contained in:
Šarūnas Nejus
2026-07-16 13:20:22 +01:00
committed by GitHub
parent 2cf20aa60c c1c931490f
commit 311662bd7d
17 changed files with 368 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
.git-blame-ignore-revs
View File

@@ -239,3 +239,17 @@ b1701c2e12b2de8e5462affe46ecc6816a5fc508
ea3d6af6169d33be009b86bf5607baae7cf9652a
# Fix #6177, remove derived types, refactor coalesce tracks
9efe87101ce03706660129633e03147a222765cf
# typing: type beets.library.exceptions
fcee48fb0a11003afab1f4eb7334dca1e4842606
# typing: type beets.library.queries
c6b87e869e44bd53fccc9603b682351c96612583
# dbcore: rename Database._fetch to Database.get_results
856087258b3466f09d557a054e4b7cc4cdc25638
# typing: type beets.library.library
f167dc29e41439e342c04c073758102b2a2e1af8
# typing: add return types to beets.library.models
8949562ef8f4e83ef281674c6125b041991d3d26
# typing: type models in beets.library.models
9b375d0b2d2dc9388892f1b901aae67ab0ad8290
# typing: type the rest of beets.library.models
9022cb5ed7fddf60c1156dc3ca7e3a540dc8f7f2

12
beets/dbcore/db.py
View File

@@ -15,6 +15,7 @@ from collections.abc import Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from functools import cached_property
from pathlib import Path
from sqlite3 import Connection, sqlite_version_info
from typing import (
TYPE_CHECKING,
@@ -52,8 +53,7 @@ if TYPE_CHECKING:
from sqlite3 import Connection
from types import TracebackType
from beets.util import PathLike
from ..util import PathLike
from .query import FieldQueryType, Query, SQLiteType
from .sort import FieldSort, Sort
@@ -1085,6 +1085,8 @@ class Database:
data is written in a transaction.
"""
path: Path
def __init__(self, path: PathLike, timeout: float = 5.0) -> None:
if sqlite3.threadsafety == 0:
raise RuntimeError(
@@ -1098,7 +1100,7 @@ class Database:
if hasattr(sqlite3, "enable_callback_tracebacks"):
sqlite3.enable_callback_tracebacks(True)
self.path = path
self.path = Path(os.fsdecode(path))
self.timeout = timeout
self._connections: dict[int, sqlite3.Connection] = {}
@@ -1382,7 +1384,7 @@ class Database:
# Querying.
def _fetch(
def _get_results(
self,
model_cls: type[AnyModel],
query: Query | None = None,
@@ -1442,7 +1444,7 @@ class Database:
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()
return self._get_results(model_cls, MatchQuery("id", id_)).get()
class Index(NamedTuple):

2
beets/dbcore/query.py
View File

@@ -48,7 +48,7 @@ class InvalidQueryError(ParsingError):
"""
def __init__(
self, query: str | Sequence[str] | Query, explanation: Exception
self, query: str | Sequence[str] | Query | None, explanation: Exception
) -> None:
if isinstance(query, list):
query = " ".join(query)

8
beets/library/exceptions.py
View File

@@ -8,7 +8,7 @@ class FileOperationError(Exception):
error, and an unhandled Mutagen exception.
"""
def __init__(self, path, reason):
def __init__(self, path: bytes, reason: Exception) -> None:
"""Create an exception describing an operation on the file at
`path` with the underlying (chained) exception `reason`.
"""
@@ -16,7 +16,7 @@ class FileOperationError(Exception):
self.path = path
self.reason = reason
def __str__(self):
def __str__(self) -> str:
"""Get a string representing the error.
Describe both the underlying reason and the file path in question.
@@ -27,7 +27,7 @@ class FileOperationError(Exception):
class ReadError(FileOperationError):
"""An error while reading a file (i.e. in `Item.read`)."""
def __str__(self):
def __str__(self) -> str:
# Formatting super() directly stringifies the proxy object, so call
# __str__ explicitly to forward to FileOperationError.__str__.
return f"error reading {super().__str__()}"
@@ -36,7 +36,7 @@ class ReadError(FileOperationError):
class WriteError(FileOperationError):
"""An error while writing a file (i.e. in `Item.write`)."""
def __str__(self):
def __str__(self) -> str:
# Formatting super() directly stringifies the proxy object, so call
# __str__ explicitly to forward to FileOperationError.__str__.
return f"error writing {super().__str__()}"

69
beets/library/library.py
View File

@@ -3,12 +3,14 @@ from __future__ import annotations
import re
from contextlib import contextmanager
from functools import cached_property
from typing import TYPE_CHECKING
from pathlib import Path
from typing import TYPE_CHECKING, TypeVar
import platformdirs
import beets
from beets import config, context, dbcore
from beets.dbcore.query import Query
from beets.dbcore.sort import NullSort
from beets.exceptions import UserError
from beets.util import normpath
@@ -19,10 +21,16 @@ from .models import Album, Item
from .queries import parse_query_parts, parse_query_string
if TYPE_CHECKING:
from beets.dbcore import Results
from beets.util import Replacements
from collections.abc import Iterator, Sequence
from beets.dbcore.sort import Sort
from beets.util import PathLike, Replacements
from beets.util.pathformats import PathFormat
from .models import LibModel
LM = TypeVar("LM", bound=LibModel)
class Library(dbcore.Database):
"""A database of music containing songs and albums."""
@@ -39,6 +47,9 @@ class Library(dbcore.Database):
(migrations.RemoveInheritedArtpathMigration, (Item,)),
(migrations.InstrumentalLyricsInFlexFieldMigration, (Item,)),
)
# Used for template substitution performance.
_memotable: dict[tuple[str | None, str | None, str | None, int | None], str]
replacements: Replacements
@cached_property
@@ -61,10 +72,10 @@ class Library(dbcore.Database):
def __init__(
self,
path="library.blb",
path: PathLike = Path("library.blb"),
directory: str | None = None,
set_music_dir: bool = True,
):
) -> None:
self.directory = normpath(directory or platformdirs.user_music_path())
if set_music_dir:
context.set_music_dir(self.directory)
@@ -72,19 +83,17 @@ class Library(dbcore.Database):
super().__init__(path, timeout=beets.config["timeout"].as_number())
self.replacements = self.get_replacements()
# Used for template substitution performance.
self._memotable: dict[tuple[str, ...], str] = {}
self._memotable = {}
@contextmanager
def music_dir_context(self):
def music_dir_context(self) -> Iterator[Library]:
"""Temporarily bind this library's directory to path conversion."""
with context.music_dir(self.directory):
yield self
# Adding objects to the database.
def add(self, obj):
def add(self, obj: LibModel) -> int | None:
"""Add the :class:`Item` or :class:`Album` object to the library
database.
@@ -94,7 +103,7 @@ class Library(dbcore.Database):
self._memotable = {}
return obj.id
def add_album(self, items):
def add_album(self, items: list[Item]) -> Album:
"""Create a new album consisting of a list of items.
The items are added to the database if they don't yet have an
@@ -123,22 +132,34 @@ class Library(dbcore.Database):
# Querying.
def _fetch(self, model_cls, query, sort=None):
def _fetch(
self,
model_cls: type[LM],
query: str | Sequence[str] | Query | None = None,
sort: Sort | None = None,
) -> dbcore.Results[LM]:
"""Parse a query and fetch.
If an order specification is present in the query string
the `sort` argument is ignored.
"""
# Parse the query, if necessary.
parsed_sort = None
parsed_query = None
try:
parsed_sort = None
# Query parsing needs the library root, but keeping it scoped here
# avoids leaking one Library's directory into another's work.
with context.music_dir(self.directory):
if isinstance(query, Query):
parsed_query = query
if isinstance(query, str):
query, parsed_sort = parse_query_string(query, model_cls)
parsed_query, parsed_sort = parse_query_string(
query, model_cls
)
elif isinstance(query, (list, tuple)):
query, parsed_sort = parse_query_parts(query, model_cls)
parsed_query, parsed_sort = parse_query_parts(
query, model_cls
)
except dbcore.query.InvalidQueryArgumentValueError as exc:
raise dbcore.InvalidQueryError(query, exc)
@@ -147,27 +168,35 @@ class Library(dbcore.Database):
if parsed_sort and not isinstance(parsed_sort, NullSort):
sort = parsed_sort
return super()._fetch(model_cls, query, sort)
return super()._get_results(model_cls, parsed_query, sort)
@staticmethod
def get_default_album_sort():
def get_default_album_sort() -> Sort:
"""Get a :class:`Sort` object for albums from the config option."""
return dbcore.sort_from_strings(
Album, beets.config["sort_album"].as_str_seq()
)
@staticmethod
def get_default_item_sort():
def get_default_item_sort() -> Sort:
"""Get a :class:`Sort` object for items from the config option."""
return dbcore.sort_from_strings(
Item, beets.config["sort_item"].as_str_seq()
)
def albums(self, query=None, sort=None) -> Results[Album]:
def albums(
self,
query: str | Sequence[str] | Query | None = None,
sort: Sort | None = None,
) -> dbcore.Results[Album]:
"""Get :class:`Album` objects matching the query."""
return self._fetch(Album, query, sort or self.get_default_album_sort())
def items(self, query=None, sort=None) -> Results[Item]:
def items(
self,
query: str | Sequence[str] | Query | None = None,
sort: Sort | None = None,
) -> dbcore.Results[Item]:
"""Get :class:`Item` objects matching the query."""
return self._fetch(Item, query, sort or self.get_default_item_sort())

317
beets/library/models.py
View File

@@ -6,7 +6,7 @@ import time
from contextlib import suppress
from functools import cached_property
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar, TypeVar
from typing import TYPE_CHECKING, Any, ClassVar, TypeVar
from mediafile import MediaFile, UnreadableFileError
from typing_extensions import Self
@@ -34,12 +34,14 @@ from .fields import TYPE_BY_FIELD
from .queries import parse_query_string
if TYPE_CHECKING:
from collections.abc import Callable, KeysView
from collections.abc import Callable, Iterable, Iterator, KeysView, Mapping
from beets.dbcore import Results
from beets.dbcore.query import FieldQuery, FieldQueryType
from beets.dbcore.sort import FieldSort
from beets.util.pathformats import PathFormat
from .library import Library # noqa: F401
from .library import Library
log = logging.getLogger("beets")
@@ -81,34 +83,34 @@ class LibModel(dbcore.Model["Library"]):
"""The path to the entity as pathlib.Path."""
return Path(os.fsdecode(self.path))
def _template_funcs(self):
def _template_funcs(self) -> Mapping[str, Callable[[str], str]]:
funcs = DefaultTemplateFunctions(self, self._db).functions()
funcs.update(plugins.template_funcs())
return funcs
def store(self, fields=None):
def store(self, fields: Iterable[str] | None = None) -> None:
super().store(fields)
plugins.send("database_change", lib=self._db, model=self)
plugins.send("database_change", lib=self.db, model=self)
def remove(self):
def remove(self) -> None:
super().remove()
plugins.send("database_change", lib=self._db, model=self)
plugins.send("database_change", lib=self.db, model=self)
def add(self, lib=None):
def add(self, lib: Library | None = None) -> None:
# super().add() calls self.store(), which sends `database_change`,
# so don't do it here
super().add(lib)
def __format__(self, spec):
def __format__(self, spec: str) -> str:
if not spec:
spec = beets.config[self._format_config_key].as_str()
assert isinstance(spec, str)
return self.evaluate_template(spec)
def __str__(self):
def __str__(self) -> str:
return format(self)
def __bytes__(self):
def __bytes__(self) -> bytes:
return self.__str__().encode("utf-8")
# Convenient queries.
@@ -141,16 +143,20 @@ class LibModel(dbcore.Model["Library"]):
return query_cls(field, pattern, fast)
@classmethod
def any_field_query(cls, *args, **kwargs) -> dbcore.OrQuery:
def any_field_query(
cls, pattern: str, query_cls: FieldQueryType
) -> dbcore.OrQuery:
return dbcore.OrQuery(
[cls.field_query(f, *args, **kwargs) for f in cls._search_fields]
[cls.field_query(f, pattern, query_cls) for f in cls._search_fields]
)
@classmethod
def any_writable_media_field_query(cls, *args, **kwargs) -> dbcore.OrQuery:
def any_writable_media_field_query(
cls, pattern: str, query_cls: FieldQueryType
) -> dbcore.OrQuery:
fields = cls.writable_media_fields
return dbcore.OrQuery(
[cls.field_query(f, *args, **kwargs) for f in fields]
[cls.field_query(f, pattern, query_cls) for f in fields]
)
def duplicates_query(self, fields: list[str]) -> dbcore.AndQuery:
@@ -190,40 +196,40 @@ class FormattedItemMapping(dbcore.db.FormattedMapping):
self.item = item
@cached_property
def all_keys(self):
def all_keys(self) -> set[str]:
return set(self.model_keys).union(self.album_keys)
@cached_property
def album_keys(self):
def album_keys(self) -> list[str]:
album_keys = []
if self.album:
if self.included_keys == self.ALL_KEYS:
if isinstance(self.included_keys, list):
album_keys = self.included_keys
else:
# Performance note: this triggers a database query.
for key in self.album.keys(computed=True):
if key in Album.item_keys or key not in self.item._fields:
album_keys.append(key)
else:
album_keys = self.included_keys
return album_keys
@property
def album(self):
def album(self) -> Album | None:
return self.item._cached_album
def _get(self, key):
def _get(self, key: str) -> str:
"""Get the value for a key, either from the album or the item.
Raise a KeyError for invalid keys.
"""
if self.for_path and key in self.album_keys:
if self.album and self.for_path and key in self.album_keys:
return self._get_formatted(self.album, key)
if key in self.model_keys:
return self._get_formatted(self.model, key)
if key in self.album_keys:
if self.album and key in self.album_keys:
return self._get_formatted(self.album, key)
raise KeyError(key)
def __getitem__(self, key):
def __getitem__(self, key: str) -> str:
"""Get the value for a key.
`artist` and `albumartist` are fallback values for each other
@@ -244,10 +250,10 @@ class FormattedItemMapping(dbcore.db.FormattedMapping):
return value
def __iter__(self):
def __iter__(self) -> Iterator[str]:
return iter(self.all_keys)
def __len__(self):
def __len__(self) -> int:
return len(self.all_keys)
@@ -258,7 +264,7 @@ class Album(LibModel):
Reflects the library's "albums" table, including album art.
"""
artpath: bytes
artpath: bytes | None
_table = "albums"
_flex_table = "album_attributes"
@@ -358,7 +364,7 @@ class Album(LibModel):
"albumtotal": Album._albumtotal,
}
def items(self):
def items(self) -> Results[Item]: # type: ignore[override]
"""Return an iterable over the items associated with this
album.
@@ -367,9 +373,12 @@ class Album(LibModel):
Since :meth:`Album.items` predates these methods, and is
likely to be used by plugins, we keep this interface as-is.
"""
if self._db is None:
raise AttributeError(f"{type(self).__name__} has no database")
return self._db.items(dbcore.MatchQuery("album_id", self.id))
def remove(self, delete=False, with_items=True):
def remove(self, delete: bool = False, with_items: bool = True) -> None:
"""Remove this album and all its associated items from the
library.
@@ -395,7 +404,11 @@ class Album(LibModel):
for item in self.items():
item.remove(delete, False)
def move_art(self, operation=MoveOperation.MOVE, item_dir=None):
def move_art(
self,
operation: MoveOperation = MoveOperation.MOVE,
item_dir: bytes | None = None,
) -> None:
"""Move, copy, link or hardlink (depending on `operation`) any
existing album art so that it remains in the same directory as
the items.
@@ -432,7 +445,7 @@ class Album(LibModel):
util.move(old_art, new_art)
util.prune_dirs(
os.path.dirname(old_art),
self._db.directory,
self.db.directory,
clutter=beets.config["clutter"].as_str_seq(),
)
elif operation == MoveOperation.COPY:
@@ -449,7 +462,12 @@ class Album(LibModel):
assert False, "unknown MoveOperation"
self.artpath = new_art
def move(self, operation=MoveOperation.MOVE, basedir=None, store=True):
def move(
self,
operation: MoveOperation = MoveOperation.MOVE,
basedir: bytes | None = None,
store: bool = True,
) -> None:
"""Move, copy, link or hardlink (depending on `operation`)
all items to their destination. Any album art moves along with them.
@@ -462,7 +480,7 @@ class Album(LibModel):
the album is not stored automatically, and it will have to be manually
stored after invoking this method.
"""
basedir = basedir or self._db.directory
basedir = basedir or self.db.directory
# Ensure new metadata is available to items for destination
# computation.
@@ -483,7 +501,7 @@ class Album(LibModel):
if store:
self.store()
def item_dir(self):
def item_dir(self) -> bytes:
"""Return the directory containing the album's first item,
provided that such an item exists.
"""
@@ -492,7 +510,7 @@ class Album(LibModel):
raise ValueError(f"empty album for album id {self.id}")
return os.path.dirname(item.path)
def _albumtotal(self):
def _albumtotal(self) -> int:
"""Return the total number of tracks on all discs on the album."""
if self.disctotal == 1 or not beets.config["per_disc_numbering"]:
return self.items()[0].tracktotal
@@ -512,7 +530,9 @@ class Album(LibModel):
return total
def art_destination(self, image, item_dir=None):
def art_destination(
self, image: bytes, item_dir: bytes | None = None
) -> bytes:
"""Return a path to the destination for the album art image
for the album.
@@ -530,17 +550,15 @@ class Album(LibModel):
subpath = self.evaluate_template(filename_tmpl, True)
if beets.config["asciify_paths"]:
subpath = util.asciify_path(subpath)
subpath = util.sanitize_path(
subpath, replacements=self._db.replacements
)
subpath = bytestring_path(subpath)
subpath = util.sanitize_path(subpath, replacements=self.db.replacements)
subpath_bytes = bytestring_path(subpath)
_, ext = os.path.splitext(image)
dest = os.path.join(item_dir, subpath + ext)
dest = os.path.join(item_dir, subpath_bytes + ext)
return bytestring_path(dest)
def set_art(self, path, copy=True):
def set_art(self, path: bytes, copy: bool = True) -> None:
"""Set the album's cover art to the image at the given path.
The image is copied (or moved) into place, replacing any
@@ -572,7 +590,9 @@ class Album(LibModel):
plugins.send("art_set", album=self)
def store(self, fields=None, inherit=True):
def store(
self, fields: Iterable[str] | None = None, inherit: bool = True
) -> None:
"""Update the database with the album information.
`fields` represents the fields to be stored. If not specified,
@@ -596,7 +616,7 @@ class Album(LibModel):
else: # is a flexible attribute
track_updates[key] = self[key]
with self._db.transaction():
with self.db.transaction():
super().store(fields)
if track_updates:
for item in self.items():
@@ -610,7 +630,7 @@ class Album(LibModel):
del item[key]
item.store()
def try_sync(self, write, move, inherit=True):
def try_sync(self, write: bool, move: bool, inherit: bool = True) -> None:
"""Synchronize the album and its items with the database.
Optionally, also write any new tags into the files and update
their paths.
@@ -746,7 +766,7 @@ class Item(LibModel):
)
@property
def _cached_album(self):
def _cached_album(self) -> Album | None:
"""The Album object that this item belongs to, if any, or
None if the item is a singleton or is not associated with a
library.
@@ -762,7 +782,7 @@ class Item(LibModel):
return self.__album
@_cached_album.setter
def _cached_album(self, album):
def _cached_album(self, album: Album | None) -> None:
self.__album = album
@classmethod
@@ -781,7 +801,7 @@ class Item(LibModel):
)
@classmethod
def from_path(cls, path):
def from_path(cls, path: util.PathLike) -> Self:
"""Create a new item from the media file at the specified path."""
# Initiate with values that aren't read from files.
i = cls(album_id=None)
@@ -789,7 +809,7 @@ class Item(LibModel):
i.mtime = i.current_mtime() # Initial mtime.
return i
def __setitem__(self, key, value):
def __setitem__(self, key: str, value: Any) -> None:
"""Set the item's value for a standard field or a flexattr."""
# Encode unicode paths and read buffers.
if key == "path":
@@ -805,7 +825,7 @@ class Item(LibModel):
if changed and key in MediaFile.fields():
self.mtime = 0 # Reset mtime on dirty.
def __getitem__(self, key):
def __getitem__(self, key: str) -> Any:
"""Get the value for a field, falling back to the album if
necessary.
@@ -818,7 +838,7 @@ class Item(LibModel):
return self._cached_album[key]
raise
def __repr__(self):
def __repr__(self) -> str:
# This must not use `with_album=True`, because that might access
# the database. When debugging, that is not guaranteed to succeed, and
# can even deadlock due to the database lock.
@@ -827,18 +847,22 @@ class Item(LibModel):
f"({', '.join(f'{k}={self[k]!r}' for k in self.keys(with_album=False))})"
)
def keys(self, computed=False, with_album=True) -> KeysView[str]:
def keys(
self, computed: bool = False, with_album: bool = True
) -> KeysView[str]:
"""Get a list of available field names.
`with_album` controls whether the album's fields are included.
"""
keys = super().keys(computed=computed)
keys: set[str] = set(super().keys(computed=computed))
if with_album and self._cached_album:
keys |= self._cached_album.keys(computed=computed)
return dict.fromkeys(keys).keys()
def get(self, key, default=None, with_album=True):
def get(
self, key: str, default: Any = None, with_album: bool = True
) -> Any:
"""Get the value for a given key or `default` if it does not
exist.
@@ -851,7 +875,7 @@ class Item(LibModel):
return self._cached_album.get(key, default)
return default
def update(self, values):
def update(self, values: Mapping[str, Any]) -> None:
"""Set all key/value pairs in the mapping.
If mtime is specified, it is not reset (as it might otherwise be).
@@ -860,12 +884,12 @@ class Item(LibModel):
if self.mtime == 0 and "mtime" in values:
self.mtime = values["mtime"]
def clear(self):
def clear(self) -> None:
"""Set all key/value pairs to None."""
for key in self._media_tag_fields:
setattr(self, key, None)
def get_album(self):
def get_album(self) -> Album | None:
"""Get the Album object that this item belongs to, if any, or
None if the item is a singleton or is not associated with a
library.
@@ -876,7 +900,7 @@ class Item(LibModel):
# Interaction with file metadata.
def read(self, read_path=None):
def read(self, read_path: util.PathLike | None = None) -> None:
"""Read the metadata from the associated file.
If `read_path` is specified, read metadata from that file
@@ -907,7 +931,12 @@ class Item(LibModel):
self.path = read_path
def write(self, path=None, tags=None, id3v23=None):
def write(
self,
path: bytes | None = None,
tags: Mapping[str, Any] | None = None,
id3v23: bool | None = None,
) -> None:
"""Write the item's metadata to a media file.
All fields in `_media_fields` are written to disk according to
@@ -959,20 +988,27 @@ class Item(LibModel):
self.mtime = self.current_mtime()
plugins.send("after_write", item=self, path=path)
def try_write(self, *args, **kwargs):
def try_write(
self,
path: bytes | None = None,
tags: Mapping[str, Any] | None = None,
id3v23: bool | None = None,
) -> bool:
"""Call `write()` but catch and log `FileOperationError`
exceptions.
Return `False` an exception was caught and `True` otherwise.
"""
try:
self.write(*args, **kwargs)
self.write(path=path, tags=tags, id3v23=id3v23)
return True
except FileOperationError as exc:
log.error("{}", exc)
return False
def try_sync(self, write, move, with_album=True):
def try_sync(
self, write: bool, move: bool, with_album: bool = True
) -> None:
"""Synchronize the item with the database and, possibly, update its
tags on disk and its path (by moving the file).
@@ -995,7 +1031,9 @@ class Item(LibModel):
# Files themselves.
def move_file(self, dest, operation=MoveOperation.MOVE):
def move_file(
self, dest: bytes, operation: MoveOperation = MoveOperation.MOVE
) -> None:
"""Move, copy, link or hardlink the item depending on `operation`,
updating the path value if the move succeeds.
@@ -1047,13 +1085,13 @@ class Item(LibModel):
# Either copying or moving succeeded, so update the stored path.
self.path = dest
def current_mtime(self):
def current_mtime(self) -> int:
"""Return the current mtime of the file, rounded to the nearest
integer.
"""
return int(os.path.getmtime(syspath(self.path)))
def try_filesize(self):
def try_filesize(self) -> int:
"""Get the size of the underlying file in bytes.
If the file is missing, return 0 (and log a warning).
@@ -1064,7 +1102,7 @@ class Item(LibModel):
log.warning("could not get filesize: {}", exc)
return 0
def has_cover_art(self):
def has_cover_art(self) -> bool:
"""Check if item has embedded cover art.
Return True if images embedded in file, False otherwise.
@@ -1077,7 +1115,7 @@ class Item(LibModel):
# Model methods.
def remove(self, delete=False, with_album=True):
def remove(self, delete: bool = False, with_album: bool = True) -> None:
"""Remove the item.
If `delete`, then the associated file is removed from disk.
@@ -1101,19 +1139,19 @@ class Item(LibModel):
util.remove(self.path)
util.prune_dirs(
os.path.dirname(self.path),
self._db.directory,
self.db.directory,
clutter=beets.config["clutter"].as_str_seq(),
)
self._db._memotable = {}
self.db._memotable = {}
def move(
self,
operation=MoveOperation.MOVE,
basedir=None,
with_album=True,
store=True,
):
operation: MoveOperation = MoveOperation.MOVE,
basedir: bytes | None = None,
with_album: bool = True,
store: bool = True,
) -> None:
"""Move the item to its designated location within the library
directory (provided by destination()).
@@ -1174,14 +1212,17 @@ class Item(LibModel):
if operation == MoveOperation.MOVE:
util.prune_dirs(
os.path.dirname(old_path),
self._db.directory,
self.db.directory,
clutter=beets.config["clutter"].as_str_seq(),
)
# Templating.
def destination(
self, relative_to_libdir=False, basedir=None, path_formats=None
self,
relative_to_libdir: bool = False,
basedir: bytes | None = None,
path_formats: list[PathFormat] | None = None,
) -> bytes:
"""Return the path in the library directory designated for the item
(i.e., where the file ought to be).
@@ -1194,20 +1235,18 @@ class Item(LibModel):
basedir = basedir or self.db.directory
path_formats = path_formats or self.db.path_formats
# Use a path format based on a query, falling back on the
# default.
for query, path_format in path_formats:
if query == PF_KEY_DEFAULT:
for query_str, path_format in path_formats:
if query_str == PF_KEY_DEFAULT:
continue
query, _ = parse_query_string(query, type(self))
query, _ = parse_query_string(query_str, type(self))
if query.match(self):
# The query matches the item! Use the corresponding path
# format.
break
else:
# No query matched; fall back to default.
for query, path_format in path_formats:
if query == PF_KEY_DEFAULT:
for query_str, path_format in path_formats:
if query_str == PF_KEY_DEFAULT:
break
else:
assert False, "no default path format"
@@ -1242,7 +1281,7 @@ class Item(LibModel):
return normpath(os.path.join(basedir, lib_path_bytes))
def _int_arg(s):
def _int_arg(s: str) -> int:
"""Convert a string argument to an integer for use in a template
function.
@@ -1267,7 +1306,7 @@ class DefaultTemplateFunctions:
"""Names of tmpl_* functions in this class."""
return [s for s in dir(cls) if s.startswith(cls._prefix)]
def __init__(self, item=None, lib=None):
def __init__(self, item: LibModel, lib: Library | None) -> None:
"""Parametrize the functions.
If `item` or `lib` is None, then some functions (namely, ``aunique``)
@@ -1276,7 +1315,7 @@ class DefaultTemplateFunctions:
self.item = item
self.lib = lib
def functions(self):
def functions(self) -> dict[str, Callable[..., str]]:
"""Return a dictionary containing the functions defined in this
object.
@@ -1289,64 +1328,66 @@ class DefaultTemplateFunctions:
return out
@staticmethod
def tmpl_lower(s):
def tmpl_lower(s: str) -> str:
"""Convert a string to lower case."""
return s.lower()
@staticmethod
def tmpl_upper(s):
def tmpl_upper(s: str) -> str:
"""Convert a string to upper case."""
return s.upper()
@staticmethod
def tmpl_capitalize(s):
def tmpl_capitalize(s: str) -> str:
"""Converts to a capitalized string."""
return s.capitalize()
@staticmethod
def tmpl_title(s):
def tmpl_title(s: str) -> str:
"""Convert a string to title case."""
return string.capwords(s)
@staticmethod
def tmpl_left(s, chars):
def tmpl_left(s: str, chars: str) -> str:
"""Get the leftmost characters of a string."""
return s[0 : _int_arg(chars)]
@staticmethod
def tmpl_right(s, chars):
def tmpl_right(s: str, chars: str) -> str:
"""Get the rightmost characters of a string."""
return s[-_int_arg(chars) :]
@staticmethod
def tmpl_if(condition, trueval, falseval=""):
def tmpl_if(condition: str, trueval: str, falseval: str = "") -> str:
"""If ``condition`` is nonempty and nonzero, emit ``trueval``;
otherwise, emit ``falseval`` (if provided).
"""
_condition: str | int = condition
try:
int_condition = _int_arg(condition)
_condition = _int_arg(condition)
except ValueError:
if condition.lower() == "false":
return falseval
else:
condition = int_condition
if condition:
return trueval
return falseval
return trueval if _condition else falseval
@staticmethod
def tmpl_asciify(s):
def tmpl_asciify(s: str) -> str:
"""Translate non-ASCII characters to their ASCII equivalents."""
return util.asciify_path(s)
@staticmethod
def tmpl_time(s, fmt):
def tmpl_time(s: str, fmt: str) -> str:
"""Format a time value using `strftime`."""
cur_fmt = beets.config["time_format"].as_str()
return time.strftime(fmt, time.strptime(s, cur_fmt))
def tmpl_aunique(self, keys=None, disam=None, bracket=None):
def tmpl_aunique(
self,
keys: str | None = None,
disam: str | None = None,
bracket: str | None = None,
) -> str:
"""Generate a string that is guaranteed to be unique among all
albums in the library who share the same set of keys.
@@ -1374,7 +1415,7 @@ class DefaultTemplateFunctions:
if memoval is not None:
return memoval
album = self.lib.get_album(album_id)
album: Album = self.lib.get_album(album_id) # type: ignore[assignment]
return self._tmpl_unique(
"aunique",
@@ -1383,12 +1424,16 @@ class DefaultTemplateFunctions:
bracket,
album_id,
album,
album.item_keys,
# Do nothing for singletons.
lambda a: a is None,
)
def tmpl_sunique(self, keys=None, disam=None, bracket=None):
def tmpl_sunique(
self,
keys: str | None = None,
disam: str | None = None,
bracket: str | None = None,
) -> str:
"""Generate a string that is guaranteed to be unique among all
singletons in the library who share the same set of keys.
@@ -1418,20 +1463,32 @@ class DefaultTemplateFunctions:
bracket,
item_id,
self.item,
Item.all_keys(),
# Do nothing for non singletons.
lambda i: i.album_id is not None,
)
def _tmpl_unique_memokey(self, name, keys, disam, item_id):
def _tmpl_unique_memokey(
self,
name: str | None,
keys: str | None,
disam: str | None,
item_id: int | None,
) -> tuple[str | None, str | None, str | None, int | None]:
"""Get the memokey for the unique template named "name" for the
specific parameters.
"""
return (name, keys, disam, item_id)
def _tmpl_unique(
self, name, keys, disam, bracket, item_id, db_item, item_keys, skip_item
):
self,
name: str,
keys: str | None,
disam: str | None,
bracket: str | None,
item_id: int,
db_item: LibModel,
skip_item: Callable[[LibModel], bool],
) -> str:
"""Generate a string that is guaranteed to be unique among all items of
the same type as "db_item" who share the same set of keys.
@@ -1452,21 +1509,25 @@ class DefaultTemplateFunctions:
"initial_subqueries" is a list of subqueries that should be included
in the query to find the ambiguous items.
"""
lib = self.lib
if lib is None:
return ""
memokey = self._tmpl_unique_memokey(name, keys, disam, item_id)
memoval = self.lib._memotable.get(memokey)
memoval = lib._memotable.get(memokey)
if memoval is not None:
return memoval
if skip_item(db_item):
self.lib._memotable[memokey] = ""
lib._memotable[memokey] = ""
return ""
keys = keys or beets.config[name]["keys"].as_str()
disam = disam or beets.config[name]["disambiguators"].as_str()
if bracket is None:
bracket = beets.config[name]["bracket"].as_str()
keys = keys.split()
disam = disam.split()
keys_list = keys.split()
disam_list = disam.split()
# Assign a left and right bracket or leave blank if argument is empty.
if len(bracket) == 2:
@@ -1477,21 +1538,19 @@ class DefaultTemplateFunctions:
bracket_r = ""
# Find matching items to disambiguate with.
query = db_item.duplicates_query(keys)
query = db_item.duplicates_query(keys_list)
ambigous_items = (
self.lib.items(query)
if isinstance(db_item, Item)
else self.lib.albums(query)
lib.items(query) if isinstance(db_item, Item) else lib.albums(query)
)
# If there's only one item to matching these details, then do
# nothing.
if len(ambigous_items) == 1:
self.lib._memotable[memokey] = ""
lib._memotable[memokey] = ""
return ""
# Find the first disambiguator that distinguishes the items.
for disambiguator in disam:
for disambiguator in disam_list:
# Get the value for each item for the current field.
disam_values = {s.get(disambiguator, "") for s in ambigous_items}
@@ -1503,7 +1562,7 @@ class DefaultTemplateFunctions:
else:
# No disambiguator distinguished all fields.
res = f" {bracket_l}{item_id}{bracket_r}"
self.lib._memotable[memokey] = res
lib._memotable[memokey] = res
return res
# Flatten disambiguation value into a string.
@@ -1515,11 +1574,17 @@ class DefaultTemplateFunctions:
else:
res = ""
self.lib._memotable[memokey] = res
lib._memotable[memokey] = res
return res
@staticmethod
def tmpl_first(s, count=1, skip=0, sep="; ", join_str="; "):
def tmpl_first(
s: str,
count: int = 1,
skip: int = 0,
sep: str = "; ",
join_str: str = "; ",
) -> str:
"""Get the item(s) from x to y in a string separated by something
and join then with something.
@@ -1534,7 +1599,9 @@ class DefaultTemplateFunctions:
count = skip + int(count)
return join_str.join(s.split(sep)[skip:count])
def tmpl_ifdef(self, field, trueval="", falseval=""):
def tmpl_ifdef(
self, field: str, trueval: str = "", falseval: str = ""
) -> str:
"""If field exists return trueval or the field (default)
otherwise, emit return falseval (if provided).

17
beets/library/queries.py
View File

@@ -1,17 +1,26 @@
from __future__ import annotations
import shlex
from typing import TYPE_CHECKING
import beets
from beets import dbcore, logging, plugins
if TYPE_CHECKING:
from collections.abc import Sequence
from beets.dbcore.queryparse import Prefixes
from beets.library import LibModel
log = logging.getLogger("beets")
# Query construction helpers.
def parse_query_parts(parts, model_cls):
def parse_query_parts(
parts: Sequence[str], model_cls: type[LibModel]
) -> tuple[dbcore.query.Query, dbcore.sort.Sort]:
"""Given a beets query string as a list of components, return the
`Query` and `Sort` they represent.
@@ -19,7 +28,7 @@ def parse_query_parts(parts, model_cls):
ensuring that implicit path queries are made explicit with 'path::<query>'
"""
# Get query types and their prefix characters.
prefixes = {
prefixes: Prefixes = {
":": dbcore.query.RegexpQuery,
"=~": dbcore.query.StringQuery,
"=": dbcore.query.MatchQuery,
@@ -43,7 +52,9 @@ def parse_query_parts(parts, model_cls):
return query, sort
def parse_query_string(s, model_cls):
def parse_query_string(
s: str, model_cls: type[LibModel]
) -> tuple[dbcore.query.Query, dbcore.sort.Sort]:
"""Given a beets query string, return the `Query` and `Sort` they
represent.

17
beets/plugins.py
View File

@@ -25,7 +25,6 @@ if TYPE_CHECKING:
from confuse import Subview
from beets.dbcore import Query
from beets.dbcore.db import FieldQueryType
from beets.dbcore.types import Type
from beets.importer import ImportSession, ImportTask
@@ -57,6 +56,7 @@ EventType = Literal[
"album_removed",
"albuminfo_received",
"album_matched",
"art_set",
"before_choose_candidate",
"before_item_moved",
"cli_exit",
@@ -306,7 +306,7 @@ class BeetsPlugin(metaclass=BeetsPluginMeta):
return wrapper
def queries(self) -> dict[str, type[Query]]:
def queries(self) -> dict[str, FieldQueryType]:
"""Return a dict mapping prefixes to Query subclasses."""
return {}
@@ -476,14 +476,11 @@ def commands() -> list[Subcommand]:
return out
def queries() -> dict[str, type[Query]]:
"""Returns a dict mapping prefix strings to Query subclasses all loaded
plugins.
"""
out: dict[str, type[Query]] = {}
for plugin in find_plugins():
out.update(plugin.queries())
return out
def queries() -> dict[str, FieldQueryType]:
"""Return configured query prefixes from all plugins."""
return {
p: q for plugin in find_plugins() for p, q in plugin.queries().items()
}
def types(model_cls: type[AnyModel]) -> dict[str, Type]:

4
beets/test/helper.py
View File

@@ -270,9 +270,9 @@ class TestHelper(RunMixin, PathsMixin, ConfigMixin):
self.config["directory"] = str(self.lib_path)
dbpath = (
util.bytestring_path(self.config["library"].as_filename())
self.config["library"].as_path()
if self.db_on_disk
else ":memory:"
else Path(":memory:")
)
self.lib = Library(dbpath, str(self.lib_path))

9
beets/ui/__init__.py
View File

@@ -767,9 +767,12 @@ def _setup() -> tuple[list[Subcommand], library.Library]:
def _ensure_db_directory_exists(path):
if path == b":memory:": # in memory db
dbpath = os.fspath(path)
if dbpath in (":memory:", b":memory:"): # in memory db
return
newpath = os.path.dirname(dbpath)
if not newpath:
return
newpath = os.path.dirname(path)
if not os.path.isdir(newpath):
if input_yn(
f"The database directory {util.displayable_path(newpath)} does not"
@@ -780,7 +783,7 @@ def _ensure_db_directory_exists(path):
def _open_library(config: confuse.LazyConfig) -> library.Library:
"""Create a new library instance from the configuration."""
dbpath = util.bytestring_path(config["library"].as_filename())
dbpath = config["library"].as_path()
_ensure_db_directory_exists(dbpath)
try:
lib = library.Library(dbpath, config["directory"].as_filename())

6
beetsplug/convert.py
View File

@@ -10,7 +10,7 @@ import tempfile
import threading
from functools import cached_property
from string import Template
from typing import TYPE_CHECKING, NamedTuple
from typing import TYPE_CHECKING, Literal, NamedTuple
import mediafile
from confuse import ConfigTypeError, Optional
@@ -527,7 +527,9 @@ class ConvertPlugin(BeetsPlugin):
if pretend:
return
id3v23 = self.config["id3v23"].as_choice([True, False, "inherit"])
id3v23: bool | Literal["inherit"] | None = self.config[
"id3v23"
].as_choice([True, False, "inherit"])
if id3v23 == "inherit":
id3v23 = None

6
beetsplug/fetchart.py
View File

@@ -1220,7 +1220,7 @@ class Spotify(RemoteArtSource):
paths: None | Sequence[bytes],
) -> Iterator[Candidate]:
try:
url = f"{self.SPOTIFY_ALBUM_URL}{album.items().get().spotify_album_id}"
url = f"{self.SPOTIFY_ALBUM_URL}{album.items().get().spotify_album_id}" # type: ignore[union-attr]
except AttributeError:
self._log.debug("Fetchart: no Spotify album ID found")
return
@@ -1273,7 +1273,7 @@ class CoverArtUrl(RemoteArtSource):
if album.get("cover_art_url"):
image_url = album.cover_art_url
else:
image_url = album.items().get().cover_art_url
image_url = album.items().get().cover_art_url # type: ignore[union-attr]
self._log.debug("Cover art URL {} found for {}", image_url, album)
except (AttributeError, TypeError):
self._log.debug("Cover art URL not found for {}", album)
@@ -1494,7 +1494,7 @@ class FetchArtPlugin(plugins.BeetsPlugin, RequestMixin):
failure (e.g. permission errors when writing the art file).
"""
try:
album.set_art(candidate.path, delete)
album.set_art(candidate.path, delete) # type: ignore[arg-type]
except OSError as exc:
self._log.warning(
"fetchart: could not write art for {0.albumartist} - "

10
beetsplug/ipfs.py
View File

@@ -200,8 +200,7 @@ class IPFSPlugin(BeetsPlugin):
lib_name = args[1]
else:
lib_name = _hash
lib_root = os.path.dirname(lib.path)
remote_libs = os.path.join(lib_root, b"remotes")
remote_libs = self._remote_libs_path(lib)
if not os.path.exists(remote_libs):
try:
os.makedirs(remote_libs)
@@ -255,9 +254,12 @@ class IPFSPlugin(BeetsPlugin):
rlib = self.get_remote_lib(lib)
return rlib.albums(args)
def _remote_libs_path(self, lib):
lib_root = os.path.dirname(os.fsencode(lib.path))
return os.path.join(lib_root, b"remotes")
def get_remote_lib(self, lib):
lib_root = os.path.dirname(lib.path)
remote_libs = os.path.join(lib_root, b"remotes")
remote_libs = self._remote_libs_path(lib)
path = os.path.join(remote_libs, b"joined.db")
if not os.path.isfile(path):
raise OSError

2
beetsplug/replaygain.py
View File

@@ -1415,7 +1415,7 @@ class ReplayGainPlugin(BeetsPlugin):
discs[item.disc] = []
discs[item.disc].append(item)
else:
discs[1] = album.items()
discs[1] = list(album.items())
def store_cb(task: RgTask):
task.store(write)

48
test/dbcore/test_db.py
View File

@@ -3,6 +3,7 @@
import os
import shutil
import unittest
from pathlib import Path
from tempfile import mkstemp
from typing import ClassVar
@@ -90,6 +91,28 @@ class DatabaseFixtureTwoModels(dbcore.Database):
_models = (ModelFixture2, AnotherModelFixture)
class TestDatabasePath:
@pytest.mark.parametrize(
"path", [":memory:", b":memory:", Path(":memory:")]
)
def test_path_inputs_are_stored_as_path(self, path):
db = DatabaseFixture1(path)
try:
assert db.path == Path(":memory:")
assert isinstance(db.path, Path)
finally:
db._connection().close()
def test_bytes_filesystem_path_opens_decoded_path(self, tmp_path):
db_path = tmp_path / "library.db"
db = DatabaseFixture1(os.fsencode(db_path))
try:
assert db.path == db_path
assert db_path.exists()
finally:
db._connection().close()
class ModelFixtureWithGetters(dbcore.Model):
@classmethod
def _getters(cls):
@@ -504,16 +527,16 @@ class ResultsIteratorTest(unittest.TestCase):
self.db._connection().close()
def test_iterate_once(self):
objs = self.db._fetch(ModelFixture1)
objs = self.db._get_results(ModelFixture1)
assert len(list(objs)) == 2
def test_iterate_twice(self):
objs = self.db._fetch(ModelFixture1)
objs = self.db._get_results(ModelFixture1)
list(objs)
assert len(list(objs)) == 2
def test_concurrent_iterators(self):
results = self.db._fetch(ModelFixture1)
results = self.db._get_results(ModelFixture1)
it1 = iter(results)
it2 = iter(results)
next(it1)
@@ -522,43 +545,46 @@ class ResultsIteratorTest(unittest.TestCase):
def test_slow_query(self):
q = query.SubstringQuery("foo", "ba", False)
objs = self.db._fetch(ModelFixture1, q)
objs = self.db._get_results(ModelFixture1, q)
assert len(list(objs)) == 2
def test_slow_query_negative(self):
q = query.SubstringQuery("foo", "qux", False)
objs = self.db._fetch(ModelFixture1, q)
objs = self.db._get_results(ModelFixture1, q)
assert len(list(objs)) == 0
def test_iterate_slow_sort(self):
s = sort.SlowFieldSort("foo")
res = self.db._fetch(ModelFixture1, sort=s)
res = self.db._get_results(ModelFixture1, sort=s)
objs = list(res)
assert objs[0].foo == "bar"
assert objs[1].foo == "baz"
def test_unsorted_subscript(self):
objs = self.db._fetch(ModelFixture1)
objs = self.db._get_results(ModelFixture1)
assert objs[0].foo == "baz"
assert objs[1].foo == "bar"
def test_slow_sort_subscript(self):
s = sort.SlowFieldSort("foo")
objs = self.db._fetch(ModelFixture1, sort=s)
objs = self.db._get_results(ModelFixture1, sort=s)
assert objs[0].foo == "bar"
assert objs[1].foo == "baz"
def test_length(self):
objs = self.db._fetch(ModelFixture1)
objs = self.db._get_results(ModelFixture1)
assert len(objs) == 2
def test_out_of_range(self):
objs = self.db._fetch(ModelFixture1)
objs = self.db._get_results(ModelFixture1)
with pytest.raises(IndexError):
objs[100]
def test_no_results(self):
assert self.db._fetch(ModelFixture1, query.FalseQuery()).get() is None
assert (
self.db._get_results(ModelFixture1, query.FalseQuery()).get()
is None
)
class TestException:

17
test/plugins/test_ipfs.py
View File

@@ -1,7 +1,7 @@
import os
from unittest.mock import Mock, patch
from beets import util
from beets import library, util
from beets.test import _common
from beets.test.helper import PluginTestCase
from beetsplug.ipfs import IPFSPlugin
@@ -35,6 +35,21 @@ class IPFSPluginTest(PluginTestCase):
found = True
assert found
def test_get_remote_lib_accepts_library_path(self):
self.lib.path = self.temp_dir_path / "library.db"
remote_dir = self.temp_dir_path / "remotes"
remote_dir.mkdir()
remote_lib = library.Library(remote_dir / "joined.db")
remote_lib._close()
ipfs = IPFSPlugin()
added_lib = ipfs.get_remote_lib(self.lib)
try:
assert added_lib.path == remote_dir / "joined.db"
finally:
added_lib._close()
def mk_test_album(self):
items = [_common.item() for _ in range(3)]
items[0].title = "foo bar"

5
test/ui/test_ui_init.py
View File

@@ -3,6 +3,7 @@
import os
import unittest
from copy import deepcopy
from pathlib import Path
from random import random
import pytest
@@ -74,6 +75,10 @@ class InputMethodsTest(IOMixin, unittest.TestCase):
class ParentalDirCreation(IOMixin, BeetsTestCase):
def test_memory_path_skips_creation_prompt(self):
ui._ensure_db_directory_exists(Path(":memory:"))
assert not self.io.getoutput()
def test_create_yes(self):
non_exist_path = _common.os.fsdecode(
os.path.join(self.temp_dir, b"nonexist", str(random()).encode())