refactor: split run read queries

rohans0509 · rohans0509 · commit 6be83fa4e44e · 2026-07-06T17:07:05.000-07:00
diff --git a/src/codealmanac/services/runs/queries.py b/src/codealmanac/services/runs/queries.py
@@ -0,0 +1,120 @@
+from codealmanac.database.sqlite import SQLiteConnection, SQLiteRow
+from codealmanac.services.runs.models import (
+    QueuedRun,
+    RunRecord,
+    RunSpec,
+    RunStatus,
+)
+
+
+def list_run_records(
+    connection: SQLiteConnection,
+    limit: int | None,
+    repository_id: str | None = None,
+) -> tuple[RunRecord, ...]:
+    query = """
+        SELECT record_json
+        FROM runs
+    """
+    parameters: tuple[object, ...]
+    if repository_id is None:
+        parameters = ()
+    else:
+        query = f"{query} WHERE repository_id = ?"
+        parameters = (repository_id,)
+    query = f"""
+        {query}
+        ORDER BY created_at DESC, run_id DESC
+    """
+    if limit is not None:
+        query = f"{query} LIMIT ?"
+        parameters = (*parameters, limit)
+    rows = connection.execute(query, parameters).fetchall()
+    return tuple(run_record_from_row(row) for row in rows)
+
+
+def read_run_record(
+    connection: SQLiteConnection,
+    run_id: str,
+) -> RunRecord | None:
+    row = connection.execute(
+        "SELECT record_json FROM runs WHERE run_id = ?",
+        (run_id,),
+    ).fetchone()
+    if row is None:
+        return None
+    return run_record_from_row(row)
+
+
+def read_run_spec(
+    connection: SQLiteConnection,
+    run_id: str,
+) -> RunSpec | None:
+    result = read_run_with_spec(connection, run_id)
+    if result is None:
+        return None
+    return result[1]
+
+
+def read_run_with_spec(
+    connection: SQLiteConnection,
+    run_id: str,
+) -> tuple[RunRecord, RunSpec | None] | None:
+    row = connection.execute(
+        "SELECT record_json, spec_json FROM runs WHERE run_id = ?",
+        (run_id,),
+    ).fetchone()
+    if row is None:
+        return None
+    spec = None
+    if row["spec_json"] is not None:
+        spec = RunSpec.model_validate_json(row["spec_json"])
+    return (run_record_from_row(row), spec)
+
+
+def next_queued_run(connection: SQLiteConnection) -> QueuedRun | None:
+    row = connection.execute(
+        """
+        SELECT record_json, spec_json
+        FROM runs
+        WHERE status = ? AND spec_json IS NOT NULL
+        ORDER BY created_at ASC, run_id ASC
+        LIMIT 1
+        """,
+        (RunStatus.QUEUED.value,),
+    ).fetchone()
+    if row is None:
+        return None
+    return QueuedRun(
+        record=run_record_from_row(row),
+        spec=RunSpec.model_validate_json(row["spec_json"]),
+    )
+
+
+def count_queued_runs_before(
+    connection: SQLiteConnection,
+    record: RunRecord,
+) -> int:
+    row = connection.execute(
+        """
+        SELECT COUNT(*)
+        FROM runs
+        WHERE status = ?
+          AND spec_json IS NOT NULL
+          AND (
+            created_at < ?
+            OR (created_at = ? AND run_id < ?)
+          )
+        """,
+        (
+            RunStatus.QUEUED.value,
+            record.created_at.isoformat(),
+            record.created_at.isoformat(),
+            record.run_id,
+        ),
+    ).fetchone()
+    return int(row[0])
+
+
+def run_record_from_row(row: SQLiteRow) -> RunRecord:
+    return RunRecord.model_validate_json(row["record_json"])
diff --git a/src/codealmanac/services/runs/store.py b/src/codealmanac/services/runs/store.py
@@ -22,6 +22,13 @@
     RunSpec,
     RunStatus,
 )
+from codealmanac.services.runs.queries import (
+    count_queued_runs_before,
+    list_run_records,
+    next_queued_run,
+    read_run_record,
+    read_run_with_spec,
+)
 from codealmanac.services.runs.tables import RUN_TABLES
 from codealmanac.services.runs.transitions import (
     attach_harness_transcript,
@@ -73,89 +80,31 @@ def list(
         limit: int | None,
         repository_id: str | None = None,
     ) -> tuple[RunRecord, ...]:
-        query = """
-            SELECT record_json
-            FROM runs
-        """
-        parameters: tuple[object, ...]
-        if repository_id is None:
-            parameters = ()
-        else:
-            query = f"{query} WHERE repository_id = ?"
-            parameters = (repository_id,)
-        query = f"""
-            {query}
-            ORDER BY created_at DESC, run_id DESC
-        """
-        if limit is not None:
-            query = f"{query} LIMIT ?"
-            parameters = (*parameters, limit)
         with self.connect() as connection:
-            rows = connection.execute(query, parameters).fetchall()
-        return tuple(run_record_from_json(row["record_json"]) for row in rows)
+            return list_run_records(connection, limit, repository_id)
 
     def read(self, run_id: str) -> RunRecord:
         with self.connect() as connection:
-            row = connection.execute(
-                "SELECT record_json FROM runs WHERE run_id = ?",
-                (run_id,),
-            ).fetchone()
-        if row is None:
+            record = read_run_record(connection, run_id)
+        if record is None:
             raise NotFoundError("run", run_id)
-        return run_record_from_json(row["record_json"])
+        return record
 
     def read_spec(self, run_id: str) -> RunSpec | None:
         with self.connect() as connection:
-            row = connection.execute(
-                "SELECT spec_json FROM runs WHERE run_id = ?",
-                (run_id,),
-            ).fetchone()
-        if row is None:
+            result = read_run_with_spec(connection, run_id)
+        if result is None:
             raise NotFoundError("run", run_id)
-        if row["spec_json"] is None:
-            return None
-        return RunSpec.model_validate_json(row["spec_json"])
+        _, spec = result
+        return spec
 
     def next_queued(self) -> QueuedRun | None:
         with self.connect() as connection:
-            row = connection.execute(
-                """
-                SELECT record_json, spec_json
-                FROM runs
-                WHERE status = ? AND spec_json IS NOT NULL
-                ORDER BY created_at ASC, run_id ASC
-                LIMIT 1
-                """,
-                (RunStatus.QUEUED.value,),
-            ).fetchone()
-        if row is None:
-            return None
-        return QueuedRun(
-            record=run_record_from_json(row["record_json"]),
-            spec=RunSpec.model_validate_json(row["spec_json"]),
-        )
+            return next_queued_run(connection)
 
     def queued_before(self, record: RunRecord) -> int:
         with self.connect() as connection:
-            row = connection.execute(
-                """
-                SELECT COUNT(*)
-                FROM runs
-                WHERE status = ?
-                  AND spec_json IS NOT NULL
-                  AND (
-                    created_at < ?
-                    OR (created_at = ? AND run_id < ?)
-                  )
-                """,
-                (
-                    RunStatus.QUEUED.value,
-                    record.created_at.isoformat(),
-                    record.created_at.isoformat(),
-                    record.run_id,
-                ),
-            ).fetchone()
-        return int(row[0])
+            return count_queued_runs_before(connection, record)
 
     def acquire_worker_lock(
         self,
@@ -280,7 +229,3 @@ def connect(self):
         connection.executescript(RUN_TABLES)
         connection.commit()
         return connection
-
-
-def run_record_from_json(value: str) -> RunRecord:
-    return RunRecord.model_validate_json(value)
diff --git a/tests/test_architecture.py b/tests/test_architecture.py
@@ -1817,6 +1817,7 @@ def test_run_persistence_stays_split_by_responsibility():
     events_text = (runs_root / "events.py").read_text(encoding="utf-8")
     factory_text = (runs_root / "factory.py").read_text(encoding="utf-8")
     locks_text = (runs_root / "locks.py").read_text(encoding="utf-8")
+    queries_text = (runs_root / "queries.py").read_text(encoding="utf-8")
     service_text = (runs_root / "service.py").read_text(encoding="utf-8")
     streaming_text = (runs_root / "streaming.py").read_text(encoding="utf-8")
     tables_text = (runs_root / "tables.py").read_text(encoding="utf-8")
@@ -1839,6 +1840,7 @@ def test_run_persistence_stays_split_by_responsibility():
         "events.py",
         "factory.py",
         "locks.py",
+        "queries.py",
         "streaming.py",
         "tables.py",
         "transitions.py",
@@ -1851,6 +1853,9 @@ def test_run_persistence_stays_split_by_responsibility():
     assert "def new_run_record(" in factory_text
     assert "uuid4" in factory_text
     assert "strftime" in factory_text
+    assert "def list_run_records(" in queries_text
+    assert "def read_run_record(" in queries_text
+    assert "def next_queued_run(" in queries_text
     assert "def start_run(" in transitions_text
     assert "def finish_run(" in transitions_text
     assert "def cancel_run(" in transitions_text
