Merge branch 'master' into feature/add-artist-to-item-entry-template

Author: matlads

.git-blame-ignore-revs

@@ -51,3 +51,5 @@ c490ac5810b70f3cf5fd8649669838e8fdb19f4d
 9147577b2b19f43ca827e9650261a86fb0450cef
 # Copy paste query, types from library to dbcore
 1a045c91668c771686f4c871c84f1680af2e944b
+# Library restructure (split library.py into multiple modules)
+0ad4e19d4f870db757373f44d12ff3be2441363a

.github/workflows/ci.yaml

@@ -52,7 +52,7 @@ jobs:
       - if: ${{ env.IS_MAIN_PYTHON != 'true' }}
         name: Test without coverage
         run: |
-          poetry install --extras=autobpm --extras=lyrics
+          poetry install --extras=autobpm --extras=lyrics --extras=embedart
           poe test
 
       - if: ${{ env.IS_MAIN_PYTHON == 'true' }}

beets/dbcore/query.py

@@ -411,39 +411,6 @@ def __init__(
         super().__init__(field_name, pattern_int, fast)
 
 
-class BytesQuery(FieldQuery[bytes]):
-    """Match a raw bytes field (i.e., a path). This is a necessary hack
-    to work around the `sqlite3` module's desire to treat `bytes` and
-    `unicode` equivalently in Python 2. Always use this query instead of
-    `MatchQuery` when matching on BLOB values.
-    """
-
-    def __init__(self, field_name: str, pattern: bytes | str | memoryview):
-        # Use a buffer/memoryview representation of the pattern for SQLite
-        # matching. This instructs SQLite to treat the blob as binary
-        # rather than encoded Unicode.
-        if isinstance(pattern, (str, bytes)):
-            if isinstance(pattern, str):
-                bytes_pattern = pattern.encode("utf-8")
-            else:
-                bytes_pattern = pattern
-            self.buf_pattern = memoryview(bytes_pattern)
-        elif isinstance(pattern, memoryview):
-            self.buf_pattern = pattern
-            bytes_pattern = bytes(pattern)
-        else:
-            raise ValueError("pattern must be bytes, str, or memoryview")
-
-        super().__init__(field_name, bytes_pattern)
-
-    def col_clause(self) -> tuple[str, Sequence[SQLiteType]]:
-        return self.field + " = ?", [self.buf_pattern]
-
-    @classmethod
-    def value_match(cls, pattern: bytes, value: Any) -> bool:
-        return pattern == value
-
-
 class NumericQuery(FieldQuery[str]):
     """Matches numeric fields. A syntax using Ruby-style range ellipses
     (``..``) lets users specify one- or two-sided ranges. For example,

beets/importer/tasks.py

@@ -26,7 +26,8 @@
 
 import mediafile
 
-from beets import autotag, config, dbcore, library, plugins, util
+from beets import autotag, config, library, plugins, util
+from beets.dbcore.query import PathQuery
 
 from .state import ImportState
 
@@ -520,9 +521,7 @@ def record_replaced(self, lib: library.Library):
         )
         replaced_album_ids = set()
         for item in self.imported_items():
-            dup_items = list(
-                lib.items(query=dbcore.query.BytesQuery("path", item.path))
-            )
+            dup_items = list(lib.items(query=PathQuery("path", item.path)))
             self.replaced_items[item] = dup_items
             for dup_item in dup_items:
                 if (

beets/library/__init__.py

@@ -0,0 +1,16 @@
+from .exceptions import FileOperationError, ReadError, WriteError
+from .library import Library
+from .models import Album, Item, LibModel
+from .queries import parse_query_parts, parse_query_string
+
+__all__ = [
+    "Library",
+    "LibModel",
+    "Album",
+    "Item",
+    "parse_query_parts",
+    "parse_query_string",
+    "FileOperationError",
+    "ReadError",
+    "WriteError",
+]

beets/library/exceptions.py

@@ -0,0 +1,38 @@
+from beets import util
+
+
+class FileOperationError(Exception):
+    """Indicate an error when interacting with a file on disk.
+
+    Possibilities include an unsupported media type, a permissions
+    error, and an unhandled Mutagen exception.
+    """
+
+    def __init__(self, path, reason):
+        """Create an exception describing an operation on the file at
+        `path` with the underlying (chained) exception `reason`.
+        """
+        super().__init__(path, reason)
+        self.path = path
+        self.reason = reason
+
+    def __str__(self):
+        """Get a string representing the error.
+
+        Describe both the underlying reason and the file path in question.
+        """
+        return f"{util.displayable_path(self.path)}: {self.reason}"
+
+
+class ReadError(FileOperationError):
+    """An error while reading a file (i.e. in `Item.read`)."""
+
+    def __str__(self):
+        return "error reading " + str(super())
+
+
+class WriteError(FileOperationError):
+    """An error while writing a file (i.e. in `Item.write`)."""
+
+    def __str__(self):
+        return "error writing " + str(super())

beets/library/library.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import platformdirs
+
+import beets
+from beets import dbcore
+from beets.util import normpath
+
+from .models import Album, Item
+from .queries import PF_KEY_DEFAULT, parse_query_parts, parse_query_string
+
+if TYPE_CHECKING:
+    from beets.dbcore import Results
+
+
+class Library(dbcore.Database):
+    """A database of music containing songs and albums."""
+
+    _models = (Item, Album)
+
+    def __init__(
+        self,
+        path="library.blb",
+        directory: str | None = None,
+        path_formats=((PF_KEY_DEFAULT, "$artist/$album/$track $title"),),
+        replacements=None,
+    ):
+        timeout = beets.config["timeout"].as_number()
+        super().__init__(path, timeout=timeout)
+
+        self.directory = normpath(directory or platformdirs.user_music_path())
+
+        self.path_formats = path_formats
+        self.replacements = replacements
+
+        # Used for template substitution performance.
+        self._memotable: dict[tuple[str, ...], str] = {}
+
+    # Adding objects to the database.
+
+    def add(self, obj):
+        """Add the :class:`Item` or :class:`Album` object to the library
+        database.
+
+        Return the object's new id.
+        """
+        obj.add(self)
+        self._memotable = {}
+        return obj.id
+
+    def add_album(self, items):
+        """Create a new album consisting of a list of items.
+
+        The items are added to the database if they don't yet have an
+        ID. Return a new :class:`Album` object. The list items must not
+        be empty.
+        """
+        if not items:
+            raise ValueError("need at least one item")
+
+        # Create the album structure using metadata from the first item.
+        values = {key: items[0][key] for key in Album.item_keys}
+        album = Album(self, **values)
+
+        # Add the album structure and set the items' album_id fields.
+        # Store or add the items.
+        with self.transaction():
+            album.add(self)
+            for item in items:
+                item.album_id = album.id
+                if item.id is None:
+                    item.add(self)
+                else:
+                    item.store()
+
+        return album
+
+    # Querying.
+
+    def _fetch(self, model_cls, query, sort=None):
+        """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.
+        try:
+            parsed_sort = None
+            if isinstance(query, str):
+                query, parsed_sort = parse_query_string(query, model_cls)
+            elif isinstance(query, (list, tuple)):
+                query, parsed_sort = parse_query_parts(query, model_cls)
+        except dbcore.query.InvalidQueryArgumentValueError as exc:
+            raise dbcore.InvalidQueryError(query, exc)
+
+        # Any non-null sort specified by the parsed query overrides the
+        # provided sort.
+        if parsed_sort and not isinstance(parsed_sort, dbcore.query.NullSort):
+            sort = parsed_sort
+
+        return super()._fetch(model_cls, query, sort)
+
+    @staticmethod
+    def get_default_album_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():
+        """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]:
+        """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]:
+        """Get :class:`Item` objects matching the query."""
+        return self._fetch(Item, query, sort or self.get_default_item_sort())
+
+    # Convenience accessors.
+
+    def get_item(self, id):
+        """Fetch a :class:`Item` by its ID.
+
+        Return `None` if no match is found.
+        """
+        return self._get(Item, id)
+
+    def get_album(self, item_or_id):
+        """Given an album ID or an item associated with an album, return
+        a :class:`Album` object for the album.
+
+        If no such album exists, return `None`.
+        """
+        if isinstance(item_or_id, int):
+            album_id = item_or_id
+        else:
+            album_id = item_or_id.album_id
+        if album_id is None:
+            return None
+        return self._get(Album, album_id)