Define InQuery and use it in PlaylistQuery

While working on the DB optimisation I discovered one query which does
not follow the 'FieldQuery' interface - 'PlaylistQuery', so I looked
into it in more detail.

One special thing about it is that it uses 'IN' SQL operator, so
I defined 'InQuery' query class to have this logic outside of the
playlist context.

Otherwise, it seems like 'PlaylistQuery' is a field query, even if it
has a very special way of resolving values it wants to query. In the
future, we may want to consider moving this kind of custom
_initialisation_ logic away from '__init__' methods to
factory/@classmethod: this should make it more clear that the purpose of
such logic is to resolve the data that is required to define
a particular FieldQuery class fully.
This commit is contained in:
Šarūnas Nejus 2024-04-25 17:26:07 +01:00 committed by Šarūnas Nejus
parent 4354ba4f97
commit 68eee96c03
No known key found for this signature in database
GPG key ID: DD28F6704DBE3435
2 changed files with 35 additions and 20 deletions

View file

@ -12,8 +12,7 @@
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
"""The Query type hierarchy for DBCore.
"""
"""The Query type hierarchy for DBCore."""
from __future__ import annotations
@ -155,9 +154,7 @@ class FieldQuery(Query, Generic[P]):
@classmethod
def value_match(cls, pattern: P, value: Any):
"""Determine whether the value matches the pattern. Both
arguments are strings.
"""
"""Determine whether the value matches the pattern."""
raise NotImplementedError()
def match(self, obj: Model) -> bool:
@ -428,6 +425,28 @@ class NumericQuery(FieldQuery):
return "1", ()
class InQuery(FieldQuery[Sequence[AnySQLiteType]]):
"""Query which matches values in the given set."""
field: str
pattern: Sequence[AnySQLiteType]
fast: bool = True
@property
def subvals(self) -> Sequence[AnySQLiteType]:
return self.pattern
def col_clause(self) -> Tuple[str, Sequence[AnySQLiteType]]:
placeholders = ", ".join(["?"] * len(self.subvals))
return f"{self.field} IN ({placeholders})", self.subvals
@classmethod
def value_match(
cls, pattern: Sequence[AnySQLiteType], value: AnySQLiteType
) -> bool:
return value in pattern
class CollectionQuery(Query):
"""An abstract query class that aggregates other queries. Can be
indexed like a list to access the sub-queries.

View file

@ -15,17 +15,18 @@
import fnmatch
import os
import tempfile
from typing import Any, Optional, Sequence, Tuple
from typing import Sequence
import beets
from beets.dbcore.query import InQuery
from beets.library import BLOB_TYPE
from beets.util import path_as_posix
class PlaylistQuery(beets.dbcore.NamedQuery):
class PlaylistQuery(InQuery):
"""Matches files listed by a playlist file."""
def __init__(self, pattern):
self.pattern = pattern
def __init__(self, _, pattern: str, __):
config = beets.config["playlist"]
# Get the full path to the playlist
@ -39,7 +40,7 @@ class PlaylistQuery(beets.dbcore.NamedQuery):
),
)
self.paths = []
paths = []
for playlist_path in playlist_paths:
if not fnmatch.fnmatch(playlist_path, "*.[mM]3[uU]"):
# This is not am M3U playlist, skip this candidate
@ -63,23 +64,18 @@ class PlaylistQuery(beets.dbcore.NamedQuery):
# ignore comments, and extm3u extension
continue
self.paths.append(
paths.append(
beets.util.normpath(
os.path.join(relative_to, line.rstrip())
)
)
f.close()
break
super().__init__("path", paths)
def clause(self) -> Tuple[Optional[str], Sequence[Any]]:
if not self.paths:
# Playlist is empty
return "0", ()
clause = "path IN ({})".format(", ".join("?" for path in self.paths))
return clause, (beets.library.BLOB_TYPE(p) for p in self.paths)
def match(self, item):
return item.path in self.paths
@property
def subvals(self) -> Sequence[BLOB_TYPE]:
return [BLOB_TYPE(p) for p in self.pattern]
class PlaylistPlugin(beets.plugins.BeetsPlugin):