Add detections_count to sequence and alert responses (#559)

* feat(schemas): add detection_count to SequenceRead schema

* feat(endpoints): add helper function to get sequence count

* feat(endpoints): add sequence count to alerts endpoints

* feat(endpoints): add sequence count to sequences endpoint

* Fix local CI ordering and ruff issues

* refactor(services): move detection-count helper to services/

Per review feedback on PR #559: get_detection_counts_by_sequence_ids
is an aggregation query, not a CRUD/route helper, so it belongs in
services/ where it can be reused outside of endpoint modules.

* refactor(alerts): align _serialize_sequence signature across endpoints

Per review feedback on PR #559: _serialize_sequence in alerts.py took
the full Dict[int, int] of counts and did the lookup inside. Match the
sequences.py version (scalar int param, caller does the lookup) so the
helper has one shape repo-wide and only does serialization.

* refactor: drop redundant int() casts on model id attributes

Per review feedback on PR #559: Sequence.id and Alert.id are typed
int on the model, so int(sequence.id) / int(alert.id) is redundant.
Same for the dict comprehension in get_detection_counts_by_sequence_ids
where the row values are already int.

The int(alert_id) at alerts.py:46 stays because alert_id there is
unpacked from a raw SQL row tuple, not a model attribute.

* test(alerts): use strict=True when zipping fixture sequences and counts

Per review feedback on PR #559: strict=False silently swallows length
mismatches. Both lists have length 3 today, so flipping to strict=True
guards against future fixture edits drifting out of sync.

* test(sequences): use utcnow() helper in detections-count test

Rebase fallout from #574 (datetime.utcnow deprecation): one new test
function still called datetime.utcnow() while the datetime import had
been removed from the file, causing NameError on collection.

Author: ThbltLmr

src/app/api/api_v1/endpoints/alerts.py

@@ -20,6 +20,7 @@
 from app.schemas.alerts import AlertReadWithSequences
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceRead
+from app.services.sequence_counts import get_detection_counts_by_sequence_ids
 from app.services.telemetry import telemetry_client
 
 router = APIRouter()
@@ -46,10 +47,16 @@ async def _fetch_sequences_by_alert_ids(session: AsyncSession, alert_ids: List[i
     return mapping
 
 
-def _serialize_alert(alert: Alert, sequences: List[Sequence]) -> AlertReadWithSequences:
+def _serialize_sequence(sequence: Sequence, detections_count: int = 0) -> SequenceRead:
+    return SequenceRead(**sequence.model_dump(), detections_count=detections_count)
+
+
+def _serialize_alert(
+    alert: Alert, sequences: List[Sequence], detection_counts: Dict[int, int]
+) -> AlertReadWithSequences:
     return AlertReadWithSequences(
         **alert.model_dump(),
-        sequences=[SequenceRead(**seq.model_dump()) for seq in sequences],
+        sequences=[_serialize_sequence(sequence, detection_counts.get(sequence.id, 0)) for sequence in sequences],
     )
 
 
@@ -66,9 +73,11 @@ async def get_alert(
     if UserRole.ADMIN not in token_payload.scopes:
         verify_org_rights(token_payload.organization_id, alert)
 
-    alert_id_int = int(alert.id)
-    seq_map = await _fetch_sequences_by_alert_ids(session, [alert_id_int])
-    return _serialize_alert(alert, seq_map.get(alert_id_int, []))
+    seq_map = await _fetch_sequences_by_alert_ids(session, [alert.id])
+    detection_counts = await get_detection_counts_by_sequence_ids(
+        session, [sequence.id for sequence in seq_map.get(alert.id, [])]
+    )
+    return _serialize_alert(alert, seq_map.get(alert.id, []), detection_counts)
 
 
 @router.get(
@@ -81,7 +90,7 @@ async def fetch_alert_sequences(
     alerts: AlertCRUD = Depends(get_alert_crud),
     session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
-) -> List[Sequence]:
+) -> List[SequenceRead]:
     telemetry_client.capture(token_payload.sub, event="alerts-sequences-get", properties={"alert_id": alert_id})
     alert = cast(Alert, await alerts.get(alert_id, strict=True))
     if UserRole.ADMIN not in token_payload.scopes:
@@ -93,7 +102,9 @@ async def fetch_alert_sequences(
     seq_stmt = seq_stmt.where(AlertSequence.alert_id == alert_id).order_by(order_clause).limit(limit)
 
     res = await session.exec(seq_stmt)
-    return list(res.all())
+    sequences = list(res.all())
+    detection_counts = await get_detection_counts_by_sequence_ids(session, [sequence.id for sequence in sequences])
+    return [_serialize_sequence(sequence, detection_counts.get(sequence.id, 0)) for sequence in sequences]
 
 
 @router.get(
@@ -118,9 +129,13 @@ async def fetch_latest_unlabeled_alerts(
     )
     alerts_res = await session.exec(alerts_stmt)
     alerts = alerts_res.unique().all()
-    alert_ids = [int(alert.id) for alert in alerts]
+    alert_ids = [alert.id for alert in alerts]
     seq_map = await _fetch_sequences_by_alert_ids(session, alert_ids)
-    return [_serialize_alert(alert, seq_map.get(int(alert.id), [])) for alert in alerts]
+    detection_counts = await get_detection_counts_by_sequence_ids(
+        session,
+        list({sequence.id for sequences in seq_map.values() for sequence in sequences}),
+    )
+    return [_serialize_alert(alert, seq_map.get(alert.id, []), detection_counts) for alert in alerts]
 
 
 @router.get("/all/fromdate", status_code=status.HTTP_200_OK, summary="Fetch all the alerts for a specific date")
@@ -143,9 +158,13 @@ async def fetch_alerts_from_date(
     )
     alerts_res = await session.exec(alerts_stmt)
     alerts = alerts_res.all()
-    alert_ids = [int(alert.id) for alert in alerts]
+    alert_ids = [alert.id for alert in alerts]
     seq_map = await _fetch_sequences_by_alert_ids(session, alert_ids)
-    return [_serialize_alert(alert, seq_map.get(int(alert.id), [])) for alert in alerts]
+    detection_counts = await get_detection_counts_by_sequence_ids(
+        session,
+        list({sequence.id for sequences in seq_map.values() for sequence in sequences}),
+    )
+    return [_serialize_alert(alert, seq_map.get(alert.id, []), detection_counts) for alert in alerts]
 
 
 @router.delete("/{alert_id}", status_code=status.HTTP_200_OK, summary="Delete an alert")

src/app/api/api_v1/endpoints/detections.py

@@ -554,7 +554,7 @@ async def fetch_detections(
 ) -> List[DetectionRead]:
     telemetry_client.capture(token_payload.sub, event="detections-fetch")
     if UserRole.ADMIN in token_payload.scopes:
-        return [DetectionRead(**elt.model_dump()) for elt in await detections.fetch_all()]
+        return [DetectionRead(**elt.model_dump()) for elt in await detections.fetch_all(order_by="id")]
 
     cameras_list = await cameras.fetch_all(filters=("organization_id", token_payload.organization_id))
     camera_ids = [camera.id for camera in cameras_list]

src/app/api/api_v1/endpoints/sequences.py

@@ -22,6 +22,7 @@
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceLabel, SequenceRead
 from app.services.overlap import compute_overlap
+from app.services.sequence_counts import get_detection_counts_by_sequence_ids
 from app.services.storage import s3_service
 from app.services.telemetry import telemetry_client
 
@@ -83,20 +84,26 @@ async def _refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Ale
     )
 
 
+def _serialize_sequence(sequence: Sequence, detections_count: int = 0) -> SequenceRead:
+    return SequenceRead(**sequence.model_dump(), detections_count=detections_count)
+
+
 @router.get("/{sequence_id}", status_code=status.HTTP_200_OK, summary="Fetch the information of a specific sequence")
 async def get_sequence(
     sequence_id: int = Path(..., gt=0),
     cameras: CameraCRUD = Depends(get_camera_crud),
     sequences: SequenceCRUD = Depends(get_sequence_crud),
+    session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
-) -> Sequence:
+) -> SequenceRead:
     telemetry_client.capture(token_payload.sub, event="sequences-get", properties={"sequence_id": sequence_id})
     sequence = cast(Sequence, await sequences.get(sequence_id, strict=True))
 
     if UserRole.ADMIN not in token_payload.scopes:
         await verify_org_rights(token_payload.organization_id, sequence.camera_id, cameras)
 
-    return SequenceRead(**sequence.model_dump())
+    counts = await get_detection_counts_by_sequence_ids(session, [sequence.id])
+    return _serialize_sequence(sequence, counts.get(sequence.id, 0))
 
 
 @router.get(
@@ -155,7 +162,8 @@ async def fetch_latest_unlabeled_sequences(
             .limit(15)
         )
     ).all()
-    return [SequenceRead(**elt.model_dump()) for elt in fetched_sequences]
+    counts = await get_detection_counts_by_sequence_ids(session, [sequence.id for sequence in fetched_sequences])
+    return [_serialize_sequence(sequence, counts.get(sequence.id, 0)) for sequence in fetched_sequences]
 
 
 @router.get("/all/fromdate", status_code=status.HTTP_200_OK, summary="Fetch all the sequences for a specific date")
@@ -180,7 +188,8 @@ async def fetch_sequences_from_date(
             .offset(offset)
         )
     ).all()
-    return [SequenceRead(**elt.model_dump()) for elt in fetched_sequences]
+    counts = await get_detection_counts_by_sequence_ids(session, [sequence.id for sequence in fetched_sequences])
+    return [_serialize_sequence(sequence, counts.get(sequence.id, 0)) for sequence in fetched_sequences]
 
 
 @router.delete("/{sequence_id}", status_code=status.HTTP_200_OK, summary="Delete a sequence")

src/app/schemas/sequences.py

@@ -22,4 +22,4 @@ class SequenceLabel(BaseModel):
 
 
 class SequenceRead(Sequence):
-    pass
+    detections_count: int = 0

src/app/services/sequence_counts.py

@@ -0,0 +1,25 @@
+# Copyright (C) 2025-2026, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
+
+
+from typing import Any, Dict, List, cast
+
+from sqlmodel import func, select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from app.models import Detection
+
+
+async def get_detection_counts_by_sequence_ids(session: AsyncSession, sequence_ids: List[int]) -> Dict[int, int]:
+    if not sequence_ids:
+        return {}
+
+    stmt: Any = (
+        select(cast(Any, Detection.sequence_id), func.count(Detection.id))
+        .where(cast(Any, Detection.sequence_id).in_(sequence_ids))
+        .group_by(cast(Any, Detection.sequence_id))
+    )
+    res = await session.exec(stmt)
+    return {sequence_id: detections_count for sequence_id, detections_count in res.all() if sequence_id is not None}

src/tests/endpoints/test_alerts.py

@@ -14,14 +14,18 @@
 
 from app.core.config import settings
 from app.core.time import utcnow
-from app.models import Alert, AlertSequence, AnnotationType, Camera, Organization, Pose, Sequence
+from app.models import Alert, AlertSequence, AnnotationType, Camera, Detection, Organization, Pose, Sequence
 from app.services.overlap import compute_overlap
 
 
 async def _create_alert_with_sequences(
     session: AsyncSession, org_id: int, camera_id: int, lat: float, lon: float
-) -> Tuple[Alert, List[int]]:
+) -> Tuple[Alert, List[int], List[int]]:
     now = utcnow()
+    pose = (
+        await session.exec(select(Pose).where(Pose.camera_id == camera_id).order_by(Pose.id))  # type: ignore[attr-defined]
+    ).first()
+    assert pose is not None
     seq_payloads = [
         {
             "camera_id": camera_id,
@@ -48,6 +52,7 @@ async def _create_alert_with_sequences(
             "cone_angle": 3.0,
         },
     ]
+    detections_count_by_sequence = [2, 1, 0]
     sequences: List[Sequence] = []
     for idx, payload in enumerate(seq_payloads):
         seq = Sequence(
@@ -60,6 +65,20 @@ async def _create_alert_with_sequences(
     await session.commit()
     for seq in sequences:
         await session.refresh(seq)
+    for sequence, detections_count in zip(sequences, detections_count_by_sequence, strict=True):
+        for det_idx in range(detections_count):
+            session.add(
+                Detection(
+                    camera_id=sequence.camera_id,
+                    pose_id=pose.id,
+                    sequence_id=sequence.id,
+                    bucket_key=f"alert-seq-{sequence.id}-{det_idx}.jpg",
+                    bbox="[(.1,.1,.7,.8,.9)]",
+                    others_bboxes=None,
+                    created_at=now - timedelta(seconds=det_idx),
+                )
+            )
+    await session.commit()
 
     alert = Alert(
         organization_id=org_id,
@@ -75,14 +94,15 @@ async def _create_alert_with_sequences(
     for seq in sequences:
         session.add(AlertSequence(alert_id=alert.id, sequence_id=seq.id))
     await session.commit()
-    return alert, [seq.id for seq in sequences]
+    return alert, [seq.id for seq in sequences], detections_count_by_sequence
 
 
 @pytest.mark.asyncio
 async def test_get_alert_and_sequences(async_client: AsyncClient, detection_session: AsyncSession):
-    alert, seq_ids = await _create_alert_with_sequences(
+    alert, seq_ids, detections_count_by_sequence = await _create_alert_with_sequences(
         detection_session, org_id=1, camera_id=1, lat=48.3856355, lon=2.7323256
     )
+    expected_counts = dict(zip(seq_ids, detections_count_by_sequence, strict=False))
 
     auth = pytest.get_token(
         pytest.user_table[0]["id"], pytest.user_table[0]["role"].split(), pytest.user_table[0]["organization_id"]
@@ -97,19 +117,23 @@ async def test_get_alert_and_sequences(async_client: AsyncClient, detection_sess
     assert payload["started_at"] == alert.started_at.isoformat()
     assert payload["last_seen_at"] == alert.last_seen_at.isoformat()
     assert {seq["id"] for seq in payload["sequences"]} == set(seq_ids)
+    assert {seq["id"]: seq["detections_count"] for seq in payload["sequences"]} == expected_counts
 
     resp = await async_client.get(f"/alerts/{alert.id}/sequences?limit=5&desc=true", headers=auth)
     assert resp.status_code == 200, resp.text
     returned = resp.json()
     last_seen_times = [item["last_seen_at"] for item in returned]
     assert last_seen_times == sorted(last_seen_times, reverse=True)
+    assert {sequence["id"]: sequence["detections_count"] for sequence in returned} == expected_counts
+    assert any(sequence["detections_count"] == 0 for sequence in returned)
 
 
 @pytest.mark.asyncio
 async def test_alerts_unlabeled_latest(async_client: AsyncClient, detection_session: AsyncSession):
-    alert, seq_ids = await _create_alert_with_sequences(
+    alert, seq_ids, detections_count_by_sequence = await _create_alert_with_sequences(
         detection_session, org_id=1, camera_id=1, lat=48.3856355, lon=2.7323256
     )
+    expected_counts = dict(zip(seq_ids, detections_count_by_sequence, strict=False))
 
     auth = pytest.get_token(
         pytest.user_table[0]["id"], pytest.user_table[0]["role"].split(), pytest.user_table[0]["organization_id"]
@@ -124,13 +148,16 @@ async def test_alerts_unlabeled_latest(async_client: AsyncClient, detection_sess
     assert returned["started_at"] == alert.started_at.isoformat()
     assert returned["last_seen_at"] == alert.last_seen_at.isoformat()
     assert {seq["id"] for seq in returned["sequences"]} == set(seq_ids)
+    assert {seq["id"]: seq["detections_count"] for seq in returned["sequences"]} == expected_counts
+    assert any(seq["detections_count"] == 0 for seq in returned["sequences"])
 
 
 @pytest.mark.asyncio
 async def test_alerts_from_date(async_client: AsyncClient, detection_session: AsyncSession):
-    alert, seq_ids = await _create_alert_with_sequences(
+    alert, seq_ids, detections_count_by_sequence = await _create_alert_with_sequences(
         detection_session, org_id=1, camera_id=1, lat=48.3856355, lon=2.7323256
     )
+    expected_counts = dict(zip(seq_ids, detections_count_by_sequence, strict=False))
     date_str = alert.started_at.date().isoformat()
 
     auth = pytest.get_token(
@@ -146,6 +173,8 @@ async def test_alerts_from_date(async_client: AsyncClient, detection_session: As
     assert started_times == sorted(started_times, reverse=True)
     alert_payload = next(item for item in returned if item["id"] == alert.id)
     assert {seq["id"] for seq in alert_payload["sequences"]} == set(seq_ids)
+    assert {seq["id"]: seq["detections_count"] for seq in alert_payload["sequences"]} == expected_counts
+    assert any(seq["detections_count"] == 0 for seq in alert_payload["sequences"])
 
 
 @pytest.mark.asyncio