refactor: enhance row validation in activity and tasks APIs for better type safety

backend/app/api/activity.py

@@ -103,16 +103,39 @@ def _coerce_task_comment_rows(
 ) -> list[tuple[ActivityEvent, Task, Board, Agent | None]]:
     rows: list[tuple[ActivityEvent, Task, Board, Agent | None]] = []
     for item in items:
+        first: Any
+        second: Any
+        third: Any
+        fourth: Any
+
+        if isinstance(item, tuple):
+            if len(item) != TASK_COMMENT_ROW_LEN:
+                msg = "Expected (ActivityEvent, Task, Board, Agent | None) rows"
+                raise TypeError(msg)
+            first, second, third, fourth = item
+        else:
+            try:
+                row_len = len(item)
+                first = item[0]
+                second = item[1]
+                third = item[2]
+                fourth = item[3]
+            except (IndexError, KeyError, TypeError):
+                msg = "Expected (ActivityEvent, Task, Board, Agent | None) rows"
+                raise TypeError(msg) from None
+            if row_len != TASK_COMMENT_ROW_LEN:
+                msg = "Expected (ActivityEvent, Task, Board, Agent | None) rows"
+                raise TypeError(msg)
+
         if (
-            isinstance(item, tuple)
-            and len(item) == TASK_COMMENT_ROW_LEN
-            and isinstance(item[0], ActivityEvent)
-            and isinstance(item[1], Task)
-            and isinstance(item[2], Board)
-            and (isinstance(item[3], Agent) or item[3] is None)
+            isinstance(first, ActivityEvent)
+            and isinstance(second, Task)
+            and isinstance(third, Board)
+            and (isinstance(fourth, Agent) or fourth is None)
         ):
-            rows.append((item[0], item[1], item[2], item[3]))
+            rows.append((first, second, third, fourth))
             continue
+
         msg = "Expected (ActivityEvent, Task, Board, Agent | None) rows"
         raise TypeError(msg)
     return rows

backend/app/api/tasks.py

@@ -164,14 +164,32 @@ def _coerce_task_event_rows(
 ) -> list[tuple[ActivityEvent, Task | None]]:
     rows: list[tuple[ActivityEvent, Task | None]] = []
     for item in items:
-        if (
-            isinstance(item, tuple)
-            and len(item) == TASK_EVENT_ROW_LEN
-            and isinstance(item[0], ActivityEvent)
-            and (isinstance(item[1], Task) or item[1] is None)
+        first: object
+        second: object
+
+        if isinstance(item, tuple):
+            if len(item) != TASK_EVENT_ROW_LEN:
+                msg = "Expected (ActivityEvent, Task | None) rows"
+                raise TypeError(msg)
+            first, second = item
+        else:
+            try:
+                row_len = len(item)  # type: ignore[arg-type]
+                first = item[0]  # type: ignore[index]
+                second = item[1]  # type: ignore[index]
+            except (IndexError, KeyError, TypeError):
+                msg = "Expected (ActivityEvent, Task | None) rows"
+                raise TypeError(msg) from None
+            if row_len != TASK_EVENT_ROW_LEN:
+                msg = "Expected (ActivityEvent, Task | None) rows"
+                raise TypeError(msg)
+
+        if isinstance(first, ActivityEvent) and (
+            isinstance(second, Task) or second is None
         ):
-            rows.append((item[0], item[1]))
+            rows.append((first, second))
             continue
+
         msg = "Expected (ActivityEvent, Task | None) rows"
         raise TypeError(msg)
     return rows

backend/tests/test_activity_api_rows.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from uuid import uuid4
+
+import pytest
+
+from app.api.activity import _coerce_task_comment_rows
+from app.models.activity_events import ActivityEvent
+from app.models.agents import Agent
+from app.models.boards import Board
+from app.models.tasks import Task
+
+
+@dataclass
+class _FakeSqlRow4:
+    first: object
+    second: object
+    third: object
+    fourth: object
+
+    def __len__(self) -> int:
+        return 4
+
+    def __getitem__(self, index: int) -> object:
+        if index == 0:
+            return self.first
+        if index == 1:
+            return self.second
+        if index == 2:
+            return self.third
+        if index == 3:
+            return self.fourth
+        raise IndexError(index)
+
+
+def _make_event() -> ActivityEvent:
+    return ActivityEvent(event_type="task.comment", message="hello")
+
+
+def _make_board() -> Board:
+    return Board(
+        organization_id=uuid4(),
+        name="B",
+        slug="b",
+    )
+
+
+def _make_task(board_id) -> Task:
+    return Task(board_id=board_id, title="T")
+
+
+def _make_agent(board_id) -> Agent:
+    return Agent(
+        board_id=board_id,
+        gateway_id=uuid4(),
+        name="A",
+    )
+
+
+def test_coerce_task_comment_rows_accepts_plain_tuple():
+    board = _make_board()
+    task = _make_task(board.id)
+    event = _make_event()
+    agent = _make_agent(board.id)
+
+    rows = _coerce_task_comment_rows([(event, task, board, agent)])
+    assert rows == [(event, task, board, agent)]
+
+
+def test_coerce_task_comment_rows_accepts_row_like_values():
+    board = _make_board()
+    task = _make_task(board.id)
+    event = _make_event()
+    row = _FakeSqlRow4(event, task, board, None)
+
+    rows = _coerce_task_comment_rows([row])
+    assert rows == [(event, task, board, None)]
+
+
+def test_coerce_task_comment_rows_rejects_invalid_values():
+    board = _make_board()
+    task = _make_task(board.id)
+
+    with pytest.raises(
+        TypeError,
+        match="Expected \\(ActivityEvent, Task, Board, Agent \\| None\\) rows",
+    ):
+        _coerce_task_comment_rows([(uuid4(), task, board, None)])

backend/tests/test_tasks_api_rows.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from uuid import uuid4
+
+import pytest
+
+from app.api.tasks import _coerce_task_event_rows
+from app.models.activity_events import ActivityEvent
+from app.models.tasks import Task
+
+
+@dataclass
+class _FakeSqlRow:
+    first: object
+    second: object
+
+    def __len__(self) -> int:
+        return 2
+
+    def __getitem__(self, index: int) -> object:
+        if index == 0:
+            return self.first
+        if index == 1:
+            return self.second
+        raise IndexError(index)
+
+
+def _make_event() -> ActivityEvent:
+    return ActivityEvent(event_type="task.updated")
+
+
+def _make_task() -> Task:
+    return Task(board_id=uuid4(), title="T")
+
+
+def test_coerce_task_event_rows_accepts_plain_tuple():
+    event = _make_event()
+    task = _make_task()
+    rows = _coerce_task_event_rows([(event, task)])
+    assert rows == [(event, task)]
+
+
+def test_coerce_task_event_rows_accepts_row_like_values():
+    event = _make_event()
+    task = _make_task()
+    rows = _coerce_task_event_rows([_FakeSqlRow(event, task)])
+    assert rows == [(event, task)]
+
+
+def test_coerce_task_event_rows_rejects_invalid_values():
+    with pytest.raises(TypeError, match="Expected \\(ActivityEvent, Task \\| None\\) rows"):
+        _coerce_task_event_rows([("bad", "row")])