fix: commit multi-step CRUD workflows atomically

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

@@ -493,15 +493,14 @@ async def unmatch_alert_sequence(
         .where(cast(Any, AlertSequence.sequence_id) == sequence_id)
     )
     await session.exec(delete_stmt)
-    await session.commit()
-
-    await refresh_alert_state(alert_id, session, alerts)
+    await refresh_alert_state(alert_id, session, alerts, commit=False)
 
     other_links_stmt: Any = (
         select(func.count()).select_from(AlertSequence).where(AlertSequence.sequence_id == sequence_id)
     )
     other_links = int((await session.exec(other_links_stmt)).one())
     if other_links > 0:
+        await session.commit()
         return None
 
     sequence = cast(Sequence, await sequences.get(sequence_id, strict=True))
@@ -518,7 +517,8 @@ async def unmatch_alert_sequence(
             last_seen_at=sequence.last_seen_at,
             lat=None,
             lon=None,
-        )
+        ),
+        commit=False,
     )
     session.add(AlertSequence(alert_id=new_alert.id, sequence_id=sequence_id))
     await session.commit()
@@ -543,6 +543,6 @@ async def delete_alert(
     # Delete associations
     delete_stmt: Any = delete(AlertSequence).where(AlertSequence.alert_id == cast(Any, alert_id))
     await session.exec(delete_stmt)
-    await session.commit()
     # Delete alert
-    await alerts.delete(alert_id)
+    await alerts.delete(alert_id, commit=False)
+    await session.commit()

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

@@ -145,7 +145,8 @@ async def _create_continuity_detection(
             bbox=EMPTY_BBOXES,
             sequence_id=sequence_id,
             recorded_at=recorded_at,
-        )
+        ),
+        commit=False,
     )
 
 
@@ -171,7 +172,7 @@ async def _attach_continuity_detections(
         created.append(
             await _create_continuity_detection(detections, camera_id, pose_id, bucket_key, seq.id, recorded_at)
         )
-        await sequences.enqueue_validation(seq.id)
+        await sequences.enqueue_validation(seq.id, commit=False)
     return created
 
 
@@ -297,6 +298,7 @@ async def _maybe_update_alert(
         await alerts.update(
             target_alert_id,
             AlertUpdate(lat=lat, lon=lon, started_at=new_start_at, last_seen_at=new_last_seen),
+            commit=False,
         )
 
 
@@ -357,7 +359,7 @@ async def _merge_alerts(
     await session.exec(delete_links_stmt)
     delete_alerts_stmt: Any = delete(Alert).where(cast(Any, Alert.id).in_(list(absorbed_ids)))
     await session.exec(delete_alerts_stmt)
-    await session.commit()
+    await session.flush()
 
 
 def _rewrite_after_merge(
@@ -435,7 +437,8 @@ async def _get_or_create_alert_id(
             lon=location[1] if isinstance(location, tuple) else None,
             started_at=start_at,
             last_seen_at=last_seen_at,
-        )
+        ),
+        commit=False,
     )
     return alert.id, set()
 
@@ -462,6 +465,8 @@ async def _attach_sequence_to_alert(
     sequences: SequenceCRUD,
     alerts: AlertCRUD,
     anchor_on_sequence: bool = False,
+    *,
+    commit: bool = True,
 ) -> Optional[int]:
     """Assign the given sequence to an alert based on cone/time overlap."""
     camera_by_id = await _get_camera_by_id(camera, cameras, sequence_.camera_id)
@@ -507,7 +512,10 @@ async def _attach_sequence_to_alert(
 
     if to_link:
         session.add_all(to_link)
+    if commit:
         await session.commit()
+    else:
+        await session.flush()
 
     return alert_id
 
@@ -589,6 +597,7 @@ async def create_detection(
         continuity_dets = await _attach_continuity_detections(
             detections, sequences, continuity_sequences, token_payload.sub, pose_id, bucket_key, effective_recorded_at
         )
+        await detections.session.commit()
         return DetectionRead(**continuity_dets[0].model_dump())
 
     # Upload media
@@ -619,7 +628,8 @@ async def create_detection(
                 bbox=single_bboxes,
                 others_bboxes=others_bboxes,
                 recorded_at=effective_recorded_at,
-            )
+            ),
+            commit=False,
         )
 
         det_bbox = _parse_bbox(bbox_str)
@@ -645,12 +655,12 @@ async def create_detection(
                 break
 
         if matched_sequence is not None:
-            await sequences.update(matched_sequence.id, SequenceUpdate(last_seen_at=det.created_at))
-            det = await detections.update(det.id, DetectionSequence(sequence_id=matched_sequence.id))
+            await sequences.update(matched_sequence.id, SequenceUpdate(last_seen_at=det.created_at), commit=False)
+            det = await detections.update(det.id, DetectionSequence(sequence_id=matched_sequence.id), commit=False)
             # Only the primary bbox tracks the sequence; siblings in others_bboxes are unrelated detections.
             det_max_conf = max_conf_from_bboxes(det.bbox)
             if det_max_conf is not None:
-                await sequences.bump_max_conf(matched_sequence.id, det_max_conf)
+                await sequences.bump_max_conf(matched_sequence.id, det_max_conf, commit=False)
             affected_sequences.add(matched_sequence.id)
         else:
             det_filters: List[tuple[str, Any]] = [
@@ -694,10 +704,13 @@ async def create_detection(
                         started_at=first_det.recorded_at,
                         last_seen_at=det.created_at,
                         max_conf=seq_max_conf,
-                    )
+                    ),
+                    commit=False,
                 )
                 for det_ in overlapping_dets:
-                    updated = await detections.update(det_.id, DetectionSequence(sequence_id=sequence_.id))
+                    updated = await detections.update(
+                        det_.id, DetectionSequence(sequence_id=sequence_.id), commit=False
+                    )
                     if det_.id == det.id:
                         det = updated
                 affected_sequences.add(sequence_.id)
@@ -724,7 +737,10 @@ async def create_detection(
     # claims due sequences from the DB and runs the gated pipeline: triangulation and ALL
     # notification channels (webhooks, Telegram, Slack) fire only once validated.
     for seq_id in affected_sequences:
-        await sequences.enqueue_validation(seq_id)
+        await sequences.enqueue_validation(seq_id, commit=False)
+
+    # Detection rows, sequence membership/confidence and validation jobs become visible together.
+    await detections.session.commit()
 
     first_det = cast(Detection, await detections.get(created[0].id, strict=True))
     return DetectionRead(**first_det.model_dump())

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

@@ -259,16 +259,16 @@ async def delete_sequence(
     # Unset the sequence_id in the detections
     det_ids = await session.exec(select(Detection.id).where(Detection.sequence_id == sequence_id))
     for det_id in det_ids.all():
-        await detections.update(det_id, DetectionSequence(sequence_id=None))
+        await detections.update(det_id, DetectionSequence(sequence_id=None), commit=False)
     # Drop alert links for this sequence to avoid FK issues
     delete_stmt: Any = delete(AlertSequence).where(cast(Any, AlertSequence.sequence_id) == sequence_id)
     await session.exec(delete_stmt)
-    await session.commit()
     # Delete the sequence
-    await sequences.delete(sequence_id)
+    await sequences.delete(sequence_id, commit=False)
     # Refresh affected alerts
     for aid in alert_ids:
-        await refresh_alert_state(aid, session, alerts)
+        await refresh_alert_state(aid, session, alerts, commit=False)
+    await session.commit()
 
 
 @router.patch("/{sequence_id}/label", status_code=status.HTTP_200_OK, summary="Label the nature of the sequence")
@@ -288,7 +288,7 @@ async def label_sequence(
         await verify_org_rights(token_payload.organization_id, sequence.camera_id, cameras)
 
     previous_label = sequence.is_wildfire
-    updated = await sequences.update(sequence_id, payload)
+    updated = await sequences.update(sequence_id, payload, commit=False)
 
     # Reverting a non-wildfire label: re-run cone matching. The sequence keeps its lonely alert
     # until the matching merges it into an overlapping one, so it is never left without an alert.
@@ -300,10 +300,14 @@ async def label_sequence(
     ):
         camera = cast(Camera, await cameras.get(sequence.camera_id, strict=True))
         async with _organization_alert_lock(camera.organization_id):
-            await _attach_sequence_to_alert(updated, camera, cameras, sequences, alerts, anchor_on_sequence=True)
+            await _attach_sequence_to_alert(
+                updated, camera, cameras, sequences, alerts, anchor_on_sequence=True, commit=False
+            )
+            await session.commit()
         return updated
 
     if payload.is_wildfire is None or payload.is_wildfire == AnnotationType.WILDFIRE_SMOKE:
+        await session.commit()
         return updated
 
     alert_ids_res = await session.exec(select(AlertSequence.alert_id).where(AlertSequence.sequence_id == sequence_id))
@@ -320,13 +324,13 @@ async def label_sequence(
         )
         siblings_res = await session.exec(siblings_stmt)
         if siblings_res.first() is None:
+            await session.commit()
             return updated
 
         delete_links: Any = delete(AlertSequence).where(cast(Any, AlertSequence.sequence_id) == sequence_id)
         await session.exec(delete_links)
-        await session.commit()
         for aid in alert_ids:
-            await refresh_alert_state(aid, session, alerts)
+            await refresh_alert_state(aid, session, alerts, commit=False)
 
     # Create a fresh alert for this sequence alone
     camera = cast(Camera, await cameras.get(sequence.camera_id, strict=True))
@@ -337,7 +341,8 @@ async def label_sequence(
             last_seen_at=sequence.last_seen_at,
             lat=None,
             lon=None,
-        )
+        ),
+        commit=False,
     )
     session.add(AlertSequence(alert_id=new_alert.id, sequence_id=sequence_id))
     await session.commit()

src/app/crud/base.py

@@ -23,18 +23,27 @@ def __init__(self, session: AsyncSession, model: Type[ModelType]) -> None:
         self.session = session
         self.model = model
 
-    async def create(self, payload: CreateSchemaType) -> ModelType:
+    async def create(self, payload: CreateSchemaType, *, commit: bool = True) -> ModelType:
+        """Insert a row, optionally leaving the transaction to the caller.
+
+        Deferred writes are flushed so generated IDs/defaults and subsequent queries work.
+        The caller must commit the complete workflow; session cleanup rolls it back on error.
+        """
         entry = self.model(**payload.model_dump())
         try:
             self.session.add(entry)
-            await self.session.commit()
+            if commit:
+                await self.session.commit()
+            else:
+                await self.session.flush()
         except exc.IntegrityError as error:
             await self.session.rollback()
             raise HTTPException(
                 status_code=status.HTTP_409_CONFLICT,
                 detail=f"An entry with the same index already exists : {error!s}",
             )
-        await self.session.refresh(entry)
+        if commit:
+            await self.session.refresh(entry)
 
         return entry
 
@@ -105,25 +114,29 @@ async def fetch_all(
         result = await self.session.exec(statement=statement)
         return [r for r in result]
 
-    async def update(self, entry_id: int, payload: UpdateSchemaType) -> ModelType:
+    async def update(self, entry_id: int, payload: UpdateSchemaType, *, commit: bool = True) -> ModelType:
         access = cast(ModelType, await self.get(entry_id, strict=True))
         values = payload.model_dump(exclude_unset=True)
 
         for k, v in values.items():
             setattr(access, k, v)
 
         self.session.add(access)
-        await self.session.commit()
-        await self.session.refresh(access)
+        if commit:
+            await self.session.commit()
+            await self.session.refresh(access)
+        else:
+            await self.session.flush()
 
         return access
 
-    async def delete(self, entry_id: int) -> None:
+    async def delete(self, entry_id: int, *, commit: bool = True) -> None:
         await self.get(entry_id, strict=True)
         statement = delete(self.model).where(cast(Any, self.model).id == entry_id)  # ty: ignore[invalid-argument-type]
 
         await self.session.exec(statement=statement)  # type: ignore[call-overload]
-        await self.session.commit()
+        if commit:
+            await self.session.commit()
 
     async def get_in(self, list_: List[Any], field_name: str) -> List[ModelType]:
         statement: Any = select(self.model).where(getattr(self.model, field_name).in_(list_))

src/app/crud/crud_sequence.py

@@ -26,7 +26,7 @@ class SequenceCRUD(BaseCRUD[Sequence, Sequence, Union[SequenceUpdate, SequenceLa
     def __init__(self, session: AsyncSession) -> None:
         super().__init__(session, Sequence)
 
-    async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
+    async def bump_max_conf(self, sequence_id: int, candidate: float, *, commit: bool = True) -> None:
         """Atomically raise sequences.max_conf to candidate if higher (or set if NULL).
 
         Uses a portable CASE expression so it runs on SQLite as well as Postgres.
@@ -38,7 +38,8 @@ async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
         )
         stmt: Any = update(Sequence).where(cast(Any, Sequence.id) == sequence_id).values(max_conf=bumped)
         await self.session.exec(stmt)
-        await self.session.commit()
+        if commit:
+            await self.session.commit()
 
     async def set_temporal_score(
         self,
@@ -83,7 +84,7 @@ async def claim_validation(self, sequence_id: int, validation_status: Union[str,
         await self.session.commit()
         return bool(getattr(result, "rowcount", 0))
 
-    async def enqueue_validation(self, sequence_id: int) -> None:
+    async def enqueue_validation(self, sequence_id: int, *, commit: bool = True) -> None:
         """Mark the sequence as due for temporal validation (the DB-backed queue).
 
         Idempotent and FIFO-preserving: ``COALESCE`` keeps the oldest due timestamp, so a
@@ -101,7 +102,8 @@ async def enqueue_validation(self, sequence_id: int) -> None:
             .values(validation_due_at=func.coalesce(due_col, utcnow()))
         )
         await self.session.exec(stmt)
-        await self.session.commit()
+        if commit:
+            await self.session.commit()
 
     async def claim_due_validation(self, lease_seconds: float) -> Union[Sequence, None]:
         """Claim the oldest due sequence for validation, or None when nothing is due.

src/app/services/alerts.py

@@ -17,7 +17,7 @@
 __all__ = ["refresh_alert_state"]
 
 
-async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: AlertCRUD) -> None:
+async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: AlertCRUD, *, commit: bool = True) -> None:
     """Recompute an alert's bounds and location from its remaining sequences, or delete it if empty."""
     remaining_stmt: Any = (
         select(Sequence, Camera)
@@ -28,7 +28,7 @@ async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Aler
     remaining_res = await session.exec(remaining_stmt)
     rows = remaining_res.all()
     if not rows:
-        await alerts.delete(alert_id)
+        await alerts.delete(alert_id, commit=commit)
         return
 
     seqs = [row[0] for row in rows]
@@ -62,4 +62,5 @@ async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Aler
             lat=loc[0] if loc else None,
             lon=loc[1] if loc else None,
         ),
+        commit=commit,
     )

src/tests/endpoints/test_sequences.py

@@ -753,11 +753,11 @@ async def test_unit_label_sequence_as_other_smoke_refreshes_alert(
 
     # 3. Assertions
     mock_sequences_crud.get.assert_called_once_with(1, strict=True)
-    mock_sequences_crud.update.assert_called_once_with(1, payload)
+    mock_sequences_crud.update.assert_called_once_with(1, payload, commit=False)
 
     # Three session.exec calls: alert_ids lookup, siblings probe, delete links
     assert mock_session.exec.call_count == 3
-    mock_refresh_alert_state.assert_called_once_with(101, mock_session, mock_alerts_crud)
+    mock_refresh_alert_state.assert_called_once_with(101, mock_session, mock_alerts_crud, commit=False)
 
     # Verify a new alert was created for this sequence
     mock_alerts_crud.create.assert_called_once()
@@ -813,6 +813,7 @@ async def test_unit_relabel_sequence_to_wildfire_smoke_reattaches(
         mock_sequences_crud,
         mock_alerts_crud,
         anchor_on_sequence=True,
+        commit=False,
     )
     mock_session.exec.assert_not_called()
     mock_alerts_crud.create.assert_not_called()
@@ -899,7 +900,7 @@ async def test_unit_label_sequence_as_wildfire_smoke_does_not_refresh():
     )
 
     # 3. Assertions
-    mock_sequences_crud.update.assert_called_once_with(1, payload)
+    mock_sequences_crud.update.assert_called_once_with(1, payload, commit=False)
     mock_session.exec.assert_not_called()
     mock_alerts_crud.create.assert_not_called()