Split library file into different files inside library folder.

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)

beets/library/queries.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import shlex
+
+import beets
+from beets import dbcore, logging, plugins
+
+log = logging.getLogger("beets")
+
+
+# Special path format key.
+PF_KEY_DEFAULT = "default"
+
+# Query construction helpers.
+
+
+def parse_query_parts(parts, model_cls):
+    """Given a beets query string as a list of components, return the
+    `Query` and `Sort` they represent.
+
+    Like `dbcore.parse_sorted_query`, with beets query prefixes and
+    ensuring that implicit path queries are made explicit with 'path::<query>'
+    """
+    # Get query types and their prefix characters.
+    prefixes = {
+        ":": dbcore.query.RegexpQuery,
+        "=~": dbcore.query.StringQuery,
+        "=": dbcore.query.MatchQuery,
+    }
+    prefixes.update(plugins.queries())
+
+    # Special-case path-like queries, which are non-field queries
+    # containing path separators (/).
+    parts = [
+        f"path:{s}" if dbcore.query.PathQuery.is_path_query(s) else s
+        for s in parts
+    ]
+
+    case_insensitive = beets.config["sort_case_insensitive"].get(bool)
+
+    query, sort = dbcore.parse_sorted_query(
+        model_cls, parts, prefixes, case_insensitive
+    )
+    log.debug("Parsed query: {!r}", query)
+    log.debug("Parsed sort: {!r}", sort)
+    return query, sort
+
+
+def parse_query_string(s, model_cls):
+    """Given a beets query string, return the `Query` and `Sort` they
+    represent.
+
+    The string is split into components using shell-like syntax.
+    """
+    message = f"Query is not unicode: {s!r}"
+    assert isinstance(s, str), message
+    try:
+        parts = shlex.split(s)
+    except ValueError as exc:
+        raise dbcore.InvalidQueryError(s, exc)
+    return parse_query_parts(parts, model_cls)