Complete SQLAlchemy 2.0 migration for test suite and models (#756)

This completes the SQLAlchemy 2.0 migration started in commits 545c4e2 and ea54f49.

Changes:
- Migrate 12 test files from deprecated .query() API to select() API
  - Replace session.query(Model).filter() with session.scalars(select(Model).where())
  - Replace session.query().count() with session.scalar(select(func.count()).select_from())
  - Update composite key .get() from dict to tuple format
- Add explicit String column lengths to model fields
  - models/cue.py: colour (16), ident (50)
  - models/mics.py: name (100), description (500)

All tests pass with no SQLAlchemy deprecation warnings.

Files modified:
- 12 test files under test/
- 2 model files: models/cue.py, models/mics.py

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: Tim020

server/models/cue.py

@@ -16,7 +16,7 @@ class CueType(db.Model):
 
     prefix: Mapped[str | None] = mapped_column(String(5))
     description: Mapped[str | None] = mapped_column(String(100))
-    colour: Mapped[str | None] = mapped_column(String())
+    colour: Mapped[str | None] = mapped_column(String(16))
 
     cues: Mapped[List["Cue"]] = relationship(
         back_populates="cue_type", cascade="all, delete-orphan"
@@ -28,7 +28,7 @@ class Cue(db.Model):
 
     id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
     cue_type_id: Mapped[int | None] = mapped_column(ForeignKey("cuetypes.id"))
-    ident: Mapped[str | None] = mapped_column(String())
+    ident: Mapped[str | None] = mapped_column(String(50))
 
     cue_type: Mapped["CueType"] = relationship(
         foreign_keys=[cue_type_id], back_populates="cues"

server/models/mics.py

@@ -16,8 +16,8 @@ class Microphone(db.Model):
     id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
     show_id: Mapped[int | None] = mapped_column(ForeignKey("shows.id"))
 
-    name: Mapped[str | None] = mapped_column(String)
-    description: Mapped[str | None] = mapped_column(String)
+    name: Mapped[str | None] = mapped_column(String(100))
+    description: Mapped[str | None] = mapped_column(String(500))
 
     allocations: Mapped[List["MicrophoneAllocation"]] = relationship(
         cascade="all, delete-orphan", back_populates="microphone"

server/test/api/show/script/test_config.py

@@ -1,4 +1,5 @@
 import tornado.escape
+from sqlalchemy import select
 
 from models.session import Session
 from test.utils import DigiScriptTestCase
@@ -11,7 +12,7 @@ def test_get_script_config_no_editors(self):
         """Test GET /api/v1/show/script/config with no editors.
 
         This tests the query at line 13:
-        session.query(Session).filter(Session.is_editor).all()
+        session.scalars(select(Session).where(Session.is_editor)).all()
         """
         response = self.fetch("/api/v1/show/script/config")
         self.assertEqual(200, response.code)

server/test/api/show/script/test_revisions.py

@@ -1,4 +1,5 @@
 import tornado.escape
+from sqlalchemy import func, select
 
 from models.script import Script, ScriptRevision
 from models.show import Show
@@ -55,7 +56,7 @@ def test_get_revisions(self):
         """Test GET /api/v1/show/script/revisions.
 
         This tests the query at line 36:
-        session.query(Script).filter(Script.show_id == show.id).first()
+        session.scalars(select(Script).where(Script.show_id == show.id)).first()
         """
         response = self.fetch("/api/v1/show/script/revisions")
         self.assertEqual(200, response.code)
@@ -111,7 +112,7 @@ def test_get_current_revision(self):
         """Test GET /api/v1/show/script/revisions/current.
 
         This tests the query at line 255:
-        session.query(Script).filter(Script.show_id == show.id).first()
+        session.scalars(select(Script).where(Script.show_id == show.id)).first()
         """
         response = self.fetch("/api/v1/show/script/revisions/current")
         self.assertEqual(200, response.code)
@@ -138,9 +139,10 @@ class TestScriptRevisionCreate(DigiScriptTestCase):
     """Test POST /api/v1/show/script/revisions endpoint.
 
     This tests the func.max() query at lines 89-93 in controllers/api/show/script/revisions.py:
-    session.query(func.max(ScriptRevision.revision))
-        .filter(ScriptRevision.script_id == script.id)
-        .one()[0]
+    session.scalar(
+        select(func.max(ScriptRevision.revision))
+        .where(ScriptRevision.script_id == script.id)
+    )
 
     The query is used to determine the next revision number when creating a new revision.
     """
@@ -215,9 +217,10 @@ class TestScriptRevisionDelete(DigiScriptTestCase):
     """Test DELETE /api/v1/show/script/revisions endpoint.
 
     This tests the "find revision 1" query at lines 206-213 in controllers/api/show/script/revisions.py:
-    session.query(ScriptRevision)
-        .filter(ScriptRevision.script_id == script.id, ScriptRevision.revision == 1)
-        .one()
+    session.scalars(
+        select(ScriptRevision)
+        .where(ScriptRevision.script_id == script.id, ScriptRevision.revision == 1)
+    ).one()
 
     The query is used as a fallback when deleting the current revision and it has no previous_revision_id.
     """

server/test/api/show/script/test_script.py

@@ -81,9 +81,11 @@ def test_get_script_page_with_no_lines(self):
         """Test GET /api/v1/show/script?page=1 with empty script.
 
         This tests the query at line 59-66:
-        session.query(ScriptLineRevisionAssociation).filter(
-            ScriptLineRevisionAssociation.revision_id == revision.id,
-            ScriptLineRevisionAssociation.line.has(page=page),
+        session.scalars(
+            select(ScriptLineRevisionAssociation).where(
+                ScriptLineRevisionAssociation.revision_id == revision.id,
+                ScriptLineRevisionAssociation.line.has(page=page),
+            )
         ).all()
         """
         response = self.fetch("/api/v1/show/script?page=1")
@@ -191,11 +193,11 @@ def test_composite_key_get_pattern_used_throughout(self):
             session.commit()
             line_id = line.id
 
-        # Test the composite dict .get() pattern (will be migrated to tuple)
+        # Test the composite key .get() pattern (SQLAlchemy 2.0 uses tuple)
         with self._app.get_db().sessionmaker() as session:
-            # This is the legacy pattern used throughout the controller
-            found_assoc = session.query(ScriptLineRevisionAssociation).get(
-                {"revision_id": self.revision_id, "line_id": line_id}
+            # SQLAlchemy 2.0 pattern for composite primary keys
+            found_assoc = session.get(
+                ScriptLineRevisionAssociation, (self.revision_id, line_id)
             )
             self.assertIsNotNone(found_assoc)
             self.assertEqual(self.revision_id, found_assoc.revision_id)
@@ -230,7 +232,7 @@ def test_get_compiled_script(self):
         """Test GET /api/v1/show/script/compiled.
 
         This tests the query at line 666:
-        session.query(Script).filter(Script.show_id == show.id).first()
+        session.scalars(select(Script).where(Script.show_id == show.id)).first()
         """
         response = self.fetch("/api/v1/show/script/compiled")
         # Empty script won't have compiled form yet, so expect 404
@@ -322,17 +324,15 @@ def test_script_cuts_query_patterns(self):
         # Test the query patterns used in the controller
         with self._app.get_db().sessionmaker() as session:
             # Pattern 1: Get script by show_id (line 741)
-            script = (
-                session.query(Script).filter(Script.show_id == self.show_id).first()
-            )
+            script = session.scalars(
+                select(Script).where(Script.show_id == self.show_id)
+            ).first()
             self.assertIsNotNone(script)
 
             # Pattern 2: Get all cuts for revision (lines 766-770)
-            cuts = (
-                session.query(ScriptCuts)
-                .filter(ScriptCuts.revision_id == self.revision_id)
-                .all()
-            )
+            cuts = session.scalars(
+                select(ScriptCuts).where(ScriptCuts.revision_id == self.revision_id)
+            ).all()
             self.assertEqual(1, len(cuts))
             self.assertEqual(self.line_part_id, cuts[0].line_part_id)
 

server/test/api/show/script/test_stage_direction_styles.py

@@ -1,4 +1,5 @@
 import tornado.escape
+from sqlalchemy import select
 
 from models.script import Script, ScriptRevision, StageDirectionStyle
 from models.show import Show
@@ -55,7 +56,7 @@ def test_get_stage_direction_styles(self):
         """Test GET /api/v1/show/script/stage_direction_styles.
 
         This tests the query at line 24:
-        session.query(Script).filter(Script.show_id == show.id).first()
+        session.scalars(select(Script).where(Script.show_id == show.id)).first()
         """
         response = self.fetch("/api/v1/show/script/stage_direction_styles")
         self.assertEqual(200, response.code)

server/test/api/show/test_microphones.py

@@ -1,4 +1,5 @@
 import tornado.escape
+from sqlalchemy import select
 
 from models.mics import Microphone
 from models.show import Show
@@ -24,7 +25,7 @@ def test_get_microphones_empty(self):
         """Test GET /api/v1/show/microphones with no microphones.
 
         This tests the query at line 25-28 in controllers/api/show/microphones.py:
-        session.query(Microphone).filter(Microphone.show_id == show.id).all()
+        session.scalars(select(Microphone).where(Microphone.show_id == show.id)).all()
         """
         response = self.fetch("/api/v1/show/microphones")
         self.assertEqual(200, response.code)
@@ -70,7 +71,7 @@ def test_get_allocations_empty(self):
         """Test GET /api/v1/show/microphones/allocations with no microphones.
 
         This tests the query at line 192-195 in microphones.py:
-        session.query(Microphone).filter(Microphone.show_id == show.id).all()
+        session.scalars(select(Microphone).where(Microphone.show_id == show.id)).all()
         """
         response = self.fetch("/api/v1/show/microphones/allocations")
         self.assertEqual(200, response.code)

server/test/api/show/test_sessions.py

@@ -1,4 +1,5 @@
 import tornado.escape
+from sqlalchemy import select
 
 from models.session import ShowSession
 from models.show import Show
@@ -24,7 +25,7 @@ def test_get_sessions_empty(self):
         """Test GET /api/v1/show/sessions with no sessions.
 
         This tests the query at line 26-29 in controllers/api/show/sessions.py:
-        session.query(ShowSession).filter(ShowSession.show_id == show.id).all()
+        session.scalars(select(ShowSession).where(ShowSession.show_id == show.id)).all()
         """
         response = self.fetch("/api/v1/show/sessions")
         self.assertEqual(200, response.code)

server/test/api/test_auth.py

@@ -1,4 +1,5 @@
 from tornado import escape
+from sqlalchemy import select
 
 from models.user import User
 from test.utils import DigiScriptTestCase
@@ -53,7 +54,7 @@ def test_create_user_duplicate_username(self):
         """Test POST /api/v1/auth/create with duplicate username.
 
         This specifically tests the query at lines 48-49 in controllers/api/auth.py:
-        session.query(User).filter(User.username == username).first()
+        session.scalars(select(User).where(User.username == username)).first()
 
         When a user with the same username already exists, the query should return
         that user and the endpoint should return a 400 error.
@@ -391,7 +392,7 @@ def test_delete_user(self):
         """Test DELETE /api/v1/auth/delete endpoint.
 
         This tests the queries at lines 116-118 and 129-131 in controllers/api/auth.py:
-        session.query(Session).filter(Session.user_id == user_to_delete.id).all()
+        session.scalars(select(Session).where(Session.user_id == user_to_delete.id)).all()
 
         When deleting a user, the endpoint queries for all active WebSocket sessions
         belonging to that user. In this test, there are no active sessions, so the
@@ -440,7 +441,9 @@ def test_delete_user(self):
         with self._app.get_db().sessionmaker() as session:
             from models.user import User
 
-            user = session.query(User).filter(User.username == "userToDelete").first()
+            user = session.scalars(
+                select(User).where(User.username == "userToDelete")
+            ).first()
             user_id = user.id
 
         # Delete the user - the session query will return empty list
@@ -460,14 +463,16 @@ def test_delete_user(self):
         with self._app.get_db().sessionmaker() as session:
             from models.user import User
 
-            deleted_user = session.query(User).filter(User.id == user_id).first()
+            deleted_user = session.scalars(
+                select(User).where(User.id == user_id)
+            ).first()
             self.assertIsNone(deleted_user)
 
     def test_get_users(self):
         """Test GET /api/v1/auth/users endpoint.
 
         This tests the query at line 266 in controllers/api/auth.py:
-        session.query(User).all()
+        session.scalars(select(User)).all()
 
         The endpoint retrieves all users in the system.
         """

server/test/api/test_rbac.py

@@ -1,4 +1,5 @@
 import tornado.escape
+from sqlalchemy import select
 
 from models.script import Script
 from models.show import Show
@@ -12,7 +13,9 @@ def test_delete_actor_removes_rbac_assignments(self):
         """Test that delete_actor removes all RBAC assignments for a user.
 
         This tests the query at line 247 in rbac/rbac_db.py:
-        session.query(self._mappings[table_name]).filter_by(**cols).all()
+        model = self._mappings[table_name]
+        conditions = [getattr(model, k) == v for k, v in cols.items()]
+        session.scalars(select(model).where(*conditions)).all()
 
         We create a user, assign them roles for resources, then call delete_actor
         and verify the RBAC table entries were deleted.
@@ -62,9 +65,9 @@ def test_get_objects_for_resource(self):
         """Test GET /api/v1/rbac/user/objects endpoint.
 
         This tests the query at lines 391-396 in rbac/rbac_db.py:
-        session.query(self._db.get_mapper_for_table(table.fullname))
-            .filter_by(**cols)
-            .all()
+        model = self._db.get_mapper_for_table(table.fullname)
+        conditions = [getattr(model, k) == v for k, v in cols.items()]
+        session.scalars(select(model).where(*conditions)).all()
 
         The query walks the database relationship graph to find all instances
         of a resource type that are related to the current show.