change to .rose.toml instead of uuids in folder names

conftest.py

@@ -13,6 +13,11 @@
 logger = logging.getLogger(__name__)
 
 
+@pytest.fixture(autouse=True)
+def debug_logging() -> None:
+    logging.getLogger().setLevel(logging.DEBUG)
+
+
 @pytest.fixture()
 def isolated_dir() -> Iterator[Path]:
     with CliRunner().isolated_filesystem():
@@ -48,7 +53,7 @@ def config(isolated_dir: Path) -> Config:
 
 
 @pytest.fixture()
-def seeded_cache(config: Config) -> Iterator[None]:
+def seeded_cache(config: Config) -> None:
     dirpaths = [
         config.music_source_dir / "r1",
         config.music_source_dir / "r2",
@@ -104,9 +109,6 @@ def seeded_cache(config: Config) -> Iterator[None]:
     for f in musicpaths + imagepaths:
         f.touch()
 
-    yield
-    return None
-
 
 def freeze_database_time(conn: sqlite3.Connection) -> None:
     """

flake.nix

@@ -30,6 +30,7 @@
         click
         fuse
         mutagen
+        tomli-w
         uuid6-python
       ];
       dev-deps = with python.pkgs; [

pyproject.toml

@@ -49,6 +49,8 @@ select = [
 ignore = [
   # Allow shadowing builtins on attributes.
   "A003",
+  # Fixtures that do not return anything do not need leading underscore.
+  "PT004",
 ]
 line-length = 100
 exclude = [".venv"]

rose/cache/testdata/Test Release 2/.rose.toml

@@ -0,0 +1,2 @@
+uuid = "ilovecarly"
+new = true

rose/cache/update.py

@@ -1,9 +1,10 @@
 import logging
 import os
-import re
-from dataclasses import asdict
+from dataclasses import asdict, dataclass
 from pathlib import Path
 
+import tomli_w
+import tomllib
 import uuid6
 
 from rose.artiststr import format_artist_string
@@ -41,17 +42,15 @@
     "unknown",
 ]
 
-ID_REGEX = re.compile(r"\{id=([^\}]+)\}$")
-
 
 def update_cache_for_all_releases(c: Config) -> None:
     """
     Process and update the cache for all releases. Delete any nonexistent releases.
     """
     dirs = [Path(d.path).resolve() for d in os.scandir(c.music_source_dir) if d.is_dir()]
     logger.info(f"Found {len(dirs)} releases to update")
-    for i, d in enumerate(dirs):
-        dirs[i] = update_cache_for_release(c, d)
+    for d in dirs:
+        update_cache_for_release(c, d)
     logger.info("Deleting cached releases that are not on disk")
     with connect(c) as conn:
         conn.execute(
@@ -63,7 +62,7 @@ def update_cache_for_all_releases(c: Config) -> None:
         )
 
 
-def update_cache_for_release(c: Config, release_dir: Path) -> Path:
+def update_cache_for_release(c: Config, release_dir: Path) -> None:
     """
     Given a release's directory, update the cache entry based on the release's metadata. If this is
     a new release or track, update the directory and file names to include the UUIDs.
@@ -76,11 +75,9 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
         release: CachedRelease | None = None
         # But first, parse the release_id from the directory name. If the directory name does not
         # contain a release_id, generate one and rename the directory.
-        release_id = _parse_uuid_from_path(release_dir)
-        if not release_id:
-            release_id = str(uuid6.uuid7())
-            logger.debug(f"Assigning id={release_id} to release {release_dir.name}")
-            release_dir = _rename_with_uuid(release_dir, release_id)
+        stored_release_data = _read_stored_data_file(release_dir)
+        if not stored_release_data:
+            stored_release_data = _create_stored_data_file(release_dir)
 
         # Fetch all track tags from disk.
         track_tags: list[tuple[os.DirEntry[str], AudioFile]] = []
@@ -126,7 +123,7 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
                             SELECT * FROM releases WHERE virtual_dirname = ? AND id <> ?
                         )
                         """,
-                        (virtual_dirname, release_id),
+                        (virtual_dirname, stored_release_data.uuid),
                     )
                     if not cursor.fetchone()[0]:
                         break
@@ -142,7 +139,7 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
 
                 # Construct the cached release.
                 release = CachedRelease(
-                    id=release_id,
+                    id=stored_release_data.uuid,
                     source_path=release_dir.resolve(),
                     cover_image_path=cover_image_path,
                     virtual_dirname=virtual_dirname,
@@ -228,11 +225,8 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
             # Now process the track. Release is guaranteed to exist here.
             filepath = Path(f.path)
 
-            track_id = _parse_uuid_from_path(filepath)
-            if not track_id:
-                track_id = str(uuid6.uuid7())
-                logger.debug(f"Assigning id={release_id} to track {filepath.name}")
-                filepath = _rename_with_uuid(filepath, track_id)
+            # Track ID is transient with the cache; we don't put it in any persistent stores.
+            track_id = str(uuid6.uuid7())
 
             virtual_filename = ""
             if multidisc and tags.disc_number:
@@ -319,18 +313,41 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
                     (track.id, art.name, sanitize_filename(art.name), art.role, art.role),
                 )
 
-    return release_dir
 
+STORED_DATA_FILE_NAME = ".rose.toml"
+
+
+@dataclass
+class StoredDataFile:
+    uuid: str
+    new: bool
 
-def _parse_uuid_from_path(path: Path) -> str | None:
-    if m := ID_REGEX.search(path.name if path.is_dir() else path.stem):
-        return m[1]
+
+def _read_stored_data_file(path: Path) -> StoredDataFile | None:
+    for f in path.iterdir():
+        if f.name == STORED_DATA_FILE_NAME:
+            logger.debug(f"Found stored data file for {path}")
+            with f.open("rb") as fp:
+                diskdata = tomllib.load(fp)
+            datafile = StoredDataFile(
+                uuid=diskdata.get("uuid", str(uuid6.uuid7())),
+                new=diskdata.get("new", True),
+            )
+            resolveddata = asdict(datafile)
+            if resolveddata != diskdata:
+                logger.debug(f"Setting new default values in stored data file for {path}")
+                with f.open("wb") as fp:
+                    tomli_w.dump(resolveddata, fp)
+            return datafile
     return None
 
 
-def _rename_with_uuid(src: Path, uuid: str) -> Path:
-    if src.is_dir():
-        dst = src.with_name(src.name + f" {{id={uuid}}}")
-    else:
-        dst = src.with_stem(src.stem + f" {{id={uuid}}}")
-    return src.rename(dst)
+def _create_stored_data_file(path: Path) -> StoredDataFile:
+    logger.debug(f"Creating stored data file for {path}")
+    data = StoredDataFile(
+        uuid=str(uuid6.uuid7()),
+        new=True,
+    )
+    with (path / ".rose.toml").open("wb") as fp:
+        tomli_w.dump(asdict(data), fp)
+    return data

rose/cache/update_test.py

@@ -1,24 +1,33 @@
 import shutil
 from pathlib import Path
 
+import tomllib
+
 from rose.cache.database import connect
-from rose.cache.update import ID_REGEX, update_cache_for_all_releases, update_cache_for_release
+from rose.cache.update import (
+    STORED_DATA_FILE_NAME,
+    update_cache_for_all_releases,
+    update_cache_for_release,
+)
 from rose.foundation.conf import Config
 
 TESTDATA = Path(__file__).resolve().parent / "testdata"
 TEST_RELEASE_1 = TESTDATA / "Test Release 1"
-TEST_RELEASE_2 = TESTDATA / "Test Release 2 {id=ilovecarly}"
+TEST_RELEASE_2 = TESTDATA / "Test Release 2"
 
 
 def test_update_cache_for_release(config: Config) -> None:
     release_dir = config.music_source_dir / TEST_RELEASE_1.name
     shutil.copytree(TEST_RELEASE_1, release_dir)
-    updated_release_dir = update_cache_for_release(config, release_dir)
+    update_cache_for_release(config, release_dir)
 
     # Check that the release directory was given a UUID.
-    m = ID_REGEX.search(updated_release_dir.name)
-    assert m is not None
-    release_id = m[1]
+    release_id: str | None = None
+    for f in release_dir.iterdir():
+        if f.name == STORED_DATA_FILE_NAME:
+            with f.open("rb") as fp:
+                release_id = tomllib.load(fp)["uuid"]
+    assert release_id is not None
 
     # Assert that the release metadata was read correctly.
     with connect(config) as conn:
@@ -30,7 +39,7 @@ def test_update_cache_for_release(config: Config) -> None:
             (release_id,),
         )
         row = cursor.fetchone()
-        assert row["source_path"] == str(updated_release_dir)
+        assert row["source_path"] == str(release_dir)
         assert row["title"] == "A Cool Album"
         assert row["release_type"] == "album"
         assert row["release_year"] == 1990
@@ -60,26 +69,21 @@ def test_update_cache_for_release(config: Config) -> None:
             ("Artist B", "main"),
         }
 
-        for f in updated_release_dir.iterdir():
+        for f in release_dir.iterdir():
             if f.suffix != ".m4a":
                 continue
 
-            # Check that the track file was given a UUID.
-            m = ID_REGEX.search(f.stem)
-            assert m is not None
-            track_id = m[1]
-
             # Assert that the track metadata was read correctly.
             cursor = conn.execute(
                 """
                 SELECT
                     id, source_path, title, release_id, track_number, disc_number, duration_seconds
-                FROM tracks WHERE id = ?
+                FROM tracks WHERE source_path = ?
                 """,
-                (track_id,),
+                (str(f),),
             )
             row = cursor.fetchone()
-            assert row["source_path"] == str(f)
+            track_id = row["id"]
             assert row["title"] == "Title"
             assert row["release_id"] == release_id
             assert row["track_number"] != ""
@@ -111,26 +115,15 @@ def test_update_cache_with_existing_id(config: Config) -> None:
     """Test that IDs in filenames are read and preserved."""
     release_dir = config.music_source_dir / TEST_RELEASE_2.name
     shutil.copytree(TEST_RELEASE_2, release_dir)
-    updated_release_dir = update_cache_for_release(config, release_dir)
-    assert release_dir == updated_release_dir
+    update_cache_for_release(config, release_dir)
 
-    with connect(config) as conn:
-        m = ID_REGEX.search(release_dir.name)
-        assert m is not None
-        release_id = m[1]
-        cursor = conn.execute("SELECT EXISTS(SELECT * FROM releases WHERE id = ?)", (release_id,))
-        assert cursor.fetchone()[0]
-
-        for f in release_dir.iterdir():
-            if f.suffix != ".m4a":
-                continue
-
-            # Check that the track file was given a UUID.
-            m = ID_REGEX.search(f.stem)
-            assert m is not None
-            track_id = m[1]
-            cursor = conn.execute("SELECT EXISTS(SELECT * FROM tracks WHERE id = ?)", (track_id,))
-            assert cursor.fetchone()[0]
+    # Check that the release directory was given a UUID.
+    release_id: str | None = None
+    for f in release_dir.iterdir():
+        if f.name == STORED_DATA_FILE_NAME:
+            with f.open("rb") as fp:
+                release_id = tomllib.load(fp)["uuid"]
+    assert release_id == "ilovecarly"  # Hardcoded ID for testing.
 
 
 def test_update_cache_for_all_releases(config: Config) -> None:
@@ -153,17 +146,3 @@ def test_update_cache_for_all_releases(config: Config) -> None:
         assert cursor.fetchone()[0] == 2
         cursor = conn.execute("SELECT COUNT(*) FROM tracks")
         assert cursor.fetchone()[0] == 4
-
-
-def test_update_cache_with_dotted_dirname(config: Config) -> None:
-    # Regression test: If we use with_stem on a directory with a dot, then the directory will be
-    # renamed to like Put.ID.After.The {id=abc}.Dot" which we don't want.
-    release_dir = config.music_source_dir / "Put.ID.After.The.Dot"
-    shutil.copytree(TEST_RELEASE_1, release_dir)
-    updated_release_dir = update_cache_for_release(config, release_dir)
-    m = ID_REGEX.search(updated_release_dir.name)
-    assert m is not None
-
-    # Regression test 2: Don't create a new ID; read the existing ID.
-    updated_release_dir2 = update_cache_for_release(config, updated_release_dir)
-    assert updated_release_dir2 == updated_release_dir

setup.py

@@ -13,6 +13,7 @@
         "click",
         "fuse-python",
         "mutagen",
+        "tomli-w",
         "uuid6-python",
     ],
 )