Formalise MWL status as an enum

The database values and check constraint mirror DICOM PACS statuses

Author: steventux

src/services/init_worklist_db.sql

@@ -26,7 +26,7 @@ CREATE TABLE IF NOT EXISTS worklist_items (
     procedure_code TEXT,
 
     -- Status tracking
-    status TEXT DEFAULT 'SCHEDULED' CHECK(status IN ('SCHEDULED', 'IN_PROGRESS', 'COMPLETED', 'DISCONTINUED')),
+    status TEXT DEFAULT 'SCHEDULED' CHECK(status IN ('SCHEDULED', 'IN PROGRESS', 'COMPLETED', 'DISCONTINUED')),
 
     -- DICOM identifiers
     study_instance_uid TEXT,

src/services/mwl/__init__.py

@@ -1,5 +1,14 @@
 """Modality Worklist (MWL) services for DICOM worklist management."""
 
+from enum import Enum
+
 from services.mwl.c_find import CFindHandler
 
 __all__ = ["CFindHandler"]
+
+
+class MWLStatus(Enum):
+    SCHEDULED = "SCHEDULED"
+    IN_PROGRESS = "IN PROGRESS"
+    COMPLETED = "COMPLETED"
+    DISCONTINUED = "DISCONTINUED"

src/services/mwl/n_create.py

@@ -5,12 +5,11 @@
 from pynetdicom.sop_class import ModalityPerformedProcedureStep  # pyright: ignore[reportAttributeAccessIssue]
 
 from services.dicom import DUPLICATE_SOP_INSTANCE, INVALID_ATTRIBUTE, MISSING_ATTRIBUTE, PROCESSING_FAILURE, SUCCESS
+from services.mwl import MWLStatus
 from services.storage import MWLStorage
 
 logger = logging.getLogger(__name__)
 
-UNKNOWN = "UNKNOWN"
-
 
 class NCreate:
     def __init__(self, storage: MWLStorage):
@@ -31,11 +30,14 @@ def call(self, event: Event) -> tuple[int, Dataset | None]:
                 return DUPLICATE_SOP_INSTANCE, None
 
             attr_list = event.attribute_list
+            status = getattr(attr_list, "PerformedProcedureStepStatus", None)
 
-            if "PerformedProcedureStepStatus" not in attr_list:
+            if not status:
+                logger.warning("MPPS N-CREATE: Missing PerformedProcedureStepStatus in request")
                 return MISSING_ATTRIBUTE, None
 
-            if attr_list.PerformedProcedureStepStatus.upper() != "IN PROGRESS":
+            if status.upper() != MWLStatus.IN_PROGRESS.value:
+                logger.warning("MPPS N-CREATE: Invalid PerformedProcedureStepStatus value: %s", status)
                 return INVALID_ATTRIBUTE, None
 
             ds.SOPClassUID = ModalityPerformedProcedureStep
@@ -50,20 +52,22 @@ def call(self, event: Event) -> tuple[int, Dataset | None]:
             sps = attr_list.ScheduledStepAttributesSequence[0]
             accession_number = sps.get("AccessionNumber")
 
-            logger.info("MPPS N-CREATE: Started procedure for Accession Number: %w", accession_number)
+            logger.info("MPPS N-CREATE: Started procedure for Accession Number: %s", accession_number)
 
             if not accession_number:
                 logger.warning("MPPS N-CREATE: Missing Accession Number in ScheduledStepAttributesSequence")
                 return MISSING_ATTRIBUTE, None
 
-            source_message_id = self.storage.update_status(accession_number, "IN_PROGRESS", ds.SOPInstanceUID)
+            source_message_id = self.storage.update_status(
+                accession_number, MWLStatus.IN_PROGRESS.value, ds.SOPInstanceUID
+            )
             if source_message_id:
-                logger.info(f"Worklist item updated: {accession_number} -> IN_PROGRESS")
+                logger.info("Worklist item updated: %s -> %s", accession_number, MWLStatus.IN_PROGRESS.value)
             else:
-                logger.warning(f"Could not find accession {accession_number} in database")
+                logger.warning("Could not find accession %s in database", accession_number)
 
         except Exception as e:
-            logger.error(f"Error in handle_create: {str(e)}", exc_info=True)
+            logger.error("Error in handle_create: %s", str(e), exc_info=True)
             return PROCESSING_FAILURE, None
 
         # Success - return the created dataset

tests/integration/test_n_create_updates_worklist_status.py

@@ -73,4 +73,4 @@ def test_n_create_updates_worklist_status(self, tmp_dir, worklist_item):
 
         assert updated_item is not None
         assert updated_item.mpps_instance_uid == mpps_instance_uid
-        assert updated_item.status == "IN_PROGRESS"
+        assert updated_item.status == "IN PROGRESS"

tests/integration/test_n_set_updates_worklist_status.py

@@ -45,7 +45,7 @@ def worklist_item(self):
     def test_n_set_updates_worklist_status(self, tmp_dir, worklist_item, mpps_instance_uid):
         storage = MWLStorage(f"{tmp_dir}/test.db")
         accession_number = storage.store_worklist_item(worklist_item)
-        storage.update_status(accession_number, "IN_PROGRESS", mpps_instance_uid)
+        storage.update_status(accession_number, "IN PROGRESS", mpps_instance_uid)
 
         ae = AE(ae_title="MODALITY_SCU")
         ae.add_requested_context(ModalityPerformedProcedureStep)

tests/services/mwl/test_n_create.py

@@ -59,7 +59,7 @@ def test_ncreate_success(self, storage, event, sop_instance_uid):
         assert ds.PerformedProcedureStepStatus == "IN PROGRESS"
 
         storage.mpps_instance_exists.assert_called_once_with(sop_instance_uid)
-        storage.update_status.assert_called_once_with("ACC123", "IN_PROGRESS", sop_instance_uid)
+        storage.update_status.assert_called_once_with("ACC123", "IN PROGRESS", sop_instance_uid)
 
     def test_ncreate_missing_sop_instance_uid(self, storage, event):
         event.request.AffectedSOPInstanceUID = None