Store paths relative to music dir in DB, expand on read

Move path relativization/expansion logic from Item._setitem/__getitem__
into dbcore layer (PathType.to_sql/from_sql and PathQuery), so all
models benefit without per-model overrides.

Propagate contextvars to pipeline and replaygain pool threads so the
library root context variable is available during background processing.

Author: snejus

beets/dbcore/pathutils.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import os
+
+from beets import context, util
+
+
+def normalize_path_for_db(path: bytes) -> bytes:
+    """Convert an absolute library path to its database representation."""
+    if not path or not os.path.isabs(path):
+        return path
+
+    music_dir = context.get_music_dir()
+    if not music_dir:
+        return path
+
+    if path == music_dir:
+        return os.path.relpath(path, music_dir)
+
+    if path.startswith(os.path.join(music_dir, b"")):
+        return os.path.relpath(path, music_dir)
+
+    return path
+
+
+def expand_path_from_db(path: bytes) -> bytes:
+    """Convert a stored database path to an absolute library path."""
+    music_dir = context.get_music_dir()
+    if path and not os.path.isabs(path) and music_dir:
+        return util.normpath(os.path.join(music_dir, path))
+
+    return path

beets/dbcore/query.py

@@ -30,6 +30,8 @@
 from beets import util
 from beets.util.units import raw_seconds_short
 
+from . import pathutils
+
 if TYPE_CHECKING:
     from collections.abc import Iterator, MutableSequence
 
@@ -293,6 +295,9 @@ def __init__(self, field: str, pattern: bytes, fast: bool = True) -> None:
 
         # Case sensitivity depends on the filesystem that the query path is located on.
         self.case_sensitive = util.case_sensitive(path)
+        # Path queries compare against the DB representation, which is relative
+        # to the library root when the file lives inside it.
+        path = pathutils.normalize_path_for_db(path)
 
         # Use a normalized-case pattern for case-insensitive matches.
         if not self.case_sensitive:
@@ -333,7 +338,9 @@ def match(self, obj: Model) -> bool:
         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()
+        path = pathutils.normalize_path_for_db(obj.path)
+        if not self.case_sensitive:
+            path = path.lower()
         return (path == self.pattern) or path.startswith(self.dir_path)
 
     def col_clause(self) -> tuple[str, Sequence[SQLiteType]]:

beets/dbcore/types.py

@@ -16,17 +16,18 @@
 
 from __future__ import annotations
 
+import os
 import re
 import time
 import typing
 from abc import ABC
 from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar, cast
 
 import beets
-from beets import util
+from beets import context, util
 from beets.util.units import human_seconds_short, raw_seconds_short
 
-from . import query
+from . import pathutils, query
 
 SQLiteType = query.SQLiteType
 BLOB_TYPE = query.BLOB_TYPE
@@ -389,9 +390,10 @@ def normalize(self, value: Any) -> bytes | N:
             return value
 
     def from_sql(self, sql_value):
-        return self.normalize(sql_value)
+        return pathutils.expand_path_from_db(self.normalize(sql_value))
 
     def to_sql(self, value: bytes) -> BLOB_TYPE:
+        value = pathutils.normalize_path_for_db(value)
         if isinstance(value, bytes):
             value = BLOB_TYPE(value)
         return value

beets/library/models.py

@@ -7,7 +7,7 @@
 import unicodedata
 from functools import cached_property
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, ClassVar
+from typing import TYPE_CHECKING, ClassVar
 
 from mediafile import MediaFile, UnreadableFileError
 
@@ -81,30 +81,6 @@ def add(self, lib=None):
         # so don't do it here
         super().add(lib)
 
-    def _setitem(self, key: str, value: Any):
-        """Set the item's value for a standard field or a flexattr."""
-        # Encode unicode paths and read buffers.
-        if key == "path":
-            if isinstance(value, str):
-                value = bytestring_path(value)
-            elif isinstance(value, types.BLOB_TYPE):
-                value = bytes(value)
-            # Store paths relative to the music directory
-            # Check for absolute path because item may be initialised with
-            # a relative path already
-            if os.path.isabs(value) and (music_dir := context.get_music_dir()):
-                value = os.path.relpath(value, music_dir)
-
-        return super()._setitem(key, value)
-
-    def __getitem__(self, key: str):
-        value = super().__getitem__(key)
-        if key == "path" and value:
-            # Return absolute paths.
-            value = normpath(os.path.join(context.get_music_dir(), value))
-
-        return value
-
     def __format__(self, spec):
         if not spec:
             spec = beets.config[self._format_config_key].as_str()
@@ -125,6 +101,22 @@ def field_query(
     ) -> FieldQuery:
         """Get a `FieldQuery` for the given field on this model."""
         fast = field in cls.all_db_fields
+        if (
+            cls._type(field).query is dbcore.query.PathQuery
+            and query_cls is not dbcore.query.PathQuery
+            and (music_dir := context.get_music_dir())
+        ):
+            # Regex, exact, and string queries operate on the raw DB value, so
+            # strip the library prefix to match the stored relative path.
+            if isinstance(pattern, bytes):
+                prefix = os.path.join(music_dir, b"")
+                if pattern.startswith(prefix):
+                    pattern = os.path.relpath(pattern, music_dir)
+            else:
+                music_dir_str = os.fsdecode(music_dir)
+                prefix = music_dir_str + os.sep
+                if pattern.startswith(prefix):
+                    pattern = pattern.removeprefix(prefix)
         if field in cls.shared_db_fields:
             # This field exists in both tables, so SQLite will encounter
             # an OperationalError if we try to use it in a query.
@@ -849,7 +841,12 @@ def from_path(cls, path):
     def __setitem__(self, key, value):
         """Set the item's value for a standard field or a flexattr."""
         # Encode unicode paths and read buffers.
-        if key == "album_id":
+        if key == "path":
+            if isinstance(value, str):
+                value = bytestring_path(value)
+            elif isinstance(value, types.BLOB_TYPE):
+                value = bytes(value)
+        elif key == "album_id":
             self._cached_album = None
 
         changed = super()._setitem(key, value)

beets/util/pipeline.py

@@ -33,6 +33,7 @@
 
 from __future__ import annotations
 
+import contextvars
 import queue
 import sys
 from threading import Lock, Thread
@@ -237,12 +238,18 @@ def _allmsgs(obj):
 class PipelineThread(Thread):
     """Abstract base class for pipeline-stage threads."""
 
-    def __init__(self, all_threads):
+    def __init__(self, all_threads, ctx: contextvars.Context | None = None):
         super().__init__()
         self.abort_lock = Lock()
         self.abort_flag = False
         self.all_threads = all_threads
         self.exc_info = None
+        self.ctx = ctx
+
+    def _run_in_context(self, func, *args):
+        if self.ctx is None:
+            return func(*args)
+        return self.ctx.run(func, *args)
 
     def abort(self):
         """Shut down the thread at the next chance possible."""
@@ -267,8 +274,8 @@ class FirstPipelineThread(PipelineThread):
     The coroutine should just be a generator.
     """
 
-    def __init__(self, coro, out_queue, all_threads):
-        super().__init__(all_threads)
+    def __init__(self, coro, out_queue, all_threads, ctx=None):
+        super().__init__(all_threads, ctx)
         self.coro = coro
         self.out_queue = out_queue
         self.out_queue.acquire()
@@ -282,7 +289,7 @@ def run(self):
 
                 # Get the value from the generator.
                 try:
-                    msg = next(self.coro)
+                    msg = self._run_in_context(next, self.coro)
                 except StopIteration:
                     break
 
@@ -306,8 +313,8 @@ class MiddlePipelineThread(PipelineThread):
     last.
     """
 
-    def __init__(self, coro, in_queue, out_queue, all_threads):
-        super().__init__(all_threads)
+    def __init__(self, coro, in_queue, out_queue, all_threads, ctx=None):
+        super().__init__(all_threads, ctx)
         self.coro = coro
         self.in_queue = in_queue
         self.out_queue = out_queue
@@ -316,7 +323,7 @@ def __init__(self, coro, in_queue, out_queue, all_threads):
     def run(self):
         try:
             # Prime the coroutine.
-            next(self.coro)
+            self._run_in_context(next, self.coro)
 
             while True:
                 with self.abort_lock:
@@ -333,7 +340,7 @@ def run(self):
                         return
 
                 # Invoke the current stage.
-                out = self.coro.send(msg)
+                out = self._run_in_context(self.coro.send, msg)
 
                 # Send messages to next stage.
                 for msg in _allmsgs(out):
@@ -355,14 +362,14 @@ class LastPipelineThread(PipelineThread):
     should yield nothing.
     """
 
-    def __init__(self, coro, in_queue, all_threads):
-        super().__init__(all_threads)
+    def __init__(self, coro, in_queue, all_threads, ctx=None):
+        super().__init__(all_threads, ctx)
         self.coro = coro
         self.in_queue = in_queue
 
     def run(self):
         # Prime the coroutine.
-        next(self.coro)
+        self._run_in_context(next, self.coro)
 
         try:
             while True:
@@ -380,7 +387,7 @@ def run(self):
                         return
 
                 # Send to consumer.
-                self.coro.send(msg)
+                self._run_in_context(self.coro.send, msg)
 
         except BaseException:
             self.abort_all(sys.exc_info())
@@ -419,26 +426,37 @@ def run_parallel(self, queue_size=DEFAULT_QUEUE_SIZE):
         messages between the stages are stored in queues of the given
         size.
         """
+        base_ctx = contextvars.copy_context()
         queue_count = len(self.stages) - 1
         queues = [CountedQueue(queue_size) for i in range(queue_count)]
         threads = []
 
         # Set up first stage.
         for coro in self.stages[0]:
-            threads.append(FirstPipelineThread(coro, queues[0], threads))
+            # Each worker needs its own copy because Context objects cannot be
+            # entered concurrently from multiple threads.
+            threads.append(
+                FirstPipelineThread(coro, queues[0], threads, base_ctx.copy())
+            )
 
         # Middle stages.
         for i in range(1, queue_count):
             for coro in self.stages[i]:
                 threads.append(
                     MiddlePipelineThread(
-                        coro, queues[i - 1], queues[i], threads
+                        coro,
+                        queues[i - 1],
+                        queues[i],
+                        threads,
+                        base_ctx.copy(),
                     )
                 )
 
         # Last stage.
         for coro in self.stages[-1]:
-            threads.append(LastPipelineThread(coro, queues[-1], threads))
+            threads.append(
+                LastPipelineThread(coro, queues[-1], threads, base_ctx.copy())
+            )
 
         # Start threads.
         for thread in threads:

beetsplug/replaygain.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import collections
+import contextvars
 import enum
 import math
 import os
@@ -1427,6 +1428,9 @@ def _apply(
         callback: Callable[[AnyRgTask], Any],
     ):
         if self.pool is not None:
+            # Apply the caller's context to both the worker and its callbacks
+            # so lazy path expansion keeps the library root in pool threads.
+            ctx = contextvars.copy_context()
 
             def handle_exc(exc):
                 """Handle exceptions in the async work."""
@@ -1435,8 +1439,19 @@ def handle_exc(exc):
                 else:
                     self.exc_queue.put(exc)
 
+            def run_func():
+                return ctx.run(func, *args, **kwds)
+
+            def run_callback(task: AnyRgTask):
+                return ctx.run(callback, task)
+
+            def run_handle_exc(exc):
+                return ctx.run(handle_exc, exc)
+
             self.pool.apply_async(
-                func, args, kwds, callback, error_callback=handle_exc
+                run_func,
+                callback=run_callback,
+                error_callback=run_handle_exc,
             )
         else:
             callback(func(*args, **kwds))

test/test_library.py

@@ -1098,8 +1098,12 @@ def test_artpath_stores_special_chars(self):
         alb = self.lib.add_album([self.i])
         alb.artpath = path
         alb.store()
+        stored_path = self.lib._connection().execute(
+            "select artpath from albums where id=?", (alb.id,)
+        ).fetchone()[0]
         alb = self.lib.get_album(self.i)
-        assert path == alb.artpath
+        assert stored_path == path
+        assert alb.artpath == os.path.join(self.libdir, path)
 
     def test_sanitize_path_with_special_chars(self):
         path = "b\xe1r?"
@@ -1129,10 +1133,13 @@ def test_relative_path_is_stored(self):
         absolute_path = os.path.join(self.libdir, relative_path)
         self.i.path = absolute_path
         self.i.store()
+        stored_path = self.lib._connection().execute(
+            "select path from items where id=?", (self.i.id,)
+        ).fetchone()[0]
         album = self.lib.add_album([self.i])
 
         assert self.i.path == absolute_path
-        assert self.i._values_fixed["path"] == relative_path
+        assert stored_path == relative_path
         assert album.path == os.path.dirname(absolute_path)