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

Author: snejus

.git-blame-ignore-revs

@@ -48,4 +48,6 @@ f36bc497c8c8f89004f3f6879908d3f0b25123e1
 # Fix formatting
 c490ac5810b70f3cf5fd8649669838e8fdb19f4d
 # Importer restructure
-9147577b2b19f43ca827e9650261a86fb0450cef
+9147577b2b19f43ca827e9650261a86fb0450cef
+# Copy paste query, types from library to dbcore
+1a045c91668c771686f4c871c84f1680af2e944b

beets/dbcore/query.py

@@ -16,26 +16,33 @@
 
 from __future__ import annotations
 
+import os
 import re
 import unicodedata
 from abc import ABC, abstractmethod
 from collections.abc import Iterator, MutableSequence, Sequence
 from datetime import datetime, timedelta
-from functools import reduce
+from functools import cached_property, reduce
 from operator import mul, or_
 from re import Pattern
 from typing import TYPE_CHECKING, Any, Generic, TypeVar, Union
 
 from beets import util
 
 if TYPE_CHECKING:
-    from beets.dbcore import Model
-    from beets.dbcore.db import AnyModel
+    from beets.dbcore.db import AnyModel, Model
 
     P = TypeVar("P", default=Any)
 else:
     P = TypeVar("P")
 
+# 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.
+# needs to be defined in here due to circular import.
+# TODO: remove it from this module and define it in dbcore/types.py instead
+BLOB_TYPE = memoryview
+
 
 class ParsingError(ValueError):
     """Abstract class for any unparsable user-requested album/query
@@ -78,6 +85,7 @@ def field_names(self) -> set[str]:
         """Return a set with field names that this query operates on."""
         return set()
 
+    @abstractmethod
     def clause(self) -> tuple[str | None, Sequence[Any]]:
         """Generate an SQLite expression implementing the query.
 
@@ -88,14 +96,12 @@ def clause(self) -> tuple[str | None, Sequence[Any]]:
         The default implementation returns None, falling back to a slow query
         using `match()`.
         """
-        return None, ()
 
     @abstractmethod
     def match(self, obj: Model):
         """Check whether this query matches a given Model. Can be used to
         perform queries on arbitrary sets of Model.
         """
-        ...
 
     def __and__(self, other: Query) -> AndQuery:
         return AndQuery([self, other])
@@ -145,7 +151,7 @@ def __init__(self, field_name: str, pattern: P, fast: bool = True):
         self.fast = fast
 
     def col_clause(self) -> tuple[str, Sequence[SQLiteType]]:
-        return self.field, ()
+        raise NotImplementedError
 
     def clause(self) -> tuple[str | None, Sequence[SQLiteType]]:
         if self.fast:
@@ -157,7 +163,7 @@ def clause(self) -> tuple[str | None, Sequence[SQLiteType]]:
     @classmethod
     def value_match(cls, pattern: P, value: Any):
         """Determine whether the value matches the pattern."""
-        raise NotImplementedError()
+        raise NotImplementedError
 
     def match(self, obj: Model) -> bool:
         return self.value_match(self.pattern, obj.get(self.field_name))
@@ -227,7 +233,7 @@ def string_match(
         """Determine whether the value matches the pattern. Both
         arguments are strings. Subclasses implement this method.
         """
-        raise NotImplementedError()
+        raise NotImplementedError
 
 
 class StringQuery(StringFieldQuery[str]):
@@ -267,6 +273,91 @@ def string_match(cls, pattern: str, value: str) -> bool:
         return pattern.lower() in value.lower()
 
 
+class PathQuery(FieldQuery[bytes]):
+    """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.
+    """
+
+    def __init__(self, field: str, pattern: bytes, fast: bool = True) -> None:
+        """Create a path query.
+
+        `pattern` must be a path, either to a file or a directory.
+        """
+        path = util.normpath(pattern)
+
+        # Case sensitivity depends on the filesystem that the query path is located on.
+        self.case_sensitive = util.case_sensitive(path)
+
+        # Use a normalized-case pattern for case-insensitive matches.
+        if not self.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()
+
+        super().__init__(field, path, fast)
+
+    @cached_property
+    def dir_path(self) -> bytes:
+        return os.path.join(self.pattern, b"")
+
+    @staticmethod
+    def is_path_query(query_part: str) -> bool:
+        """Try to guess whether a unicode query part is a path query.
+
+        The path query must
+        1. precede the colon in the query, if a colon is present
+        2. contain either ``os.sep`` or ``os.altsep`` (Windows)
+        3. this path must exist on the filesystem.
+        """
+        query_part = query_part.split(":")[0]
+
+        return (
+            # make sure the query part contains a path separator
+            bool(set(query_part) & {os.sep, os.altsep})
+            and os.path.exists(util.normpath(query_part))
+        )
+
+    def match(self, obj: Model) -> bool:
+        """Check whether a model object's path matches this query.
+
+        Performs either an exact match against the pattern or checks if the path
+        starts with the given directory path. Case sensitivity depends on the object's
+        filesystem as determined during initialization.
+        """
+        path = obj.path if self.case_sensitive else obj.path.lower()
+        return (path == self.pattern) or path.startswith(self.dir_path)
+
+    def col_clause(self) -> tuple[str, Sequence[SQLiteType]]:
+        """Generate an SQL clause that implements path matching in the database.
+
+        Returns a tuple of SQL clause string and parameter values list that matches
+        paths either exactly or by directory prefix. Handles case sensitivity
+        appropriately using BYTELOWER for case-insensitive matches.
+        """
+        if self.case_sensitive:
+            left, right = self.field, "?"
+        else:
+            left, right = f"BYTELOWER({self.field})", "BYTELOWER(?)"
+
+        return f"({left} = {right}) || (substr({left}, 1, ?) = {right})", [
+            BLOB_TYPE(self.pattern),
+            len(dir_blob := BLOB_TYPE(self.dir_path)),
+            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})"
+        )
+
+
 class RegexpQuery(StringFieldQuery[Pattern[str]]):
     """A query that matches a regular expression in a specific Model field.
 
@@ -844,6 +935,24 @@ def _convert(self, s: str) -> float | None:
                 )
 
 
+class SingletonQuery(FieldQuery[str]):
+    """This query is responsible for the 'singleton' lookup.
+
+    It is based on the FieldQuery and constructs a SQL clause
+    'album_id is NULL' which yields the same result as the previous filter
+    in Python but is more performant since it's done in SQL.
+
+    Using util.str2bool ensures that lookups like singleton:true, singleton:1
+    and singleton:false, singleton:0 are handled consistently.
+    """
+
+    def __new__(cls, field: str, value: str, *args, **kwargs):
+        query = NoneQuery("album_id")
+        if util.str2bool(value):
+            return query
+        return NotQuery(query)
+
+
 # Sorting.