diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs index 0e75c53da..d88ac9154 100644 --- a/.git-blame-ignore-revs +++ b/.git-blame-ignore-revs @@ -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 diff --git a/beets/dbcore/db.py b/beets/dbcore/db.py index 572435e44..e66f4548b 100755 --- a/beets/dbcore/db.py +++ b/beets/dbcore/db.py @@ -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): diff --git a/beets/dbcore/query.py b/beets/dbcore/query.py index acb6ce301..3653434c1 100644 --- a/beets/dbcore/query.py +++ b/beets/dbcore/query.py @@ -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) diff --git a/beets/library/exceptions.py b/beets/library/exceptions.py index 39725a4e4..df095aeb8 100644 --- a/beets/library/exceptions.py +++ b/beets/library/exceptions.py @@ -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__()}" diff --git a/beets/library/library.py b/beets/library/library.py index 4dbf95923..a3613116e 100644 --- a/beets/library/library.py +++ b/beets/library/library.py @@ -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()) diff --git a/beets/library/models.py b/beets/library/models.py index 184544524..7d8172014 100644 --- a/beets/library/models.py +++ b/beets/library/models.py @@ -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). diff --git a/beets/library/queries.py b/beets/library/queries.py index 419ff5379..79484512a 100644 --- a/beets/library/queries.py +++ b/beets/library/queries.py @@ -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::' """ # 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. diff --git a/beets/plugins.py b/beets/plugins.py index 9e57120df..1e0f3d47a 100644 --- a/beets/plugins.py +++ b/beets/plugins.py @@ -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]: diff --git a/beets/test/helper.py b/beets/test/helper.py index acb6ad90e..68e3edf54 100644 --- a/beets/test/helper.py +++ b/beets/test/helper.py @@ -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)) diff --git a/beets/ui/__init__.py b/beets/ui/__init__.py index 85e22f873..e04c7dc0c 100644 --- a/beets/ui/__init__.py +++ b/beets/ui/__init__.py @@ -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()) diff --git a/beetsplug/convert.py b/beetsplug/convert.py index 0738ba1e1..92e4f2f8f 100644 --- a/beetsplug/convert.py +++ b/beetsplug/convert.py @@ -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 diff --git a/beetsplug/fetchart.py b/beetsplug/fetchart.py index 24634e8b8..d9c44c96b 100644 --- a/beetsplug/fetchart.py +++ b/beetsplug/fetchart.py @@ -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} - " diff --git a/beetsplug/ipfs.py b/beetsplug/ipfs.py index 46cc3a7ad..f7b122a45 100644 --- a/beetsplug/ipfs.py +++ b/beetsplug/ipfs.py @@ -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 diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py index b1f7ccfa6..e2b74cd87 100644 --- a/beetsplug/replaygain.py +++ b/beetsplug/replaygain.py @@ -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) diff --git a/test/dbcore/test_db.py b/test/dbcore/test_db.py index 79be0c668..829919847 100644 --- a/test/dbcore/test_db.py +++ b/test/dbcore/test_db.py @@ -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: diff --git a/test/plugins/test_ipfs.py b/test/plugins/test_ipfs.py index fa09fea55..5032ad08f 100644 --- a/test/plugins/test_ipfs.py +++ b/test/plugins/test_ipfs.py @@ -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" diff --git a/test/ui/test_ui_init.py b/test/ui/test_ui_init.py index 279b91c0a..4ff80d7d9 100644 --- a/test/ui/test_ui_init.py +++ b/test/ui/test_ui_init.py @@ -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())