app/playlistmodel.py

@@ -1,6 +1,5 @@
 from datetime import datetime
 from enum import auto, Enum
-from sqlalchemy import bindparam, update
 from typing import List, Optional, TYPE_CHECKING
 
 from dbconfig import scoped_session, Session
@@ -85,16 +84,13 @@ class PlaylistModel(QAbstractTableModel):
     """
     The Playlist Model
 
-    Update strategy: update the database and then refresh the
-    row-indexed cached copy (self.playlist_rows). Do not edit
-    self.playlist_rows directly because keeping it and the
-    database in sync is uncessarily challenging.
+    Update strategy: update the database and then refresh the cached copy (self.playlist_rows).
+    We do not try to edit playlist_rows directly. It would be too easy for a bug to get us
+    out of sync with the database, and if that wasn't immediately apparent then debugging it
+    would be hard.
 
-    refresh_row() will populate one row of playlist_rows from the
-    database
-
-    refresh_data() will repopulate all of playlist_rows from the
-    database
+    refresh_row() will populate one row of playlist_rows from the database
+    refresh_data() will repopulate all of playlist_rows from the database
     """
 
     def __init__(
@@ -106,13 +102,10 @@ class PlaylistModel(QAbstractTableModel):
 
         self.playlist_rows: dict[int, PlaylistRowData] = {}
 
-        with Session() as session:
-            self.refresh_data(session)
+        self.refresh_data()
 
     def __repr__(self) -> str:
-        return (
-            f"<PlaylistModel: playlist_id={self.playlist_id}, {self.rowCount()} rows>"
-        )
+        return f"<PlaylistModel: playlist_id={self.playlist_id}, {self.rowCount() rows>"
 
     def background_role(self, row: int, column: int, prd: PlaylistRowData) -> QBrush:
         """Return background setting"""
@@ -297,55 +290,72 @@ class PlaylistModel(QAbstractTableModel):
 
         return QVariant()
 
-    def insert_header_row(self, row_number: Optional[int], text: str) -> None:
+    def insert_header_row(self, row_number: Optional[int], text: str) -> Optional[int]:
         """
         Insert a header row. Return row number or None if insertion failed.
         """
 
         with Session() as session:
-            plr = self._insert_row(session, row_number)
-            # Update the PlaylistRows object
-            plr.note = text
-            # Repopulate self.playlist_rows
-            self.refresh_data(session)
-            # Update the display from the new row onwards
-            self.invalidate_rows(list(range(plr.plr_rownum, len(self.playlist_rows))))
+            prd = self._insert_row(session, row_number)
+            # Update playlist_rows
+            prd.note = text
+            # Get row from db and update
+            plr = session.get(PlaylistRows, prd.plrid)
+            if plr:
+                plr.note = text
+                self.refresh_row(session, plr.plr_rownum)
+                self.invalidate_row(plr.plr_rownum)
+                return plr.plr_rownum
+
+        return None
 
     def _insert_row(
         self, session: scoped_session, row_number: Optional[int]
-    ) -> PlaylistRows:
+    ) -> PlaylistRowData:
         """
-        Insert a row in the database.
+        Make space for a row at row_number. If row_number is greater
+        than length of list plus 1, or if row number is -1, put row at
+        end of list.
 
-        If row_number is greater than length of list plus 1, or if row
-        number is None, put row at end of list.
-
-        Move existing rows to make space if ncessary.
-
-        Return the new PlaylistRows object.
+        Return the new PlaylistData structure
         """
 
+        modified_rows: List[int] = []
+
         if row_number is None or row_number > len(self.playlist_rows):
-            # We are adding to the end of the list so we can optimise
             new_row_number = len(self.playlist_rows)
-            return PlaylistRows(session, self.playlist_id, new_row_number)
         elif row_number < 0:
             raise ValueError(
-                f"playlistmodel._insert_row, invalid row number ({row_number})"
+                f"playlistmodel.insert_row, invalid row number ({row_number})"
             )
         else:
             new_row_number = row_number
+        modified_rows.append(new_row_number)
 
         # Move rows below new row down
-        stmt = (
-            update(PlaylistRows)
-            .where(PlaylistRows.plr_rownum >= new_row_number)
-            .values({PlaylistRows.plr_rownum: PlaylistRows.plr_rownum + 1})
-        )
-        session.execute(stmt)
+        for i in reversed(range(new_row_number, len(self.playlist_rows))):
+            self.playlist_rows[i + 1] = self.playlist_rows[i]
+            self.playlist_rows[i + 1].plr_rownum += 1
+            modified_rows.append(i + 1)
+        # If we are not adding to the end of the list, we need to clear
+        # out the existing recored at new_row_number (which we have
+        # already copied to its new location)
+        if new_row_number in self.playlist_rows:
+            del self.playlist_rows[new_row_number]
 
-        # Insert the new row and return it
-        return PlaylistRows(session, self.playlist_id, new_row_number)
+        *** Problem here is that we haven't yet updated the database so when we insert a new row
+        with the PlaylistRows.__init__ call below, we'll get a duplicate. How best to keep
+        playlist_rows in step with database?
+
+        # Insert new row, possibly replace old row
+        plr = PlaylistRows(
+            session=session, playlist_id=self.playlist_id, row_number=new_row_number
+        )
+        prd = PlaylistRowData(plr)
+        # Add row to playlist_rows
+        self.playlist_rows[new_row_number] = prd
+
+        return prd
 
     def invalidate_row(self, modified_row: int) -> None:
         """
@@ -364,72 +374,65 @@ class PlaylistModel(QAbstractTableModel):
         for modified_row in modified_rows:
             self.invalidate_row(modified_row)
 
-    def move_rows(self, from_rows: List[int], to_row_number: int) -> None:
+    def move_rows(self, from_rows: List[int], to_row: int) -> None:
         """
         Move the playlist rows given to to_row and below.
         """
 
-        # Build a {current_row_number: new_row_number} dictionary
-        row_map: dict[int, int] = {}
+        new_playlist_rows: dict[int, PlaylistRowData] = {}
 
-        # Put the from_row row numbers into the row_map. Ultimately the
-        # total number of elements in the playlist doesn't change, so
-        # check that adding the moved rows starting at to_row won't
-        # overshoot the end of the playlist.
-        if to_row_number + len(from_rows) > len(self.playlist_rows):
+        # Move the from_row records from the playlist_rows dict to the
+        # new_playlist_rows dict. The total number of elements in the
+        # playlist doesn't change, so check that adding the moved rows
+        # starting at to_row won't overshoot the end of the playlist.
+        if to_row + len(from_rows) > len(self.playlist_rows):
             next_to_row = len(self.playlist_rows) - len(from_rows)
         else:
-            next_to_row = to_row_number
+            next_to_row = to_row
 
-        for from_row, to_row in zip(
-            from_rows, range(next_to_row, next_to_row + len(from_rows))
-        ):
-            row_map[from_row] = to_row
-        # Move the remaining rows to the row_map. We want to fill it
-        # before (if there are gaps) and after (likewise) the rows that
-        # are moving.
-        # This iterates old_row and new_row simultaneously.
-        for old_row, new_row in zip(
-            [x for x in self.playlist_rows.keys() if x not in from_rows],
-            [y for y in range(len(self.playlist_rows)) if y not in row_map.values()],
-        ):
-            # Optimise: only add to map if there is a change
-            row_map[old_row] = new_row
+        for from_row in from_rows:
+            new_playlist_rows[next_to_row] = self.playlist_rows[from_row]
+            del self.playlist_rows[from_row]
+            next_to_row += 1
 
-        # For SQLAlchemy, build a list of dictionaries that map plrid to
-        # new row number:
-        sqla_map: List[dict[str, int]] = []
-        for oldrow, newrow in row_map.items():
-            plrid = self.playlist_rows[oldrow].plrid
-            sqla_map.append({"plrid": plrid, "plr_rownum": newrow})
+        # Move the remaining rows to the gaps in new_playlist_rows
+        new_row = 0
+        for old_row in self.playlist_rows.keys():
+            # Find next gap
+            while new_row in new_playlist_rows:
+                new_row += 1
+            new_playlist_rows[new_row] = self.playlist_rows[old_row]
+            new_row += 1
 
-        # Update database. Ref:
-        # https://docs.sqlalchemy.org/en/20/core/sqlelement.html#sqlalchemy.sql.expression.case
-        stmt = (
-            update(PlaylistRows)
-            .where(
-                PlaylistRows.playlist_id == self.playlist_id,
-                PlaylistRows.id == bindparam("plrid"),
-            )
-            .values(plr_rownum=bindparam("plr_rownum"))
-        )
+        # Make copy of rows live
+        self.playlist_rows = new_playlist_rows
 
+        # Update PlaylistRows table and notify display of rows that
+        # moved
         with Session() as session:
-            session.connection().execute(stmt, sqla_map)
+            for idx in range(len(self.playlist_rows)):
+                if self.playlist_rows[idx].plr_rownum == idx:
+                    continue
+                # Row number in this row is incorred. Fix it in
+                # database:
+                plr = session.get(PlaylistRows, self.playlist_rows[idx].plrid)
+                if not plr:
+                    print(f"\nCan't find plr in playlistmodel:move_rows {idx=}")
+                    continue
+                plr.plr_rownum = idx
+                # Fix in self.playlist_rows
+                self.playlist_rows[idx].plr_rownum = idx
+                # Update display
+                self.invalidate_row(idx)
 
-            # Update playlist_rows
-            self.refresh_data(session)
-
-        # Update display
-        self.invalidate_rows(list(row_map.keys()))
-
-    def refresh_data(self, session: scoped_session):
+    def refresh_data(self):
         """Populate dicts for data calls"""
 
         # Populate self.playlist_rows with playlist data
         self.playlist_rows.clear()
-        for p in PlaylistRows.deep_rows(session, self.playlist_id):
-            self.playlist_rows[p.plr_rownum] = PlaylistRowData(p)
+        with Session() as session:
+            for p in PlaylistRows.deep_rows(session, self.playlist_id):
+                self.playlist_rows[p.plr_rownum] = PlaylistRowData(p)
 
     def refresh_row(self, session, row_number):
         """Populate dict for one row for data calls"""

archive/db_experiments.py

@@ -1,84 +0,0 @@
-#!/usr/bin/env python3
-
-from sqlalchemy import create_engine, String, update, bindparam, case
-from sqlalchemy.orm import (
-    DeclarativeBase,
-    Mapped,
-    mapped_column,
-    sessionmaker,
-    scoped_session,
-)
-from typing import Generator
-from contextlib import contextmanager
-
-db_url = "sqlite:////tmp/rhys.db"
-
-
-class Base(DeclarativeBase):
-    pass
-
-
-class Rhys(Base):
-    __tablename__ = "rhys"
-
-    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
-    ref_number: Mapped[int] = mapped_column()
-    name: Mapped[str] = mapped_column(String(256), index=True)
-
-    def __init__(self, session, ref_number: int, name: str) -> None:
-        self.ref_number = ref_number
-        self.name = name
-        session.add(self)
-        session.flush()
-
-
-@contextmanager
-def Session() -> Generator[scoped_session, None, None]:
-    Session = scoped_session(sessionmaker(bind=engine))
-    yield Session
-    Session.commit()
-    Session.close()
-
-
-engine = create_engine(db_url)
-Base.metadata.create_all(engine)
-
-inital_number_of_records = 10
-
-
-def move_rows(session):
-    new_row = 6
-
-    with Session() as session:
-        # new_record = Rhys(session, new_row, f"new {new_row=}")
-        # Move rows
-
-        stmt = (
-            update(Rhys)
-            .where(Rhys.ref_number > new_row)
-            # .where(Rhys.id.in_(session.query(Rhys.id).order_by(Rhys.id.desc())))
-            .values({Rhys.ref_number: Rhys.ref_number + 1})
-        )
-
-        session.execute(stmt)
-
-
-sqla_map = []
-for k, v in zip(range(11), [0, 1, 2, 3, 4, 7, 8, 10, 5, 6, 9]):
-    sqla_map.append({"oldrow": k, "newrow": v})
-
-# for a, b in sqla_map.items():
-#     print(f"{a} > {b}")
-
-with Session() as session:
-    for a in range(inital_number_of_records):
-        _ = Rhys(session, a, f"record: {a}")
-
-    stmt = update(Rhys).values(
-        ref_number=case(
-            {item['oldrow']: item['newrow'] for item in sqla_map},
-            value=Rhys.ref_number
-        )
-    )
-
-    session.connection().execute(stmt, sqla_map)

test_playlistmodel.py

@@ -14,13 +14,34 @@ def create_model_with_playlist_rows(
     for row in range(rows):
         plr = model._insert_row(session, row)
         newrow = plr.plr_rownum
-        plr.note = str(newrow)
-        model.playlist_rows[newrow] = playlistmodel.PlaylistRowData(plr)
+        model.playlist_rows[newrow].note = str(newrow)
 
     session.commit()
     return model
 
 
+def test_insert_row(monkeypatch, session):
+    monkeypatch.setattr(playlistmodel, "Session", session)
+    # Create a playlist
+    playlist = Playlists(session, "test playlist")
+    # Create a model
+    model = playlistmodel.PlaylistModel(playlist.id, None)
+    assert model.rowCount() == 0
+    model._insert_row(session, 0)
+    assert model.rowCount() == 1
+
+
+def test_insert_high_row(monkeypatch, session):
+    monkeypatch.setattr(playlistmodel, "Session", session)
+    # Create a playlist
+    playlist = Playlists(session, "test playlist")
+    # Create a model
+    model = playlistmodel.PlaylistModel(playlist.id, None)
+    assert model.rowCount() == 0
+    model._insert_row(session, 5)
+    assert model.rowCount() == 1
+
+
 def test_11_row_playlist(monkeypatch, session):
     # Create multirow playlist
     monkeypatch.setattr(playlistmodel, "Session", session)
@@ -166,35 +187,12 @@ def test_insert_header_row_end(monkeypatch, session):
 
     monkeypatch.setattr(playlistmodel, "Session", session)
     note_text = "test text"
-    initial_row_count = 11
 
-    model = create_model_with_playlist_rows(session, initial_row_count)
-    model.insert_header_row(None, note_text)
-    assert model.rowCount() == initial_row_count + 1
-    prd = model.playlist_rows[model.rowCount() - 1]
+    model = create_model_with_playlist_rows(session, 11)
+    row_number = model.insert_header_row(None, note_text)
+    session.flush()
+    assert model.rowCount() == row_number + 1
+    prd = model.playlist_rows[row_number]
     # Test against edit_role because display_role for headers is
     # handled differently (sets up row span)
-    assert (
-        model.edit_role(model.rowCount(), playlistmodel.Col.NOTE.value, prd)
-        == note_text
-    )
-
-
-def test_insert_header_row_middle(monkeypatch, session):
-    # insert header row in middle of playlist
-
-    monkeypatch.setattr(playlistmodel, "Session", session)
-    note_text = "test text"
-    initial_row_count = 11
-    insert_row = 6
-
-    model = create_model_with_playlist_rows(session, initial_row_count)
-    model.insert_header_row(insert_row, note_text)
-    assert model.rowCount() == initial_row_count + 1
-    prd = model.playlist_rows[insert_row]
-    # Test against edit_role because display_role for headers is
-    # handled differently (sets up row span)
-    assert (
-        model.edit_role(model.rowCount(), playlistmodel.Col.NOTE.value, prd)
-        == note_text
-    )
+    assert model.edit_role(row_number, playlistmodel.Col.NOTE.value, prd) == note_text