DB: define field types outside of the models

Author: snejus

beets/dbcore/db.py

@@ -24,7 +24,7 @@
 import time
 from abc import ABC
 from collections import defaultdict
-from collections.abc import Mapping
+from typing import Mapping
 from itertools import chain
 from sqlite3 import Connection
 from types import TracebackType

beets/dbcore/fields.py

@@ -0,0 +1,100 @@
+from typing import Dict
+
+from . import types
+
+TYPE_BY_FIELD: Dict[str, types.Type] = {
+    "acoustid_fingerprint": types.STRING,
+    "acoustid_id": types.STRING,
+    "added": types.DateType(),
+    "albumartist_credit": types.STRING,
+    "albumartists_credit": types.MULTI_VALUE_DSV,
+    "albumartist_sort": types.STRING,
+    "albumartists_sort": types.MULTI_VALUE_DSV,
+    "albumartists": types.MULTI_VALUE_DSV,
+    "albumartist": types.STRING,
+    "albumdisambig": types.STRING,
+    "album_id": types.FOREIGN_ID,
+    "albumstatus": types.STRING,
+    "album": types.STRING,
+    "albumtypes": types.SEMICOLON_SPACE_DSV,
+    "albumtype": types.STRING,
+    "arranger": types.STRING,
+    "artist_credit": types.STRING,
+    "artists_credit": types.MULTI_VALUE_DSV,
+    "artists_ids": types.MULTI_VALUE_DSV,
+    "artist_sort": types.STRING,
+    "artists_sort": types.MULTI_VALUE_DSV,
+    "artists": types.MULTI_VALUE_DSV,
+    "artist": types.STRING,
+    "artpath": types.PathType(True),
+    "asin": types.STRING,
+    "barcode": types.STRING,
+    "bitdepth": types.INTEGER,
+    "bitrate_mode": types.STRING,
+    "bitrate": types.ScaledInt(1000, "kbps"),
+    "bpm": types.INTEGER,
+    "catalognum": types.STRING,
+    "channels": types.INTEGER,
+    "comments": types.STRING,
+    "composer_sort": types.STRING,
+    "composer": types.STRING,
+    "comp": types.BOOLEAN,
+    "country": types.STRING,
+    "data_source": types.STRING,
+    "day": types.PaddedInt(2),
+    "discogs_albumid": types.INTEGER,
+    "discogs_artistid": types.INTEGER,
+    "discogs_labelid": types.INTEGER,
+    "disctitle": types.STRING,
+    "disctotal": types.PaddedInt(2),
+    "disc": types.PaddedInt(2),
+    "encoder_info": types.STRING,
+    "encoder_settings": types.STRING,
+    "encoder": types.STRING,
+    "format": types.STRING,
+    "genre": types.STRING,
+    "grouping": types.STRING,
+    "id": types.PRIMARY_ID,
+    "initial_key": types.MusicalKey(),
+    "isrc": types.STRING,
+    "label": types.STRING,
+    "language": types.STRING,
+    "length": types.DurationType(),
+    "lyricist": types.STRING,
+    "lyrics": types.STRING,
+    "mb_albumartistids": types.MULTI_VALUE_DSV,
+    "mb_albumartistid": types.STRING,
+    "mb_albumid": types.STRING,
+    "mb_artistids": types.MULTI_VALUE_DSV,
+    "mb_artistid": types.STRING,
+    "mb_releasegroupid": types.STRING,
+    "mb_releasetrackid": types.STRING,
+    "mb_trackid": types.STRING,
+    "mb_workid": types.STRING,
+    "media": types.STRING,
+    "month": types.PaddedInt(2),
+    "mtime": types.DateType(),
+    "original_day": types.PaddedInt(2),
+    "original_month": types.PaddedInt(2),
+    "original_year": types.PaddedInt(4),
+    "path": types.PathType(),
+    "r128_album_gain": types.NULL_FLOAT,
+    "r128_track_gain": types.NULL_FLOAT,
+    "releasegroupdisambig": types.STRING,
+    "release_group_title": types.STRING,
+    "remixer": types.STRING,
+    "rg_album_gain": types.NULL_FLOAT,
+    "rg_album_peak": types.NULL_FLOAT,
+    "rg_track_gain": types.NULL_FLOAT,
+    "rg_track_peak": types.NULL_FLOAT,
+    "samplerate": types.ScaledInt(1000, "kHz"),
+    "script": types.STRING,
+    "style": types.STRING,
+    "title": types.STRING,
+    "trackdisambig": types.STRING,
+    "tracktotal": types.PaddedInt(2),
+    "track": types.PaddedInt(2),
+    "work_disambig": types.STRING,
+    "work": types.STRING,
+    "year": types.PaddedInt(4),
+}

beets/dbcore/query.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import os
 import re
 import unicodedata
 from abc import ABC, abstractmethod
@@ -42,6 +43,11 @@
 
 from beets import util
 
+# To use the SQLite "blob" type, it doesn't suffice to provide a byte
+# string; SQLite treats that as encoded text. Wrapping it in a
+# `memoryview` tells it that we actually mean non-text data.
+BLOB_TYPE = memoryview
+
 if TYPE_CHECKING:
     from beets.dbcore import Model
 
@@ -824,6 +830,8 @@ class DateQuery(FieldQuery):
 
     The value of a date field can be matched against a date interval by
     using an ellipsis interval syntax similar to that of NumericQuery.
+
+    TODO: Make this inherit NumericQuery and reuse its logic.
     """
 
     def __init__(self, field, pattern, fast: bool = True):
@@ -891,6 +899,97 @@ def _convert(self, s: str) -> Optional[float]:
                 )
 
 
+class PathQuery(FieldQuery):
+    """A query that matches all items under a given path.
+
+    Matching can either be case-insensitive or case-sensitive. By
+    default, the behavior depends on the OS: case-insensitive on Windows
+    and case-sensitive otherwise.
+    """
+
+    # For tests
+    force_implicit_query_detection = False
+
+    def __init__(self, field, pattern, fast=True, case_sensitive=None):
+        """Create a path query.
+
+        `pattern` must be a path, either to a file or a directory.
+
+        `case_sensitive` can be a bool or `None`, indicating that the
+        behavior should depend on the filesystem.
+        """
+        super().__init__(field, pattern, fast)
+
+        path = util.normpath(pattern)
+
+        # By default, the case sensitivity depends on the filesystem
+        # that the query path is located on.
+        if case_sensitive is None:
+            case_sensitive = util.case_sensitive(path)
+        self.case_sensitive = case_sensitive
+
+        # Use a normalized-case pattern for case-insensitive matches.
+        if not case_sensitive:
+            # We need to lowercase the entire path, not just the pattern.
+            # In particular, on Windows, the drive letter is otherwise not
+            # lowercased.
+            # This also ensures that the `match()` method below and the SQL
+            # from `col_clause()` do the same thing.
+            path = path.lower()
+
+        # Match the path as a single file.
+        self.file_path = path
+        # As a directory (prefix).
+        self.dir_path = os.path.join(path, b"")
+
+    @classmethod
+    def is_path_query(cls, query_part):
+        """Try to guess whether a unicode query part is a path query.
+
+        Condition: separator precedes colon and the file exists.
+        """
+        colon = query_part.find(":")
+        if colon != -1:
+            query_part = query_part[:colon]
+
+        # Test both `sep` and `altsep` (i.e., both slash and backslash on
+        # Windows).
+        if not (
+            os.sep in query_part or (os.altsep and os.altsep in query_part)
+        ):
+            return False
+
+        if cls.force_implicit_query_detection:
+            return True
+        return os.path.exists(util.syspath(util.normpath(query_part)))
+
+    def match(self, item):
+        path = item.path if self.case_sensitive else item.path.lower()
+        return (path == self.file_path) or path.startswith(self.dir_path)
+
+    def col_clause(self):
+        file_blob = BLOB_TYPE(self.file_path)
+        dir_blob = BLOB_TYPE(self.dir_path)
+
+        if self.case_sensitive:
+            query_part = "({0} = ?) || (substr({0}, 1, ?) = ?)"
+        else:
+            query_part = "(BYTELOWER({0}) = BYTELOWER(?)) || \
+                         (substr(BYTELOWER({0}), 1, ?) = BYTELOWER(?))"
+
+        return query_part.format(self.field), (
+            file_blob,
+            len(dir_blob),
+            dir_blob,
+        )
+
+    def __repr__(self) -> str:
+        return (
+            f"{self.__class__.__name__}({self.field!r}, {self.pattern!r}, "
+            f"fast={self.fast}, case_sensitive={self.case_sensitive})"
+        )
+
+
 # Sorting.
 
 

beets/dbcore/queryparse.py

@@ -14,12 +14,12 @@
 
 """Parsing of strings into DBCore queries."""
 
-import itertools
 import re
 from typing import Collection, Dict, List, Optional, Sequence, Tuple, Type
 
 from .. import library
 from . import Model, query
+from .fields import TYPE_BY_FIELD
 from .query import Query, Sort
 
 PARSE_QUERY_PART_REGEX = re.compile(
@@ -128,11 +128,9 @@ def construct_query_part(
 
     # Use `model_cls` to build up a map from field (or query) names to
     # `Query` classes.
-    query_classes: Dict[str, Type[Query]] = {}
-    for k, t in itertools.chain(
-        model_cls._fields.items(), model_cls._types.items()
-    ):
-        query_classes[k] = t.query
+    query_classes: Dict[str, Type[Query]] = {
+        k: t.query for k, t in TYPE_BY_FIELD.items()
+    }
     query_classes.update(model_cls._queries)  # Non-field queries.
 
     # Parse the string.