650 anonymisation reads from queue

cli/src/pixl_cli/_message_processing.py

@@ -20,9 +20,9 @@
 
 import pandas as pd
 import tqdm
-from core.patient_queue._base import PixlBlockingInterface
-from core.patient_queue.message import Message
-from core.patient_queue.producer import PixlProducer
+from core.queue._base import PixlBlockingInterface
+from core.queue.models import ImagingRequestMessage
+from core.queue.producer import PixlProducer
 from decouple import config
 from loguru import logger
 
@@ -35,15 +35,15 @@
 
 def messages_from_df(
     df: pd.DataFrame,
-) -> list[Message]:
+) -> list[ImagingRequestMessage]:
     """
     Reads patient information from a DataFrame and transforms that into messages.
 
     :param messages_df: DataFrame containing patient information
     """
     messages = []
     for _, row in df.iterrows():
-        message = Message(
+        message = ImagingRequestMessage(
             mrn=row["mrn"],
             accession_number=row["accession_number"],
             study_uid=row["study_uid"],
@@ -129,6 +129,8 @@ def _message_count(queues_to_populate: list[str]) -> int:
     if "imaging-primary" in queues_to_populate:
         queues_to_count.add("imaging-secondary")
 
+    queues_to_count.add("anonymisation")
+
     messages_in_queues = 0
     for queue in queues_to_count:
         with PixlBlockingInterface(queue_name=queue, **SERVICE_SETTINGS["rabbitmq"]) as rabbitmq:
@@ -139,7 +141,7 @@ def _message_count(queues_to_populate: list[str]) -> int:
 
 def populate_queue_and_db(
     queues: list[str], messages_df: pd.DataFrame, messages_priority: int
-) -> list[Message]:
+) -> list[ImagingRequestMessage]:
     """
     Populate queues with messages,
     for imaging queue update the database and filter out exported or skipped studies.

cli/src/pixl_cli/main.py

@@ -23,7 +23,7 @@
 import click
 import requests
 from core.exports import ParquetExport
-from core.patient_queue.producer import PixlProducer
+from core.queue.producer import PixlProducer
 from core.telemetry import configure_logging, configure_tracing, telemetry_is_enabled
 from decouple import RepositoryEnv, UndefinedValueError
 from loguru import logger

cli/tests/conftest.py

@@ -24,8 +24,8 @@
 import pandas as pd
 import pytest
 from core.db.models import Base, Extract, Image
-from core.patient_queue.message import Message
-from core.patient_queue.producer import PixlProducer
+from core.queue.models import ImagingRequestMessage
+from core.queue.producer import PixlProducer
 from sqlalchemy import Engine, create_engine
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -136,8 +136,8 @@ def _make_message(
     accession_number: str,
     mrn: str,
     study_uid: str,
-) -> Message:
-    return Message(
+) -> ImagingRequestMessage:
+    return ImagingRequestMessage(
         project_name=project_name,
         accession_number=accession_number,
         mrn=mrn,
@@ -150,7 +150,7 @@ def _make_message(
 
 
 @pytest.fixture
-def example_messages() -> list[Message]:
+def example_messages() -> list[ImagingRequestMessage]:
     """Test input data."""
     return [
         _make_message(
@@ -174,7 +174,7 @@ def example_messages_df(example_messages):
 
 
 @pytest.fixture
-def example_messages_multiple_projects() -> list[Message]:
+def example_messages_multiple_projects() -> list[ImagingRequestMessage]:
     """Test input data."""
     return [
         _make_message(

cli/tests/test_message_processing.py

@@ -16,12 +16,17 @@
 
 import os
 from collections.abc import Generator
-from unittest.mock import Mock
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 from _pytest.monkeypatch import MonkeyPatch
-from core.patient_queue.producer import PixlProducer
-from pixl_cli._message_processing import retry_until_export_count_is_unchanged
+from core.queue.models import AnonymisationMessage
+from core.queue.producer import PixlProducer
+from pixl_cli._message_processing import (
+    _message_count,
+    retry_until_export_count_is_unchanged,
+)
+from pixl_imaging._orthanc import PIXLAnonOrthanc
 
 
 @pytest.fixture
@@ -98,3 +103,62 @@ def test_retry_with_image_exported_and_no_change_multiple_projects(
     )
 
     mock_publisher.assert_called_once()
+
+
+def test_message_count_includes_anonymisation(mocker) -> None:
+    """Checks that the anonymisation queue is included when counting messages."""
+    mock_rabbitmq = Mock()
+    mock_rabbitmq.message_count = 0
+
+    mock_interface = mocker.patch("pixl_cli._message_processing.PixlBlockingInterface")
+    mock_interface.return_value.__enter__.return_value = mock_rabbitmq
+
+    _message_count(["imaging-primary"])
+
+    queue_names = {call.kwargs["queue_name"] for call in mock_interface.call_args_list}
+
+    assert queue_names == {
+        "imaging-primary",
+        "imaging-secondary",
+        "anonymisation",
+    }
+
+
+@pytest.mark.asyncio
+async def test_notify_anon_publishes_anonymisation_message(monkeypatch) -> None:
+    """Checks that anonymisation requests are published to RabbitMQ."""
+    orthanc_raw = AsyncMock()
+    orthanc_raw.get_local_study.side_effect = [
+        {"MainDicomTags": {"StudyInstanceUID": "1.2.3"}},
+        {"MainDicomTags": {"StudyInstanceUID": "4.5.6"}},
+    ]
+
+    producer = Mock()
+    producer_context = Mock()
+    producer_context.__enter__ = Mock(return_value=producer)
+    producer_context.__exit__ = Mock(return_value=None)
+
+    monkeypatch.setattr(
+        "pixl_imaging._orthanc.AnonymisationProducer",
+        Mock(return_value=producer_context),
+    )
+
+    orthanc_anon = PIXLAnonOrthanc()
+
+    await orthanc_anon.notify_anon_to_retrieve_study_resources(
+        orthanc_raw=orthanc_raw,
+        resource_ids=["resource-1", "resource-2"],
+        series_uid="1.2.3.1\\1.2.3.2",
+        project_name="test project",
+    )
+
+    producer.publish.assert_called_once_with(
+        [
+            AnonymisationMessage(
+                resource_ids=["resource-1", "resource-2"],
+                series_uids=["1.2.3.1", "1.2.3.2"],
+                study_uids=["1.2.3", "4.5.6"],
+                project_name="test project",
+            )
+        ]
+    )

cli/tests/test_messages_from_files.py

@@ -20,7 +20,7 @@
 
 import pytest
 from core.db.models import Image
-from core.patient_queue.message import Message
+from core.queue.models import ImagingRequestMessage
 from pixl_cli._io import read_patient_info
 from pixl_cli._message_processing import messages_from_df, populate_queue_and_db
 
@@ -40,10 +40,10 @@ def test_messages_from_csv(omop_resources: Path) -> None:
     # Act
     messages = messages_from_df(messages_df)
     # Assert
-    assert all(isinstance(msg, Message) for msg in messages)
+    assert all(isinstance(msg, ImagingRequestMessage) for msg in messages)
 
     expected_messages = [
-        Message(
+        ImagingRequestMessage(
             procedure_occurrence_id=0,
             mrn="patient_identifier",
             accession_number="123456789",
@@ -71,7 +71,7 @@ def test_whitespace_and_na_processing(omop_resources: Path) -> None:
     messages = messages_from_df(messages_df)
     # Assert
     assert messages == [
-        Message(
+        ImagingRequestMessage(
             procedure_occurrence_id=0,
             mrn="patient_identifier",
             accession_number="123456789",
@@ -117,10 +117,10 @@ def test_messages_from_parquet(omop_resources: Path) -> None:
     # Act
     messages = messages_from_df(messages_df)
     # Assert
-    assert all(isinstance(msg, Message) for msg in messages)
+    assert all(isinstance(msg, ImagingRequestMessage) for msg in messages)
 
     expected_messages = [
-        Message(
+        ImagingRequestMessage(
             mrn="987654321",
             accession_number="AA12345601",
             study_uid="1.3.6.1.4.1.14519.5.2.1.99.1071.12985477682660597455732044031486",
@@ -130,7 +130,7 @@ def test_messages_from_parquet(omop_resources: Path) -> None:
             project_name="test-extract-uclh-omop-cdm",
             extract_generated_timestamp=datetime.datetime.fromisoformat("2023-12-07T14:08:58"),
         ),
-        Message(
+        ImagingRequestMessage(
             mrn="987654321",
             accession_number="AA12345605",
             study_uid="1.2.276.0.7230010.3.1.2.929116473.1.1710754859.579485",
@@ -157,10 +157,10 @@ def test_messages_from_batched_parquet(omop_resources: Path) -> None:
     # Act
     messages = messages_from_df(messages_df)
     # Assert
-    assert all(isinstance(msg, Message) for msg in messages)
+    assert all(isinstance(msg, ImagingRequestMessage) for msg in messages)
 
     expected_messages = [
-        Message(
+        ImagingRequestMessage(
             mrn="5020765",
             accession_number="MIG0234560",
             study_uid="1.2.840.114350.2.525.2.798268.2.110000014.1",
@@ -170,7 +170,7 @@ def test_messages_from_batched_parquet(omop_resources: Path) -> None:
             project_name="test-extract-uclh-omop-cdm",
             extract_generated_timestamp=datetime.datetime.fromisoformat("2023-12-07T14:08:58"),
         ),
-        Message(
+        ImagingRequestMessage(
             mrn="987654321",
             accession_number="ABC1234560",
             study_uid="1.2.840.114350.2.525.2.798268.2.190000013.1",
@@ -180,7 +180,7 @@ def test_messages_from_batched_parquet(omop_resources: Path) -> None:
             project_name="test-extract-uclh-omop-cdm",
             extract_generated_timestamp=datetime.datetime.fromisoformat("2023-12-07T14:08:58"),
         ),
-        Message(
+        ImagingRequestMessage(
             mrn="987654321",
             accession_number="AA12345601",
             study_uid="1.2.840.114350.2.525.2.798268.2.190000015.1",
@@ -190,7 +190,7 @@ def test_messages_from_batched_parquet(omop_resources: Path) -> None:
             project_name="test-extract-uclh-omop-cdm",
             extract_generated_timestamp=datetime.datetime.fromisoformat("2023-12-07T14:08:58"),
         ),
-        Message(
+        ImagingRequestMessage(
             mrn="987654321",
             accession_number="AA12345605",
             study_uid="1.2.840.114350.2.525.2.798268.2.190000016.1",
@@ -200,7 +200,7 @@ def test_messages_from_batched_parquet(omop_resources: Path) -> None:
             project_name="test-extract-uclh-omop-cdm",
             extract_generated_timestamp=datetime.datetime.fromisoformat("2023-12-07T14:08:58"),
         ),
-        Message(
+        ImagingRequestMessage(
             mrn="12345678",
             accession_number="12345678",
             study_uid="1.2.840.114350.2.525.2.798268.2.190000011.1",
@@ -210,7 +210,7 @@ def test_messages_from_batched_parquet(omop_resources: Path) -> None:
             project_name="test-extract-uclh-omop-cdm",
             extract_generated_timestamp=datetime.datetime.fromisoformat("2023-12-07T14:08:58"),
         ),
-        Message(
+        ImagingRequestMessage(
             mrn="12345678",
             accession_number="ABC1234567",
             study_uid="1.2.840.114350.2.525.2.798268.2.190000012.1",

cli/tests/test_populate.py

@@ -19,13 +19,13 @@
 
 import pixl_cli._message_processing
 from click.testing import CliRunner
-from core.patient_queue.producer import PixlProducer
+from core.queue.producer import PixlProducer
 from pixl_cli.main import populate
 
 if TYPE_CHECKING:
     from pathlib import Path
 
-    from core.patient_queue.message import Message
+    from core.queue.models import ImagingRequestMessage
 
 
 class MockProducer(PixlProducer):
@@ -39,7 +39,7 @@ def __exit__(self, *args: object, **kwargs) -> None:
         """Context exit point."""
         return
 
-    def publish(self, messages: list[Message], priority: int) -> None:  # noqa: ARG002 don't access messages or priority
+    def publish(self, messages: list[ImagingRequestMessage], priority: int) -> None:  # noqa: ARG002 don't access messages or priority
         """Dummy method for publish."""
         return
 

docker-compose.yml

@@ -95,6 +95,9 @@ services:
             args:
                 PIXL_PACKAGE_DIR: hasher
                 <<: *build-args-common
+        depends_on:
+            queue:
+                condition: service_healthy
         environment:
             <<: [*proxy-common, *pixl-common-env, *otel-common]
             OTEL_SERVICE_NAME: "hasher-api"
@@ -130,7 +133,7 @@ services:
         command: /run/secrets
         restart: always
         environment:
-            <<: [*pixl-db, *proxy-common, *pixl-common-env, *azure-keyvault, *otel-common]
+            <<: [*pixl-db, *proxy-common, *pixl-common-env, *azure-keyvault, *otel-common, *pixl-rabbit-mq]
             OTEL_SERVICE_NAME: "orthanc-anon"
             ORTHANC_NAME: "PIXL: Anon"
             ORTHANC_USERNAME: ${ORTHANC_ANON_USERNAME}
@@ -178,13 +181,15 @@ services:
         depends_on:
             postgres:
                 condition: service_healthy
+            queue:
+                condition: service_healthy
         healthcheck:
             test:
                 [
                     "CMD-SHELL",
                     "/probes/test-aliveness.py --user=$ORTHANC_USERNAME --pwd=$ORTHANC_PASSWORD",
                 ]
-            start_period: 10s
+            start_period: 90s
             retries: 10
             interval: 3s
             timeout: 2s
@@ -243,7 +248,7 @@ services:
                     "CMD-SHELL",
                     "/probes/test-aliveness.py --user=$ORTHANC_USERNAME --pwd=$ORTHANC_PASSWORD",
                 ]
-            start_period: 10s
+            start_period: 90s
             retries: 10
             interval: 3s
             timeout: 2s