Move status update transition logic to services.mwl module

Author: steventux

src/services/mwl/__init__.py

@@ -3,8 +3,37 @@
 from enum import Enum
 
 
+class InvalidStatusTransitionError(Exception):
+    """Raised when a requested status transition is not permitted."""
+
+    pass
+
+
 class MWLStatus(Enum):
     SCHEDULED = "SCHEDULED"
     IN_PROGRESS = "IN PROGRESS"
     COMPLETED = "COMPLETED"
     DISCONTINUED = "DISCONTINUED"
+
+
+class MWLStatusManager:
+    _TRANSITIONS = {
+        MWLStatus.IN_PROGRESS: MWLStatus.SCHEDULED,
+        MWLStatus.COMPLETED: MWLStatus.IN_PROGRESS,
+        MWLStatus.DISCONTINUED: MWLStatus.IN_PROGRESS,
+    }
+
+    @staticmethod
+    def transition_for(status: str) -> tuple[MWLStatus, MWLStatus]:
+        """
+        Get the previous and next status for a given MWL status.
+
+        Raises:
+            InvalidStatusTransitionError: If the transition is not permitted
+        """
+        try:
+            current_status = MWLStatus(status)
+            previous_status = MWLStatusManager._TRANSITIONS[current_status]
+            return previous_status, current_status
+        except KeyError:
+            raise InvalidStatusTransitionError(f"Cannot transition to '{status}'")

src/services/storage.py

@@ -7,7 +7,7 @@
 from typing import Dict, List, Optional
 
 from models import WorklistItem
-from services.mwl import MWLStatus
+from services.mwl import MWLStatusManager
 
 logger = logging.getLogger(__name__)
 
@@ -285,25 +285,13 @@ class WorklistItemNotFoundError(Exception):
     pass
 
 
-class InvalidStatusTransitionError(Exception):
-    """Raised when a requested status transition is not permitted."""
-
-    pass
-
-
 class WorklistItemExistsError(Exception):
     """Raised when a worklist item with the same accession number already exists."""
 
     pass
 
 
 class MWLStorage(Storage):
-    _STATUS_TRANSITIONS: dict[MWLStatus, MWLStatus] = {
-        MWLStatus.IN_PROGRESS: MWLStatus.SCHEDULED,
-        MWLStatus.COMPLETED: MWLStatus.IN_PROGRESS,
-        MWLStatus.DISCONTINUED: MWLStatus.IN_PROGRESS,
-    }
-
     def __init__(self, db_path: str = "/var/lib/pacs/worklist.db"):
         """
         Initialize Worklist storage.
@@ -481,14 +469,8 @@ def update_status(
 
         Returns:
             source_message_id if item was updated, None if not found
-
-        Raises:
-            InvalidStatusTransitionError: If the transition is not permitted
         """
-        target = MWLStatus(status)
-        if target not in self._STATUS_TRANSITIONS:
-            raise InvalidStatusTransitionError(f"Cannot transition to '{status}'")
-        from_status = self._STATUS_TRANSITIONS[target]
+        from_status, to_status = MWLStatusManager.transition_for(status)
 
         with self._get_connection() as conn:
             cursor = conn.execute(
@@ -500,7 +482,7 @@ def update_status(
                 WHERE accession_number = ?
                   AND status = ?
                 """,
-                (status, mpps_instance_uid, accession_number, from_status.value),
+                (to_status.value, mpps_instance_uid, accession_number, from_status.value),
             )
             conn.commit()
 

tests/services/test_storage.py

@@ -6,8 +6,8 @@
 from pydicom.uid import generate_uid
 
 from models import WorklistItem
+from services.mwl import InvalidStatusTransitionError
 from services.storage import (
-    InvalidStatusTransitionError,
     MWLStorage,
     PACSStorage,
     WorklistItemExistsError,
@@ -116,7 +116,7 @@ def test_store_instance_saves_to_db(self, pacs_storage):
         assert row["accession_number"] == metadata["accession_number"]
 
 
-class TestWorkingStorage:
+class TestMWLStorage:
     def _insert_item(self, storage, result):
         item = WorklistItem(**result)
         storage.store_worklist_item(item)