Listen for new PACS entries and send upstream to Manage (#11)

Listen for new PACS entries and send upstream to Manage

Author: carlosmartinez

.gitleaksignore

@@ -1,33 +1,108 @@
 # Gitleaks allowlist for false positives
 # Add file:line patterns for legitimate secrets that gitleaks flags incorrectly
 #
-# Format: path/to/file:line_number:hash
-# Example: docs/examples/sample.md:42:b1234567890abcdef
+# Format: path/to/file:rule_id:line_number
+# These are DICOM UIDs that match the IPv4 pattern
 
-# Add your allowlist entries below this line
+# Git history fingerprints (for renamed/deleted files in old commits)
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/integration/test_upload_listener.py:ipv4:38
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/integration/test_upload_listener.py:ipv4:66
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/integration/test_upload_listener.py:ipv4:71
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/integration/test_upload_listener.py:ipv4:75
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/integration/test_upload_listener.py:ipv4:85
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/integration/test_upload_listener.py:ipv4:89
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/integration/test_upload_listener.py:ipv4:99
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/integration/test_upload_listener.py:ipv4:102
+064a96209818176322c8f84ab3781f78fc43d7c5:tests/services/dicom/test_cloud_uploader.py:ipv4:18
+caf78850d3de96da17abb731305cfed073fa0986:tests/integration/test_upload_listener.py:ipv4:38
+caf78850d3de96da17abb731305cfed073fa0986:tests/integration/test_upload_listener.py:ipv4:66
+caf78850d3de96da17abb731305cfed073fa0986:tests/integration/test_upload_listener.py:ipv4:71
+caf78850d3de96da17abb731305cfed073fa0986:tests/integration/test_upload_listener.py:ipv4:75
+caf78850d3de96da17abb731305cfed073fa0986:tests/integration/test_upload_listener.py:ipv4:85
+caf78850d3de96da17abb731305cfed073fa0986:tests/integration/test_upload_listener.py:ipv4:89
+caf78850d3de96da17abb731305cfed073fa0986:tests/integration/test_upload_listener.py:ipv4:99
+caf78850d3de96da17abb731305cfed073fa0986:tests/integration/test_upload_listener.py:ipv4:102
+caf78850d3de96da17abb731305cfed073fa0986:tests/services/dicom/test_cloud_uploader.py:ipv4:18
+fda868296050b269384b6bd072136c0c474c61ec:tests/integration/test_pacs_storage_upload.py:ipv4:95
+fda868296050b269384b6bd072136c0c474c61ec:tests/integration/test_pacs_storage_upload.py:ipv4:124
+fda868296050b269384b6bd072136c0c474c61ec:tests/integration/test_pacs_storage_upload.py:ipv4:130
+fda868296050b269384b6bd072136c0c474c61ec:tests/integration/test_pacs_storage_upload.py:ipv4:134
+fda868296050b269384b6bd072136c0c474c61ec:tests/integration/test_pacs_storage_upload.py:ipv4:145
+fda868296050b269384b6bd072136c0c474c61ec:tests/integration/test_pacs_storage_upload.py:ipv4:149
+fda868296050b269384b6bd072136c0c474c61ec:tests/integration/test_pacs_storage_upload.py:ipv4:161
+fda868296050b269384b6bd072136c0c474c61ec:tests/integration/test_pacs_storage_upload.py:ipv4:164
+19d6a0432e5d103823c45938f4a4edfa65239fd4:tests/services/test_storage.py:ipv4:136
+19d6a0432e5d103823c45938f4a4edfa65239fd4:tests/services/test_storage.py:ipv4:176
+3f3067bbd4f38ad6ce996ef50e57f624a591b3e0:tests/services/dicom/test_c_store.py:ipv4:15
+3f3067bbd4f38ad6ce996ef50e57f624a591b3e0:tests/services/dicom/test_c_store.py:ipv4:48
+b7f3d0e812393af74600b5c0e8374d89066384d7:src/services/storage.py:ipv4:72
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:src/services/storage.py:ipv4:74
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/integration/test_c_store_saves_metadata.py:ipv4:21
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/integration/test_c_store_saves_metadata.py:ipv4:41
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/integration/test_c_store_saves_metadata.py:ipv4:65
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/services/test_storage.py:ipv4:37
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/services/test_storage.py:ipv4:40
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/services/test_storage.py:ipv4:52
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/services/test_storage.py:ipv4:55
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/services/test_storage.py:ipv4:88
+c7095ccbf4f5458f19746dd3ae69c9f5e3e63aa6:tests/services/test_storage.py:ipv4:107
+d9b882c916499c8cf2107879c83de2897fe99247:tests/integration/test_c_store_saves_metadata.py:ipv4:52
 
+# Source files
+src/services/storage.py:ipv4:89
 
-src/services/storage.py:ipv4:74
-tests/services/dicom/test_c_store.py:ipv4:15
-tests/services/dicom/test_c_store.py:ipv4:19
-tests/services/dicom/test_c_store.py:ipv4:48
-tests/services/dicom/test_c_store.py:ipv4:62
-tests/integration/test_c_store_saves_metadata.py:ipv4:21
+# Test files - test_c_store_saves_metadata.py
 tests/integration/test_c_store_saves_metadata.py:ipv4:23
 tests/integration/test_c_store_saves_metadata.py:ipv4:38
-tests/integration/test_c_store_saves_metadata.py:ipv4:41
-tests/integration/test_c_store_saves_metadata.py:ipv4:52
 tests/integration/test_c_store_saves_metadata.py:ipv4:55
-tests/integration/test_c_store_saves_metadata.py:ipv4:65
 tests/integration/test_c_store_saves_metadata.py:ipv4:73
 tests/integration/test_c_store_saves_metadata.py:ipv4:91
 tests/integration/test_c_store_saves_metadata.py:ipv4:108
-tests/services/test_storage.py:ipv4:37
-tests/services/test_storage.py:ipv4:40
-tests/services/test_storage.py:ipv4:52
-tests/services/test_storage.py:ipv4:55
-tests/services/test_storage.py:ipv4:88
-tests/services/test_storage.py:ipv4:107
-tests/services/test_storage.py:ipv4:136
-tests/services/test_storage.py:ipv4:176
 
+# Test files - test_pacs_storage_upload.py
+tests/integration/test_pacs_storage_upload.py:ipv4:21
+tests/integration/test_pacs_storage_upload.py:ipv4:25
+tests/integration/test_pacs_storage_upload.py:ipv4:29
+tests/integration/test_pacs_storage_upload.py:ipv4:30
+tests/integration/test_pacs_storage_upload.py:ipv4:32
+tests/integration/test_pacs_storage_upload.py:ipv4:36
+tests/integration/test_pacs_storage_upload.py:ipv4:45
+tests/integration/test_pacs_storage_upload.py:ipv4:46
+tests/integration/test_pacs_storage_upload.py:ipv4:48
+tests/integration/test_pacs_storage_upload.py:ipv4:52
+tests/integration/test_pacs_storage_upload.py:ipv4:62
+tests/integration/test_pacs_storage_upload.py:ipv4:65
+tests/integration/test_pacs_storage_upload.py:ipv4:66
+tests/integration/test_pacs_storage_upload.py:ipv4:67
+tests/integration/test_pacs_storage_upload.py:ipv4:69
+tests/integration/test_pacs_storage_upload.py:ipv4:72
+
+# Test files - test_c_store.py
+tests/services/dicom/test_c_store.py:ipv4:19
+tests/services/dicom/test_c_store.py:ipv4:62
+
+# Test files - test_dicom_uploader.py
+tests/services/dicom/test_dicom_uploader.py:ipv4:18
+tests/services/dicom/test_dicom_uploader.py:ipv4:28
+
+# Test files - test_upload_processor.py
+tests/services/dicom/test_upload_processor.py:ipv4:52
+tests/services/dicom/test_upload_processor.py:ipv4:58
+tests/services/dicom/test_upload_processor.py:ipv4:74
+tests/services/dicom/test_upload_processor.py:ipv4:89
+tests/services/dicom/test_upload_processor.py:ipv4:90
+tests/services/dicom/test_upload_processor.py:ipv4:91
+tests/services/dicom/test_upload_processor.py:ipv4:95
+tests/services/dicom/test_upload_processor.py:ipv4:108
+tests/services/dicom/test_upload_processor.py:ipv4:113
+tests/services/dicom/test_upload_processor.py:ipv4:131
+tests/services/dicom/test_upload_processor.py:ipv4:150
+tests/services/dicom/test_upload_processor.py:ipv4:174
+
+# Test files - test_storage.py
+tests/services/test_storage.py:ipv4:38
+tests/services/test_storage.py:ipv4:41
+tests/services/test_storage.py:ipv4:53
+tests/services/test_storage.py:ipv4:56
+tests/services/test_storage.py:ipv4:90
+tests/services/test_storage.py:ipv4:108

compose.yml

@@ -67,6 +67,29 @@ services:
       timeout: 5s
       retries: 3
 
+  upload:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    container_name: upload-service
+    command: ["uv", "run", "python", "-m", "upload_main"]
+    volumes:
+      - pacs-storage:/var/lib/pacs/storage
+      - pacs-db:/var/lib/pacs
+    environment:
+      - CLOUD_API_ENDPOINT=${CLOUD_API_ENDPOINT:-http://host.docker.internal:8000/api/dicom/upload/}
+      - PACS_DB_PATH=/var/lib/pacs/pacs.db
+      - PACS_STORAGE_PATH=/var/lib/pacs/storage
+      - MWL_DB_PATH=/var/lib/pacs/worklist.db
+      - UPLOAD_POLL_INTERVAL=2
+      - UPLOAD_BATCH_SIZE=10
+      - MAX_UPLOAD_RETRIES=3
+      - LOG_LEVEL=INFO
+    depends_on:
+      pacs:
+        condition: service_healthy
+    restart: unless-stopped
+
 volumes:
   pacs-storage:
     driver: local

docs/upload/README.md

@@ -0,0 +1,68 @@
+# Upload Service
+
+Background service that uploads stored DICOM images to the [Manage Breast Screening](https://github.com/NHSDigital/dtos-manage-breast-screening) cloud platform.
+
+## Overview
+
+The upload service:
+
+- Polls the PACS database for stored images pending upload
+- Uploads images to the Manage API as multipart form data
+- Links images to appointments via the `X-Source-Message-ID` header
+- Runs alongside the [PACS Server](../pacs/README.md) and [MWL Server](../mwl/README.md)
+
+## Architecture
+
+```
+┌─────────────────────────────────────────────────────────────┐
+│                    Upload Service                           │
+├─────────────────────────────────────────────────────────────┤
+│                                                             │
+│  ┌──────────────┐      ┌──────────────┐      ┌──────────┐   │
+│  │   Upload     │─────▶│   Upload     │─────▶│  DICOM   │   │
+│  │   Listener   │      │   Processor  │      │  Uploader│   │
+│  └──────────────┘      └──────────────┘      └──────────┘   │
+│         │                     │                    │        │
+│         │                     ▼                    │        │
+│         │              ┌──────────────┐            │        │
+│         │              │ PACS Storage │            │        │
+│         │              │ (files + DB) │            │        │
+│         │              └──────────────┘            │        │
+│         │                     │                    │        │
+│         │                     ▼                    ▼        │
+│         │              ┌──────────────┐     ┌────────────┐  │
+│         │              │ MWL Storage  │     │ Manage API │  │
+│         │              │ (worklist DB)│     │ (HTTP)     │  │
+│         └──────────────┴──────────────┘     └────────────┘  │
+└─────────────────────────────────────────────────────────────┘
+```
+
+### Workflow
+
+1. **Poll**: Listener polls PACS database every N seconds for pending uploads
+2. **Read**: Processor reads DICOM file from hash-based storage
+3. **Lookup**: Processor looks up `source_message_id` from MWL database via accession number
+4. **Upload**: Uploader sends multipart POST to cloud API with `X-Source-Message-ID` header
+5. **Update**: Processor updates upload status in PACS database (COMPLETE or retry)
+
+### Retry Behaviour
+
+- Failed uploads return to `PENDING` status for retry
+- After max retries (default 3), status becomes `FAILED`
+- Exponential backoff between batches when failures occur
+- Backoff starts at 1s, doubles each failure, caps at 60s
+
+## Configuration
+
+Environment variables:
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `CLOUD_API_ENDPOINT` | `http://localhost:8000/api/dicom/upload/` | Manage API upload endpoint |
+| `PACS_DB_PATH` | `/var/lib/pacs/pacs.db` | PACS SQLite database path |
+| `PACS_STORAGE_PATH` | `/var/lib/pacs/storage` | PACS file storage path |
+| `MWL_DB_PATH` | `/var/lib/pacs/worklist.db` | MWL SQLite database path |
+| `UPLOAD_POLL_INTERVAL` | `2` | Seconds between polling cycles |
+| `UPLOAD_BATCH_SIZE` | `10` | Max uploads per cycle |
+| `MAX_UPLOAD_RETRIES` | `3` | Retry attempts before permanent failure |
+| `LOG_LEVEL` | `INFO` | Logging level |

pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
   "pylibjpeg-openjpeg>=2.2.0",
   "pillow>=11.0.0",
   "websockets==15.0.1",
+  "requests>=2.32.0",
 ]
 
 [dependency-groups]

scripts/config/gitleaks.toml

@@ -29,6 +29,10 @@ paths = [
   '''^venv/''',
   '''^ENV/''',
   '''^env/''',
+
+  # Build/cache artifacts
+  '''__pycache__/''',
+  '''^htmlcov/''',
 ]
 
 regexes = [

src/services/dicom/dicom_uploader.py

@@ -0,0 +1,65 @@
+"""
+Cloud DICOM uploader
+
+Uploads DICOM files to the Manage Breast Screening HTTP API endpoint.
+"""
+
+import io
+import logging
+import os
+from typing import Optional
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class DICOMUploader:
+    def __init__(self, api_endpoint: str | None = None, timeout: int = 30, verify_ssl: bool = True):
+        self.api_endpoint = api_endpoint or os.getenv("CLOUD_API_ENDPOINT", "http://localhost:8000/api/dicom/upload/")
+        self.timeout = timeout
+        self.verify_ssl = verify_ssl
+
+    def upload_dicom(self, sop_instance_uid: str, dicom_bytes: bytes, action_id: Optional[str]) -> bool:
+        if not action_id:
+            logger.warning(f"No action_id for {sop_instance_uid}, upload will be rejected by server")
+
+        headers = {
+            "X-Source-Message-ID": action_id or "",
+        }
+
+        # Wrap bytes in BytesIO stream - Django expects a file-like object
+        file_stream = io.BytesIO(dicom_bytes)
+        files = {
+            "file": (f"{sop_instance_uid}.dcm", file_stream),
+        }
+
+        try:
+            logger.info(
+                f"Uploading {sop_instance_uid} to {self.api_endpoint} "
+                f"(size: {len(dicom_bytes)} bytes, action_id: {action_id})"
+            )
+
+            response = requests.post(
+                self.api_endpoint,
+                files=files,
+                headers=headers,
+                timeout=self.timeout,
+                verify=self.verify_ssl,
+            )
+
+            if response.status_code in (200, 201, 204):
+                logger.info(f"Successfully uploaded {sop_instance_uid} (status: {response.status_code})")
+                return True
+            else:
+                logger.error(
+                    f"Upload failed for {sop_instance_uid}: status {response.status_code}, body: {response.text}"
+                )
+                return False
+
+        except requests.exceptions.Timeout:
+            logger.error(f"Upload timeout for {sop_instance_uid} after {self.timeout}s")
+            return False
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Upload error for {sop_instance_uid}: {e}", exc_info=True)
+            return False

src/services/dicom/upload_listener.py

@@ -0,0 +1,56 @@
+"""
+DICOM upload listener service
+
+Background service that polls the PACS database for stored DICOM images
+and uploads them to the Manage Breast Screening HTTP API endpoint.
+"""
+
+import logging
+import time
+
+from services.dicom.upload_processor import UploadProcessor
+
+logger = logging.getLogger(__name__)
+
+
+class UploadListener:
+    def __init__(
+        self,
+        processor: UploadProcessor,
+        poll_interval: float = 2.0,
+        batch_size: int = 10,
+    ):
+        """
+        Initialize the upload listener.
+
+        Args:
+            processor: UploadProcessor instance for handling uploads
+            poll_interval: Base interval between polling cycles in seconds
+            batch_size: Maximum number of uploads to process per cycle
+        """
+        self.processor = processor
+        self.poll_interval = poll_interval
+        self.batch_size = batch_size
+        self._running = False
+
+    def start(self):
+        logger.info("Upload listener started")
+        self._running = True
+
+        while self._running:
+            try:
+                self.processor.process_batch(limit=self.batch_size)
+
+                # Add backoff delay to poll interval when experiencing failures
+                sleep_time = self.poll_interval + self.processor.backoff_delay
+                time.sleep(sleep_time)
+
+            except Exception as e:
+                logger.error(f"Error in upload listener: {e}", exc_info=True)
+                time.sleep(self.poll_interval)
+
+        logger.info("Upload listener stopped")
+
+    def stop(self):
+        logger.info("Stopping upload listener...")
+        self._running = False