simplified migration, one line indices for albums.

Author: semohr

beets/dbcore/db.py

@@ -1225,19 +1225,11 @@ def _migrate_indices(
         indices are created or replaced.
         """
         with self.transaction() as tx:
-            index_rows = tx.query(f"PRAGMA index_list({table})")
-            current_indices = {Index.from_db(tx, row[1]) for row in index_rows}
-
-        _indices = set(indices)
-
-        if current_indices.issuperset(_indices):
-            return
-
-        # May also include missing indices.
-        changed_indices = _indices - current_indices
-
-        with self.transaction() as tx:
-            for index in changed_indices:
+            current = {
+                Index.from_db(tx, r[1])
+                for r in tx.query(f"PRAGMA index_list({table})")
+            }
+            for index in set(indices) - current:
                 index.recreate(tx, table)
 
     # Querying.
@@ -1317,22 +1309,18 @@ class Index(NamedTuple):
     """
 
     name: str
-    columns: Sequence[str]
+    columns: tuple[str, ...]
 
     def recreate(self, tx: Transaction, table: str) -> None:
         """Recreate the index in the database.
 
         This is useful when the index has been changed and needs to be
         updated.
         """
-        tx.script(f"DROP INDEX IF EXISTS {self.name}")
-        self.create(tx, table)
-
-    def create(self, tx: Transaction, table: str) -> None:
-        """Create the index in the database."""
-        return tx.script(
-            f"CREATE INDEX {self.name} ON {table} ({', '.join(self.columns)})"
-        )
+        tx.script(f"""
+            DROP INDEX IF EXISTS {self.name};
+            CREATE INDEX {self.name} ON {table} ({", ".join(self.columns)})
+        """)
 
     @classmethod
     def from_db(cls, tx: Transaction, name: str) -> Index:
@@ -1342,7 +1330,7 @@ def from_db(cls, tx: Transaction, name: str) -> Index:
         Error will be raised.
         """
         rows = tx.query(f"PRAGMA index_info({name})")
-        columns = [row[2] for row in rows]
+        columns = tuple(row[2] for row in rows)
         return cls(name, columns)
 
     def __hash__(self) -> int:

beets/library/models.py

@@ -716,12 +716,7 @@ class Item(LibModel):
         "mtime": types.DATE,
         "added": types.DATE,
     }
-    _indices = (
-        dbcore.Index(
-            name="idx_item_album_id",
-            columns=("album_id",),
-        ),
-    )
+    _indices = (dbcore.Index("idx_item_album_id", ("album_id",)),)
 
     _search_fields = (
         "artist",

test/test_dbcore.py

@@ -800,12 +800,12 @@ def db(self):
     @pytest.fixture
     def sample_index(self):
         """Fixture for a sample Index object."""
-        return Index(name="sample_index", columns=["field_one"])
+        return Index(name="sample_index", columns=("field_one",))
 
     def test_index_creation(self, db, sample_index):
         """Test creating an index and checking its existence."""
         with db.transaction() as tx:
-            sample_index.create(tx, "test")
+            sample_index.recreate(tx, "test")
             indexes = (
                 db._connection().execute("PRAGMA index_list(test)").fetchall()
             )
@@ -814,18 +814,18 @@ def test_index_creation(self, db, sample_index):
     def test_from_db(self, db, sample_index):
         """Test retrieving an index from the database."""
         with db.transaction() as tx:
-            sample_index.create(tx, "test")
+            sample_index.recreate(tx, "test")
             retrieved = Index.from_db(tx, sample_index.name)
             assert retrieved == sample_index
 
     def test_index_hashing_and_set_behavior(self, sample_index):
         """Test the hashing and set behavior of the Index class."""
         index_set = {sample_index}
-        similar_index = Index(name="sample_index", columns=["field_one"])
+        similar_index = Index(name="sample_index", columns=("field_one",))
 
         assert similar_index in index_set  # Should recognize similar attributes
 
-        different_index = Index(name="other_index", columns=["other_field"])
+        different_index = Index(name="other_index", columns=("other_field",))
         index_set.add(different_index)
 
         assert len(index_set) == 2  # Should recognize distinct index