diff --git a/cli/src/pixl_cli/_message_processing.py b/cli/src/pixl_cli/_message_processing.py index f7bb93b65..c31418766 100644 --- a/cli/src/pixl_cli/_message_processing.py +++ b/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,7 +35,7 @@ def messages_from_df( df: pd.DataFrame, -) -> list[Message]: +) -> list[ImagingRequestMessage]: """ Reads patient information from a DataFrame and transforms that into messages. @@ -43,7 +43,7 @@ def messages_from_df( """ 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. diff --git a/cli/src/pixl_cli/main.py b/cli/src/pixl_cli/main.py index 787e75a32..94a8af475 100644 --- a/cli/src/pixl_cli/main.py +++ b/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 diff --git a/cli/tests/conftest.py b/cli/tests/conftest.py index 582583c16..9281d9c85 100644 --- a/cli/tests/conftest.py +++ b/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( diff --git a/cli/tests/test_message_processing.py b/cli/tests/test_message_processing.py index a0391cc60..ef56b2479 100644 --- a/cli/tests/test_message_processing.py +++ b/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", + ) + ] + ) diff --git a/cli/tests/test_messages_from_files.py b/cli/tests/test_messages_from_files.py index 03ea8138b..18128ae47 100644 --- a/cli/tests/test_messages_from_files.py +++ b/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", diff --git a/cli/tests/test_populate.py b/cli/tests/test_populate.py index 20a5e4d98..0d9c9bf29 100644 --- a/cli/tests/test_populate.py +++ b/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 diff --git a/docker-compose.yml b/docker-compose.yml index a78df44f3..847580027 100644 --- a/docker-compose.yml +++ b/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 diff --git a/orthanc/orthanc-anon/plugin/pixl.py b/orthanc/orthanc-anon/plugin/pixl.py index 7d0640b84..a73a9d467 100644 --- a/orthanc/orthanc-anon/plugin/pixl.py +++ b/orthanc/orthanc-anon/plugin/pixl.py @@ -21,6 +21,7 @@ from __future__ import annotations +import asyncio import json import os import threading @@ -32,6 +33,7 @@ from typing import TYPE_CHECKING, cast from zipfile import ZipFile +import aio_pika import pydicom import requests from core.exceptions import PixlDiscardError, PixlSkipInstanceError @@ -40,13 +42,14 @@ record_study_deidentification_failure, ) from core.project_config.pixl_config_model import load_project_config +from core.queue.subscriber import AnonymisationPixlConsumer from core.telemetry import configure_logging, configure_metrics, configure_tracing from decouple import config from loguru import logger from opentelemetry import trace +from opentelemetry.instrumentation.aio_pika import AioPikaInstrumentor from opentelemetry.instrumentation.requests import RequestsInstrumentor from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor -from opentelemetry.propagate import extract from pixl_dcmd._database import engine as pixl_db_engine from pixl_dcmd._database import record_skip_reasons_for_study from pixl_dcmd.dicom_helpers import get_study_info @@ -65,6 +68,7 @@ from typing import Any from core.project_config.pixl_config_model import PixlConfig + from core.queue.models import AnonymisationMessage from opentelemetry.context import Context from pixl_dcmd.dicom_helpers import StudyInfo @@ -88,6 +92,10 @@ configure_tracing() SQLAlchemyInstrumentor().instrument(engine=pixl_db_engine) RequestsInstrumentor().instrument() +# orthanc-anon runs as a plugin inside Orthanc rather than via `opentelemetry-instrument`, +# so aio-pika isn't auto-instrumented and we need to do it explicitly to pick up the trace +# context propagated from the message publisher. +AioPikaInstrumentor().instrument() tracer = trace.get_tracer("pixl.orthanc_anon") configure_metrics() @@ -231,34 +239,70 @@ def OnHeartBeat(output, uri, **request) -> Any: # noqa: ARG001 output.AnswerBuffer("OK\n", "text/plain") -def ImportStudiesFromRaw(output, uri, **request): # noqa: ARG001 +async def process_anonymisation_message( + message: AnonymisationMessage, parent_context: Context +) -> None: """ Import studies from Orthanc Raw. - Offload to a thread pool executor to avoid blocking the Orthanc main thread. + Offload to the thread pool executor, so we don't block the event loop from consuming + further messages, but await completion so the message is only acked by + AnonymisationPixlConsumer once processing has actually finished. + + :param parent_context: Trace context extracted from the queue message headers by + AnonymisationPixlConsumer, to continue the trace from the message's publisher. """ - payload = json.loads(request["body"]) - study_resource_ids = payload["ResourceIDs"] - study_uids = payload["StudyInstanceUIDs"] - series_to_keep = payload["SeriesInstanceUIDs"] - project_name = payload["ProjectName"] - - # Extract the trace context injected into the request headers by the caller, and pass it to - # the thread pool job so the import continues the same trace - headers = {key.lower(): value for key, value in request.get("headers", {}).items()} - parent_context = extract(headers) - - executor.submit( + loop = asyncio.get_running_loop() + await loop.run_in_executor( + executor, _import_studies_from_raw, - study_resource_ids, - study_uids, - project_name, - series_to_keep, + message.resource_ids, + message.study_uids, + message.project_name, + message.series_uids, parent_context, ) - response = json.dumps({"Message": "Ok"}) - output.AnswerBuffer(response, "application/json") + +RABBITMQ_RECONNECT_DELAY_SECONDS = 5 + + +async def consume_anonymisation_queue() -> None: + """ + Consume anonymisation requests from RabbitMQ and submit them for processing. + + Runs for the lifetime of the process. Once connected, aio_pika's robust connection + automatically reconnects (and re-registers the consumer) if RabbitMQ becomes + unavailable, so we only need to retry here if the initial connection attempt fails + (e.g. RabbitMQ not quite ready yet at startup). + """ + while True: + try: + async with AnonymisationPixlConsumer( + queue_name="anonymisation", + callback=process_anonymisation_message, + ) as consumer: + await consumer.run() + # Keep this coroutine (and so the event loop) alive for the lifetime of + # the process, since messages are delivered via the running loop. + await asyncio.Event().wait() + except aio_pika.exceptions.AMQPConnectionError: + logger.exception( + "Anonymisation consumer failed to connect to RabbitMQ; retrying in {} seconds", + RABBITMQ_RECONNECT_DELAY_SECONDS, + ) + await asyncio.sleep(RABBITMQ_RECONNECT_DELAY_SECONDS) + + +def _run_anonymisation_consumer() -> None: + asyncio.run(consume_anonymisation_queue()) + + +consumer_thread = threading.Thread( + target=_run_anonymisation_consumer, + daemon=True, +) +consumer_thread.start() def _import_studies_from_raw( @@ -266,7 +310,7 @@ def _import_studies_from_raw( study_uids: list[str], project_name: str, series_to_keep: list[str], - parent_context: Context | None = None, + parent_context: Context, ) -> None: """ Import studies from Orthanc Raw. @@ -596,4 +640,3 @@ def notify_export_api_of_readiness(study_id: str, project_name: str) -> None: orthanc.RegisterOnChangeCallback(OnChange) orthanc.RegisterRestCallback("/heart-beat", OnHeartBeat) -orthanc.RegisterRestCallback("/import-from-raw", ImportStudiesFromRaw) diff --git a/pixl_core/src/core/patient_queue/__init__.py b/pixl_core/src/core/queue/__init__.py similarity index 85% rename from pixl_core/src/core/patient_queue/__init__.py rename to pixl_core/src/core/queue/__init__.py index e477b448e..b4c2aa3a2 100644 --- a/pixl_core/src/core/patient_queue/__init__.py +++ b/pixl_core/src/core/queue/__init__.py @@ -15,6 +15,6 @@ from __future__ import annotations -from .subscriber import PixlConsumer +from .subscriber import AnonymisationPixlConsumer, PixlConsumer -__all__ = ["PixlConsumer"] +__all__ = ["AnonymisationPixlConsumer", "PixlConsumer"] diff --git a/pixl_core/src/core/patient_queue/_base.py b/pixl_core/src/core/queue/_base.py similarity index 96% rename from pixl_core/src/core/patient_queue/_base.py rename to pixl_core/src/core/queue/_base.py index aa9c19922..0c28b76b7 100644 --- a/pixl_core/src/core/patient_queue/_base.py +++ b/pixl_core/src/core/queue/_base.py @@ -50,6 +50,10 @@ def __init__( self._channel: Any = None self._queue: Any = None + @property + def _url(self) -> str: + return f"amqp://{self._username}:{self._password}@{self._host}:{self._port}/" + class PixlBlockingInterface(PixlQueueInterface): def __enter__(self) -> Any: diff --git a/pixl_core/src/core/patient_queue/message.py b/pixl_core/src/core/queue/models.py similarity index 64% rename from pixl_core/src/core/patient_queue/message.py rename to pixl_core/src/core/queue/models.py index d7ed2d8f6..cb81b36b7 100644 --- a/pixl_core/src/core/patient_queue/message.py +++ b/pixl_core/src/core/queue/models.py @@ -11,7 +11,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""Classes to represent messages in the patient queue.""" +"""Classes to represent imaging and anonymisation messages in their respective queues.""" from __future__ import annotations @@ -27,8 +27,8 @@ @dataclass -class Message: - """Representation of a RabbitMQ message containing the information to identify a DICOM study.""" +class ImagingRequestMessage: + """Data containing the information to identify a DICOM study for an imaging request.""" mrn: str accession_number: str @@ -59,6 +59,35 @@ def serialise(self, *, deserialisable: bool = True) -> bytes: return str.encode(encode(self, unpicklable=deserialisable)) +@dataclass +class AnonymisationMessage: + """Data containing the information to identify an anonymisation request.""" + + resource_ids: list[str] + study_uids: list[str] + series_uids: list[str] + project_name: str + + @property + def identifier(self) -> str: + """Identifier for message""" + return (f"Message({self.resource_ids=} {self.study_uids=} {self.series_uids=}").replace( + "self.", "" + ) + + def serialise(self, *, deserialisable: bool = True) -> bytes: + """ + Serialise the message into a JSON string and convert to bytes. + + :param deserialisable: If True, the serialised message will be deserialisable, by setting + the unpicklable flag to False in jsonpickle.encode(), meaning that the original Message + object can be recovered by `deserialise()`. If False, calling `deserialise()` on the + serialised message will return a dictionary. + """ + logger.trace("Serialising {}", self) + return str.encode(encode(self, unpicklable=deserialisable)) + + def deserialise(serialised_msg: bytes) -> Any: """ Deserialise a message from a bytes-encoded JSON string. diff --git a/pixl_core/src/core/patient_queue/producer.py b/pixl_core/src/core/queue/producer.py similarity index 56% rename from pixl_core/src/core/patient_queue/producer.py rename to pixl_core/src/core/queue/producer.py index 78bf21c41..171e51e96 100644 --- a/pixl_core/src/core/patient_queue/producer.py +++ b/pixl_core/src/core/queue/producer.py @@ -24,15 +24,15 @@ from ._base import PixlBlockingInterface if TYPE_CHECKING: - from core.patient_queue.message import Message + from core.queue.modles import AnonymisationMessage, ImagingRequestMessage -tracer = trace.get_tracer("pixl_core.patient_queue.producer") +tracer = trace.get_tracer("pixl_core.queue.producer") class PixlProducer(PixlBlockingInterface): """Generic publisher for RabbitMQ""" - def publish(self, messages: list[Message], priority: int) -> None: + def publish(self, messages: list[ImagingRequestMessage], priority: int) -> None: """ Sends a list of serialised messages to a queue. :param messages: list of messages to be sent to queue @@ -53,7 +53,7 @@ def publish(self, messages: list[Message], priority: int) -> None: with tracer.start_as_current_span("publish_message", attributes=attributes): self._publish_message(msg, priority) - def _publish_message(self, message: Message, priority: int) -> None: + def _publish_message(self, message: ImagingRequestMessage, priority: int) -> None: """ Publish a single serialised message to a queue. :param message: message to be sent to queue @@ -88,3 +88,60 @@ def clear_queue(self) -> None: clean after tests. """ self._channel.queue_purge(queue=self.queue_name) + + +class AnonymisationProducer(PixlBlockingInterface): + """Anonymisation publisher for RabbitMQ""" + + def publish(self, messages: list[AnonymisationMessage]) -> None: + """ + Sends a list of serialised messages to a queue. + :param messages: list of messages to be sent to queue + """ + if len(messages) == 0: + logger.warning("List of messages is empty so nothing will be published to queue.") + return + + logger.info("Publishing {} messages to queue: {}", len(messages), self.queue_name) + for msg in messages: + attributes = { + "project_name": msg.project_name, + "resource_ids": msg.resource_ids, + "series_uids": msg.series_uids, + "study_uids": msg.study_uids, + } + with tracer.start_as_current_span("publish_message", attributes=attributes): + self._publish_message(msg) + + def _publish_message(self, message: AnonymisationMessage) -> None: + """ + Publish a single serialised message to a queue. + :param message: message to be sent to queue + """ + serialised_msg = message.serialise() + self._channel.basic_publish( + exchange="", + routing_key=self.queue_name, + body=serialised_msg, + properties=BasicProperties( + delivery_mode=DeliveryMode.Persistent, + ), + ) + + logger.bind( + project_name=message.project_name, + resource_id=message.resource_ids, + series_uid=message.series_uids, + study_uid=message.study_uids, + ).debug( + "AnonymisationMessage {} published to queue {}", + message, + self.queue_name, + ) + + def clear_queue(self) -> None: + """ + Triggering a purge of all the messages currently in the queue. Mainly used to + clean after tests. + """ + self._channel.queue_purge(queue=self.queue_name) diff --git a/pixl_core/src/core/patient_queue/subscriber.py b/pixl_core/src/core/queue/subscriber.py similarity index 56% rename from pixl_core/src/core/patient_queue/subscriber.py rename to pixl_core/src/core/queue/subscriber.py index 62708efb9..0d5bba5cb 100644 --- a/pixl_core/src/core/patient_queue/subscriber.py +++ b/pixl_core/src/core/queue/subscriber.py @@ -21,6 +21,7 @@ import aio_pika from decouple import config +from opentelemetry.context import get_current from core.exceptions import ( PixlDiscardError, @@ -28,23 +29,23 @@ PixlRequeueMessageError, PixlStudyNotInPrimaryArchiveError, ) -from core.patient_queue._base import PixlQueueInterface -from core.patient_queue.message import deserialise -from core.patient_queue.producer import PixlProducer +from core.queue._base import PixlQueueInterface +from core.queue.models import AnonymisationMessage, ImagingRequestMessage, deserialise +from core.queue.producer import PixlProducer if TYPE_CHECKING: from collections.abc import Awaitable, Callable from typing import Self from aio_pika.abc import AbstractIncomingMessage + from opentelemetry.context import Context - from core.patient_queue.message import Message from core.token_buffer.tokens import TokenBucket from loguru import logger -class PixlConsumer(PixlQueueInterface): +class PixlConsumer[PixlMessage: ImagingRequestMessage](PixlQueueInterface): """Connector to RabbitMQ. Consumes messages from a queue""" def __init__( @@ -52,7 +53,7 @@ def __init__( queue_name: str, token_bucket: TokenBucket, token_bucket_key: str, - callback: Callable[[Message], Awaitable[None]], + callback: Callable[[PixlMessage], Awaitable[None]], ) -> None: """ Creating connection to RabbitMQ queue @@ -61,11 +62,7 @@ def __init__( super().__init__(queue_name=queue_name) self.token_bucket = token_bucket self.token_bucket_key = token_bucket_key - self._callback = callback - - @property - def _url(self) -> str: - return f"amqp://{self._username}:{self._password}@{self._host}:{self._port}/" + self._callback: Callable[[PixlMessage], Awaitable[None]] = callback async def __aenter__(self) -> Self: """Establishes connection to queue.""" @@ -88,7 +85,7 @@ async def _process_message(self, message: AbstractIncomingMessage) -> None: await message.reject(requeue=True) return - pixl_message: Message = deserialise(message.body) + pixl_message: PixlMessage = deserialise(message.body) logger.debug("Picked up from queue: {}", pixl_message.identifier) try: await self._callback(pixl_message) @@ -142,3 +139,74 @@ async def run(self) -> None: async def __aexit__(self, *args: object, **kwargs: Any) -> None: """Requirement for the asynchronous context manager""" + + +class AnonymisationPixlConsumer(PixlQueueInterface): + """Connector to RabbitMQ. Consumes messages from anonymisation queue""" + + def __init__( + self, + queue_name: str, + callback: Callable[[AnonymisationMessage, Context], Awaitable[None]], + ) -> None: + """Creating connection to RabbitMQ queue""" + super().__init__(queue_name=queue_name) + self._callback: Callable[[AnonymisationMessage, Context], Awaitable[None]] = callback + + async def __aenter__(self) -> Self: + """Establishes connection to queue.""" + self._connection = await aio_pika.connect_robust(self._url) + self._channel = await self._connection.channel() + # Set number of messages in flight + max_in_flight = config("PIXL_MAX_MESSAGES_IN_FLIGHT", cast=int) + logger.info("Pika will consume up to {} messages concurrently", max_in_flight) + await self._channel.set_qos(prefetch_count=max_in_flight) + self._queue = await self._channel.declare_queue( + self.queue_name, + durable=True, + arguments={"x-max-priority": 5}, + ) + return self + + async def _process_message(self, message: AbstractIncomingMessage) -> None: + pixl_message: AnonymisationMessage = deserialise(message.body) + # AioPikaInstrumentor wraps this callback and extracts the trace context from the + # message headers into the current context, so we just need to read it back here. + parent_context = get_current() + logger.debug("Picked up from queue: {}", pixl_message.identifier) + try: + # Awaiting the callback here (rather than firing-and-forgetting the work) means + # the message is only acked once processing has actually finished, so a crash + # part-way through doesn't silently lose the message. + await self._callback(pixl_message, parent_context) + except PixlRequeueMessageError as requeue: + logger.trace("Requeue message: {} from {}", pixl_message.identifier, requeue) + await asyncio.sleep(1) + await message.reject(requeue=True) + except PixlOutOfHoursError as nack_requeue: + logger.trace( + "Nack and requeue message: {} from {}", pixl_message.identifier, nack_requeue + ) + await asyncio.sleep(10) + await message.nack(requeue=True) + except PixlDiscardError as exception: + logger.warning("Failed message {}: {}", pixl_message.identifier, exception) + # ack so that we can see rate of message processing in rabbitmq admin + await message.ack() + except Exception: # noqa: BLE001 + logger.exception( + "Failed to process {}. Not re-queuing message", + pixl_message.identifier, + ) + # ack so that we can see rate of message processing in rabbitmq admin + await message.ack() + else: + logger.success("Finished message {}", pixl_message.identifier) + await message.ack() + + async def run(self) -> None: + """Processes messages from queue asynchronously.""" + await self._queue.consume(self._process_message) + + async def __aexit__(self, *args: object, **kwargs: Any) -> None: + """Requirement for the asynchronous context manager""" diff --git a/pixl_core/tests/conftest.py b/pixl_core/tests/conftest.py index 324ce88e5..e6ced2255 100644 --- a/pixl_core/tests/conftest.py +++ b/pixl_core/tests/conftest.py @@ -19,6 +19,7 @@ import shlex from pathlib import Path from typing import TYPE_CHECKING +from unittest.mock import AsyncMock, Mock import pytest import requests @@ -36,11 +37,12 @@ from core.db.models import Base, Extract, Image from core.logging import OTelSink -from core.patient_queue.message import Message +from core.queue.models import AnonymisationMessage, ImagingRequestMessage +from core.queue.subscriber import AnonymisationPixlConsumer if TYPE_CHECKING: import subprocess - from collections.abc import Generator + from collections.abc import Callable, Generator pytest_plugins = "pytest_pixl" @@ -220,9 +222,9 @@ def export_dir(tmp_path_factory: pytest.TempPathFactory) -> pathlib.Path: @pytest.fixture -def mock_message() -> Message: +def mock_message() -> ImagingRequestMessage: """An example Message used for testing""" - return Message( + return ImagingRequestMessage( mrn="111", accession_number="123", study_uid="1.2.3", @@ -236,6 +238,41 @@ def mock_message() -> Message: ) +@pytest.fixture +def mock_anon_message() -> AnonymisationMessage: + """An example AnonymisationMessage used for testing""" + return AnonymisationMessage( + resource_ids=["resource-1", "resource-2"], + study_uids=["1.2.3", "4.5.6"], + series_uids=["1.2.3.1", "1.2.3.2"], + project_name="test project", + ) + + +@pytest.fixture +def mock_incoming_message() -> Callable[..., Mock]: + """Factory for a mock aio_pika incoming message with async ack/nack/reject.""" + + def _make(body: bytes, priority: int = 1) -> Mock: + message = Mock(body=body, priority=priority) + message.reject = AsyncMock() + message.ack = AsyncMock() + message.nack = AsyncMock() + return message + + return _make + + +@pytest.fixture +def anon_consumer() -> Callable[..., AnonymisationPixlConsumer]: + """Factory for an AnonymisationPixlConsumer, without connecting to a broker.""" + + def _make(queue_name: str, callback: Mock) -> AnonymisationPixlConsumer: + return AnonymisationPixlConsumer(queue_name=queue_name, callback=callback) + + return _make + + @pytest.fixture def log_exporter() -> InMemoryLogRecordExporter: """In-memory exporter capturing the OTel log records the sink emits.""" diff --git a/pixl_core/tests/patient_queue/test_subscriber.py b/pixl_core/tests/patient_queue/test_subscriber.py deleted file mode 100644 index f0b8f0e59..000000000 --- a/pixl_core/tests/patient_queue/test_subscriber.py +++ /dev/null @@ -1,60 +0,0 @@ -# Copyright (c) 2022 University College London Hospitals NHS Foundation Trust -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -from __future__ import annotations - -import asyncio -from unittest.mock import AsyncMock - -import pytest - -from core.patient_queue.producer import PixlProducer -from core.patient_queue.subscriber import PixlConsumer -from core.token_buffer.tokens import TokenBucket - -TEST_QUEUE = "test_consume" - - -class ExpectedTestError(Exception): - """Expected error for testing.""" - - -@pytest.mark.asyncio -@pytest.mark.usefixtures("run_containers") -@pytest.mark.xfail( - reason="Sanity check that async test gets run", strict=True, raises=ExpectedTestError -) -async def test_create(mock_message) -> None: - """Checks consume is working.""" - with PixlProducer(queue_name=TEST_QUEUE) as producer: - producer.publish(messages=[mock_message], priority=1) - - consume = AsyncMock() - async with PixlConsumer( - queue_name=TEST_QUEUE, - token_bucket=TokenBucket(), - token_bucket_key="primary", # noqa: S106 - callback=consume, - ) as consumer: - # Create a Task to run pc.run in the background - task = asyncio.create_task(consumer.run()) - # Wait for a short time to allow pc.run to start - await asyncio.sleep(1) - # Cancel before assertion so the task doesn't hang - task.cancel() - # need to close the connection and channel - await consumer._channel.close() - await consumer._connection.close() - consume.assert_called_once() - # Fail on purpose to check async test awaited - raise ExpectedTestError diff --git a/pixl_core/tests/patient_queue/test_message.py b/pixl_core/tests/queue/test_models.py similarity index 60% rename from pixl_core/tests/patient_queue/test_message.py rename to pixl_core/tests/queue/test_models.py index cb5b22c70..32737fd2d 100644 --- a/pixl_core/tests/patient_queue/test_message.py +++ b/pixl_core/tests/queue/test_models.py @@ -13,11 +13,11 @@ # limitations under the License. from __future__ import annotations -from core.patient_queue.message import deserialise +from core.queue.models import deserialise -def test_serialise(mock_message) -> None: - """Checks that messages can be correctly serialised""" +def test_serialise_imagingrequests(mock_message) -> None: + """Checks that imaging request messages can be correctly serialised""" msg_body = mock_message.serialise(deserialisable=False) assert ( msg_body == b'{"mrn": "111", "accession_number": "123", "study_uid": "1.2.3", ' @@ -29,7 +29,24 @@ def test_serialise(mock_message) -> None: ) +def test_serialise_anon(mock_anon_message) -> None: + """Checks that anon messages can be correctly serialised""" + msg_body = mock_anon_message.serialise(deserialisable=False) + assert ( + msg_body == b'{"resource_ids": ["resource-1", "resource-2"], ' + b'"study_uids": ["1.2.3", "4.5.6"], ' + b'"series_uids": ["1.2.3.1", "1.2.3.2"], ' + b'"project_name": "test project"}' + ) + + def test_deserialise(mock_message) -> None: """Checks if deserialised messages are the same as the original""" serialised_msg = mock_message.serialise() assert deserialise(serialised_msg) == mock_message + + +def test_deserialise_anon(mock_anon_message) -> None: + """Checks if deserialised anon messages are the same as the original""" + serialised_msg = mock_anon_message.serialise() + assert deserialise(serialised_msg) == mock_anon_message diff --git a/pixl_core/tests/patient_queue/test_producer.py b/pixl_core/tests/queue/test_producer.py similarity index 62% rename from pixl_core/tests/patient_queue/test_producer.py rename to pixl_core/tests/queue/test_producer.py index 619e951a2..c80d170ff 100644 --- a/pixl_core/tests/patient_queue/test_producer.py +++ b/pixl_core/tests/queue/test_producer.py @@ -15,9 +15,10 @@ import pytest -from core.patient_queue.producer import PixlProducer +from core.queue.producer import AnonymisationProducer, PixlProducer TEST_QUEUE = "test_publish" +TEST_QUEUE_ANON = "test_anon_publish" @pytest.mark.usefixtures("run_containers") @@ -39,3 +40,24 @@ def test_publish(mock_message) -> None: with PixlProducer(queue_name=TEST_QUEUE) as pp: assert pp.message_count == 1 + + +@pytest.mark.usefixtures("run_containers") +def test_create_pixl_producer_anon() -> None: + """Checks that AnonymisationProducer can be instantiated.""" + with AnonymisationProducer(queue_name=TEST_QUEUE_ANON) as pp: + assert pp.connection_open + + +@pytest.mark.usefixtures("run_containers") +def test_publish_anon(mock_anon_message) -> None: + """ + Checks that after publishing, there is one message in the queue. + Will only work if nothing has been added to queue before. + """ + with AnonymisationProducer(queue_name=TEST_QUEUE_ANON) as pp: + pp.clear_queue() + pp.publish(messages=[mock_anon_message]) + + with AnonymisationProducer(queue_name=TEST_QUEUE_ANON) as pp: + assert pp.message_count == 1 diff --git a/pixl_core/tests/queue/test_subscriber.py b/pixl_core/tests/queue/test_subscriber.py new file mode 100644 index 000000000..dafcdc0e6 --- /dev/null +++ b/pixl_core/tests/queue/test_subscriber.py @@ -0,0 +1,165 @@ +# Copyright (c) 2022 University College London Hospitals NHS Foundation Trust +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from __future__ import annotations + +import asyncio +from unittest.mock import ANY, AsyncMock, Mock + +import pytest + +from core.exceptions import ( + PixlDiscardError, + PixlOutOfHoursError, + PixlRequeueMessageError, +) +from core.queue.producer import AnonymisationProducer, PixlProducer +from core.queue.subscriber import AnonymisationPixlConsumer, PixlConsumer +from core.token_buffer.tokens import TokenBucket + +TEST_QUEUE = "test_consume" +TEST_QUEUE_ANON = "test_anon_consume" + + +class ExpectedTestError(Exception): + """Expected error for testing.""" + + +# Shared by both PixlConsumer and AnonymisationPixlConsumer error-handling tests below, +# so the two consumers' behaviour for a given error can't silently drift apart. +ERROR_HANDLING_CASES = [ + pytest.param(PixlRequeueMessageError, "reject", {"requeue": True}, id="requeue"), + pytest.param(PixlOutOfHoursError, "nack", {"requeue": True}, id="out_of_hours"), + pytest.param(PixlDiscardError, "ack", {}, id="discard"), + pytest.param(ExpectedTestError, "ack", {}, id="unexpected"), +] + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("run_containers") +@pytest.mark.xfail( + reason="Sanity check that async test gets run", strict=True, raises=ExpectedTestError +) +async def test_create(mock_message) -> None: + """Checks consume is working.""" + with PixlProducer(queue_name=TEST_QUEUE) as producer: + producer.publish(messages=[mock_message], priority=1) + + consume = AsyncMock() + async with PixlConsumer( + queue_name=TEST_QUEUE, + token_bucket=TokenBucket(), + token_bucket_key="primary", # noqa: S106 + callback=consume, + ) as consumer: + # Create a Task to run pc.run in the background + task = asyncio.create_task(consumer.run()) + # Wait for a short time to allow pc.run to start + await asyncio.sleep(1) + # Cancel before assertion so the task doesn't hang + task.cancel() + # need to close the connection and channel + await consumer._channel.close() + await consumer._connection.close() + consume.assert_called_once() + # Fail on purpose to check async test awaited + raise ExpectedTestError + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("run_containers") +async def test_run_anon(mock_anon_message) -> None: + """Checks that the consumer starts consuming messages.""" + with AnonymisationProducer(queue_name=TEST_QUEUE_ANON) as producer: + producer.publish(messages=[mock_anon_message]) + + callback = AsyncMock() + async with AnonymisationPixlConsumer( + queue_name=TEST_QUEUE_ANON, + callback=callback, + ) as consumer: + # Create a Task to run consumer.run in the background + task = asyncio.create_task(consumer.run()) + # Wait for a short time to allow consumer.run to start and pick up the message + await asyncio.sleep(1) + # Cancel before assertion so the task doesn't hang + task.cancel() + # need to close the connection and channel + await consumer._channel.close() + await consumer._connection.close() + callback.assert_called_once_with(mock_anon_message, ANY) + + +@pytest.mark.asyncio +async def test_process_message_anon( + mock_anon_message, mock_incoming_message, anon_consumer +) -> None: + """Checks that a received message is passed to the callback and acked.""" + callback = AsyncMock() + consumer = anon_consumer(TEST_QUEUE_ANON, callback) + message = mock_incoming_message(mock_anon_message.serialise()) + + await consumer._process_message(message) + + callback.assert_awaited_once_with(mock_anon_message, ANY) + message.ack.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("error", "method_name", "expected_kwargs"), ERROR_HANDLING_CASES) +async def test_process_message_error_handling( # noqa: PLR0913 + monkeypatch, + mock_message, + mock_incoming_message, + error, + method_name, + expected_kwargs, +) -> None: + """Each error type from the callback results in the correct ack/nack/reject call.""" + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + callback = AsyncMock(side_effect=error) + token_bucket = Mock(has_token=Mock(return_value=True)) + + consumer = PixlConsumer( + queue_name=TEST_QUEUE, + token_bucket=token_bucket, + token_bucket_key="primary", # noqa: S106 + callback=callback, + ) + message = mock_incoming_message(mock_message.serialise()) + + await consumer._process_message(message) + + getattr(message, method_name).assert_awaited_once_with(**expected_kwargs) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("error", "method_name", "expected_kwargs"), ERROR_HANDLING_CASES) +async def test_process_message_anon_error_handling( # noqa: PLR0913 + monkeypatch, + mock_anon_message, + mock_incoming_message, + anon_consumer, + error, + method_name, + expected_kwargs, +) -> None: + """Each error type from the callback results in the correct ack/nack/reject call.""" + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + callback = AsyncMock(side_effect=error) + consumer = anon_consumer(TEST_QUEUE_ANON, callback) + message = mock_incoming_message(mock_anon_message.serialise()) + + await consumer._process_message(message) + + getattr(message, method_name).assert_awaited_once_with(**expected_kwargs) diff --git a/pixl_imaging/README.md b/pixl_imaging/README.md index 9e8a09172..c2c7c51c5 100644 --- a/pixl_imaging/README.md +++ b/pixl_imaging/README.md @@ -9,7 +9,7 @@ The imaging API has two queues: - `imaging-secondary`, for querying PACS The imaging API uses RabbitMQ to expose a single HTTP endpoint that expects a JSON-formatted message structured as -defined by the [`Message`](../pixl_core/src/core/patient_queue/message.py) class in `pixl_core/patient_queue`. +defined by the [`ImagingRequestMessage`](../pixl_core/src/core/queue/message.py) class in `pixl_core/queue`. Users should send messages to the `imaging-primary` queue only. On arrival of the input message, the imaging API will query the VNA for the requested study. If the study does not exist in the VNA, the input message will be sent diff --git a/pixl_imaging/src/pixl_imaging/_orthanc.py b/pixl_imaging/src/pixl_imaging/_orthanc.py index b87305364..9f208fc87 100644 --- a/pixl_imaging/src/pixl_imaging/_orthanc.py +++ b/pixl_imaging/src/pixl_imaging/_orthanc.py @@ -19,6 +19,8 @@ import aiohttp from core.exceptions import PixlDiscardError, PixlRequeueMessageError +from core.queue.models import AnonymisationMessage +from core.queue.producer import AnonymisationProducer from decouple import config from loguru import logger @@ -300,12 +302,12 @@ async def notify_anon_to_retrieve_study_resources( series_uids = series_uid.split("\\") if series_uid else [] - await self._post( - path="/import-from-raw", - data={ - "ResourceIDs": resource_ids, - "StudyInstanceUIDs": study_uids, - "SeriesInstanceUIDs": series_uids, - "ProjectName": project_name, - }, + message = AnonymisationMessage( + resource_ids=resource_ids, + series_uids=series_uids, + study_uids=study_uids, + project_name=project_name, ) + + with AnonymisationProducer(queue_name="anonymisation") as producer: + producer.publish([message]) diff --git a/pixl_imaging/src/pixl_imaging/_processing.py b/pixl_imaging/src/pixl_imaging/_processing.py index 9ed26d022..0d768e665 100644 --- a/pixl_imaging/src/pixl_imaging/_processing.py +++ b/pixl_imaging/src/pixl_imaging/_processing.py @@ -25,7 +25,7 @@ from pixl_imaging._orthanc import Orthanc, PIXLAnonOrthanc, PIXLRawOrthanc if TYPE_CHECKING: - from core.patient_queue.message import Message + from core.queue.models import ImagingRequestMessage from loguru import logger @@ -35,7 +35,7 @@ class DicomModality(StrEnum): secondary = config("SECONDARY_DICOM_SOURCE_MODALITY") -async def process_message(message: Message, archive: DicomModality) -> None: +async def process_message(message: ImagingRequestMessage, archive: DicomModality) -> None: """ Process message from queue by retrieving a study with the given Patient and Accession Number. We may receive multiple messages with same Patient + Acc Num, either as retries or because @@ -391,10 +391,10 @@ async def _get_missing_instances( class ImagingStudy: """Dataclass for DICOM study unique to a patient and imaging study""" - message: Message + message: ImagingRequestMessage @classmethod - def from_message(cls, message: Message) -> ImagingStudy: + def from_message(cls, message: ImagingRequestMessage) -> ImagingStudy: """Build an imaging study from a queue message.""" return ImagingStudy(message=message) diff --git a/pixl_imaging/src/pixl_imaging/main.py b/pixl_imaging/src/pixl_imaging/main.py index f664a03eb..cee0448db 100644 --- a/pixl_imaging/src/pixl_imaging/main.py +++ b/pixl_imaging/src/pixl_imaging/main.py @@ -18,7 +18,8 @@ import asyncio import importlib.metadata -from core.patient_queue.subscriber import PixlConsumer +from core.queue.models import ImagingRequestMessage +from core.queue.subscriber import PixlConsumer from core.rest_api.router import router, state from core.telemetry import configure_logging from decouple import config @@ -55,13 +56,13 @@ async def startup_event() -> None: """ background_tasks = set() async with ( - PixlConsumer( + PixlConsumer[ImagingRequestMessage]( QUEUE_NAME, token_bucket=state.token_bucket, token_bucket_key="primary", # noqa: S106 callback=lambda message: process_message(message, archive=DicomModality.primary), ) as primary_consumer, - PixlConsumer( + PixlConsumer[ImagingRequestMessage]( SECONDARY_QUEUE_NAME, token_bucket=state.token_bucket, token_bucket_key="secondary", # noqa: S106 diff --git a/pixl_imaging/tests/test_imaging_processing.py b/pixl_imaging/tests/test_imaging_processing.py index 962fcd8dc..e77cb006c 100644 --- a/pixl_imaging/tests/test_imaging_processing.py +++ b/pixl_imaging/tests/test_imaging_processing.py @@ -23,7 +23,7 @@ import pytest from core.exceptions import PixlDiscardError, PixlOutOfHoursError, PixlStudyNotInPrimaryArchiveError -from core.patient_queue.message import Message +from core.queue.models import ImagingRequestMessage from decouple import config from pydicom import dcmread from pydicom.data import get_testdata_file @@ -58,9 +58,9 @@ @pytest.fixture(scope="module") -def message() -> Message: - """A Message with a valid study_uid.""" - return Message( +def message() -> ImagingRequestMessage: + """A ImagingRequestMessage with a valid study_uid.""" + return ImagingRequestMessage( mrn=PATIENT_ID, accession_number=ACCESSION_NUMBER, study_uid=STUDY_UID, @@ -75,9 +75,9 @@ def message() -> Message: @pytest.fixture(scope="module") -def message_with_series_uids() -> Message: - """A Message for querying a subset of series within a study.""" - return Message( +def message_with_series_uids() -> ImagingRequestMessage: + """A ImagingRequestMessage for querying a subset of series within a study.""" + return ImagingRequestMessage( mrn=PATIENT_ID, accession_number=ACCESSION_NUMBER, study_uid=STUDY_UID, @@ -92,9 +92,9 @@ def message_with_series_uids() -> Message: @pytest.fixture(scope="module") -def message_with_single_series_uid() -> Message: - """A Message for querying a subset of series within a study.""" - return Message( +def message_with_single_series_uid() -> ImagingRequestMessage: + """A ImagingRequestMessage for querying a subset of series within a study.""" + return ImagingRequestMessage( mrn=PATIENT_ID, accession_number=ACCESSION_NUMBER, study_uid=STUDY_UID, @@ -109,9 +109,9 @@ def message_with_single_series_uid() -> Message: @pytest.fixture(scope="module") -def no_uid_message() -> Message: - """A Message with a valid study_uid.""" - return Message( +def no_uid_message() -> ImagingRequestMessage: + """A ImagingRequestMessage with a valid study_uid.""" + return ImagingRequestMessage( mrn=PATIENT_ID, accession_number=ACCESSION_NUMBER, study_uid="", @@ -126,9 +126,12 @@ def no_uid_message() -> Message: @pytest.fixture(scope="module") -def pacs_message() -> Message: - """A Message with a valid study_uid for a study that exists in PACS but not VNA.""" - return Message( +def pacs_message() -> ImagingRequestMessage: + """ + A ImagingRequestMessage with a valid study_uid for a study + that exists in PACS but not VNA. + """ + return ImagingRequestMessage( mrn=PACS_PATIENT_ID, accession_number=PACS_ACCESSION_NUMBER, study_uid=PACS_STUDY_UID, @@ -143,9 +146,12 @@ def pacs_message() -> Message: @pytest.fixture(scope="module") -def pacs_no_uid_message() -> Message: - """A Message without a valid study_uid for a study that exists in PACS but not the VNA.""" - return Message( +def pacs_no_uid_message() -> ImagingRequestMessage: + """ + A ImagingRequestMessage without a valid study_uid for a study + that exists in PACS but not the VNA. + """ + return ImagingRequestMessage( mrn=PACS_PATIENT_ID, accession_number=PACS_ACCESSION_NUMBER, study_uid="ialsodontexist", @@ -160,9 +166,9 @@ def pacs_no_uid_message() -> Message: @pytest.fixture(scope="module") -def missing_message() -> Message: - """A Message for a study that does not exist in PACS nor the VNA.""" - return Message( +def missing_message() -> ImagingRequestMessage: + """A ImagingRequestMessage for a study that does not exist in PACS nor the VNA.""" + return ImagingRequestMessage( mrn=MISSING_PATIENT_ID, accession_number=MISSING_ACCESSION_NUMBER, study_uid=MISSING_STUDY_UID, @@ -272,7 +278,7 @@ async def orthanc_raw(run_containers) -> PIXLRawOrthanc: @pytest.mark.processing @pytest.mark.asyncio @pytest.mark.usefixtures("_add_image_to_fake_vna") -async def test_image_saved(orthanc_raw, message: Message) -> None: +async def test_image_saved(orthanc_raw, message: ImagingRequestMessage) -> None: """ Given the VNA has images, and orthanc raw has no images When we run process_message @@ -313,7 +319,7 @@ async def test_image_saved(orthanc_raw, message: Message) -> None: @pytest.mark.asyncio @pytest.mark.usefixtures("_add_image_to_fake_vna") async def test_message_with_series_uids( - orthanc_raw, message_with_series_uids: Message, caplog + orthanc_raw, message_with_series_uids: ImagingRequestMessage, caplog ) -> None: """ Given the VNA has a single study with 2 series, and Orthanc Raw has no images @@ -350,7 +356,7 @@ async def test_message_with_series_uids( @pytest.mark.asyncio @pytest.mark.usefixtures("_add_image_to_fake_vna") async def test_message_with_one_series_uid( - orthanc_raw, message_with_single_series_uid: Message, caplog + orthanc_raw, message_with_single_series_uid: ImagingRequestMessage, caplog ) -> None: """ Given the VNA has a single study with 2 series, and Orthanc Raw has no images @@ -382,7 +388,7 @@ async def test_message_with_one_series_uid( @pytest.mark.processing @pytest.mark.asyncio @pytest.mark.usefixtures("_add_image_to_fake_vna") -async def test_partial_retrieve(orthanc_raw, message: Message, caplog) -> None: +async def test_partial_retrieve(orthanc_raw, message: ImagingRequestMessage, caplog) -> None: """ Given the VNA has a single study with 2 instances, and orthanc raw has the same study with 1 instance @@ -424,7 +430,7 @@ async def test_partial_retrieve(orthanc_raw, message: Message, caplog) -> None: @pytest.mark.processing @pytest.mark.asyncio @pytest.mark.usefixtures("_add_image_to_fake_vna") -async def test_existing_message_sent_twice(orthanc_raw, message: Message) -> None: +async def test_existing_message_sent_twice(orthanc_raw, message: ImagingRequestMessage) -> None: """ Given the VNA has images, and orthanc raw has no images When we run process_message on the same message twice @@ -474,7 +480,9 @@ async def test_existing_message_sent_twice(orthanc_raw, message: Message) -> Non @pytest.mark.processing @pytest.mark.asyncio @pytest.mark.usefixtures("_add_image_to_fake_vna") -async def test_querying_without_uid(orthanc_raw, caplog, no_uid_message: Message) -> None: +async def test_querying_without_uid( + orthanc_raw, caplog, no_uid_message: ImagingRequestMessage +) -> None: """ Given a message with non-existent study_uid When we query the VNA @@ -516,7 +524,7 @@ def now(cls, tz=None) -> datetime.datetime: @pytest.mark.asyncio @pytest.mark.usefixtures("_add_image_to_fake_pacs") async def test_querying_pacs_with_uid( - orthanc_raw, caplog, monkeypatch, pacs_message: Message + orthanc_raw, caplog, monkeypatch, pacs_message: ImagingRequestMessage ) -> None: """ Given a message with study_uid exists in PACS but not VNA, @@ -556,7 +564,7 @@ async def test_querying_pacs_with_uid( @pytest.mark.asyncio @pytest.mark.usefixtures("_add_image_to_fake_pacs") async def test_querying_pacs_without_uid( - orthanc_raw, caplog, monkeypatch, pacs_no_uid_message: Message + orthanc_raw, caplog, monkeypatch, pacs_no_uid_message: ImagingRequestMessage ) -> None: """ Given a message with non-existent study_uid exists in PACS but not VNA, @@ -591,7 +599,9 @@ async def test_querying_pacs_without_uid( @pytest.mark.processing @pytest.mark.asyncio -async def test_querying_missing_image(orthanc_raw, monkeypatch, missing_message: Message) -> None: +async def test_querying_missing_image( + orthanc_raw, monkeypatch, missing_message: ImagingRequestMessage +) -> None: """ Given a message for a study that is missing in both the VNA and PACS, When we query the archives within the window of Monday-Friday 8pm to 8am, @@ -626,7 +636,7 @@ async def test_querying_missing_image(orthanc_raw, monkeypatch, missing_message: ], ) async def test_querying_pacs_during_working_hours( - orthanc_raw, query_date, monkeypatch, missing_message: Message + orthanc_raw, query_date, monkeypatch, missing_message: ImagingRequestMessage ) -> None: """ Given a message for a study that is missing in both the VNA and PACS, @@ -647,7 +657,7 @@ async def test_querying_pacs_during_working_hours( @pytest.mark.processing @pytest.mark.asyncio async def test_querying_pacs_not_defined( - orthanc_raw, monkeypatch, missing_message: Message + orthanc_raw, monkeypatch, missing_message: ImagingRequestMessage ) -> None: """ Given a message for a study that is missing in the VNA and the SECONDARY_DICOM_SOURCE_AE_TITLE