diff --git a/cg/services/events/constants.py b/cg/services/events/constants.py new file mode 100644 index 0000000000..82aac85b82 --- /dev/null +++ b/cg/services/events/constants.py @@ -0,0 +1,11 @@ +# Event names +EXTERNAL_SAMPLES_ORDERED_SUBJECT = "external.samples_ordered" +EXTERNAL_SAMPLE_STORED_SUBJECT = "external.sample_storage_completed" +EXTERNAL_SAMPLE_TRANSFERRED_SUBJECT = "external.sample_transfer_completed" +EXTERNAL_SAMPLE_UPLOADED_SUBJECT = "external.sample_upload_completed" + +# Payload attributes +CUSTOMER_INTERNAL_ID_FIELD = "status_db.customer.internal_id" +SAMPLE_INTERNAL_ID_FIELD = "status_db.sample.internal_id" +SAMPLE_NAME_ARRAY_FIELD = "status_db.sample.name.array" +SAMPLE_NAME_FIELD = "status_db.sample.name" diff --git a/cg/services/events/event_dispatching.py b/cg/services/events/event_dispatching.py index 3dd25909e9..7261d24be0 100644 --- a/cg/services/events/event_dispatching.py +++ b/cg/services/events/event_dispatching.py @@ -2,13 +2,18 @@ from typing import Protocol from cg.models.cg_config import CGConfig +from cg.services.events.constants import ( + EXTERNAL_SAMPLE_STORED_SUBJECT, + EXTERNAL_SAMPLE_TRANSFERRED_SUBJECT, + EXTERNAL_SAMPLE_UPLOADED_SUBJECT, + EXTERNAL_SAMPLES_ORDERED_SUBJECT, +) from cg.services.events.event_handlers import ( - external_sample_transferred_handler, external_sample_stored_handler, + external_sample_transferred_handler, external_sample_uploaded_handler, external_samples_ordered_handler, ) -from cg.services.transfer_to_cluster_service import EXTERNAL_SAMPLE_TRANSFERRED_SUBJECT LOG = logging.getLogger(__name__) @@ -18,16 +23,16 @@ def __call__(self, config: CGConfig, event_payload: dict) -> None: ... EVENT_HANDLERS: dict[str, EventHandler] = { - "external.customer_uploaded_sample": external_sample_uploaded_handler.handle, - "external.sample_stored": external_sample_stored_handler.handle, - "external.samples_ordered": external_samples_ordered_handler.handle, + EXTERNAL_SAMPLE_UPLOADED_SUBJECT: external_sample_uploaded_handler.handle, + EXTERNAL_SAMPLE_STORED_SUBJECT: external_sample_stored_handler.handle, + EXTERNAL_SAMPLES_ORDERED_SUBJECT: external_samples_ordered_handler.handle, EXTERNAL_SAMPLE_TRANSFERRED_SUBJECT: external_sample_transferred_handler.handle, } def dispatch( config: CGConfig, event_name: str, event_payload: dict, event_handlers: dict = EVENT_HANDLERS -): +) -> None: """ Select the appropriate handler for the given event name and call it with the provided payload. """ diff --git a/cg/services/events/event_handlers/external_sample_stored_handler.py b/cg/services/events/event_handlers/external_sample_stored_handler.py index a71645158d..cfd3c9979c 100644 --- a/cg/services/events/event_handlers/external_sample_stored_handler.py +++ b/cg/services/events/event_handlers/external_sample_stored_handler.py @@ -7,6 +7,7 @@ from cg.models.cg_config import CGConfig from cg.services.analysis_starter.analysis_starter import AnalysisStarter from cg.services.analysis_starter.factories.starter_factory import AnalysisStarterFactory +from cg.services.events.constants import SAMPLE_INTERNAL_ID_FIELD from cg.store.models import Case, Sample from cg.store.store import Store @@ -14,7 +15,7 @@ class ExternalSampleStoredEvent(BaseModel): - sample_internal_id: str = Field(alias="status_db.sample_internal_id") + sample_internal_id: str = Field(alias=SAMPLE_INTERNAL_ID_FIELD) def handle(config: CGConfig, event_payload: dict) -> None: diff --git a/cg/services/events/event_handlers/external_sample_transferred_handler.py b/cg/services/events/event_handlers/external_sample_transferred_handler.py index f6a3515862..39b4812063 100644 --- a/cg/services/events/event_handlers/external_sample_transferred_handler.py +++ b/cg/services/events/event_handlers/external_sample_transferred_handler.py @@ -9,14 +9,14 @@ from cg.exc import CgError from cg.models.cg_config import CGConfig from cg.services.events import event_publisher +from cg.services.events.constants import EXTERNAL_SAMPLE_STORED_SUBJECT, SAMPLE_INTERNAL_ID_FIELD from cg.store.models import Sample LOG = logging.getLogger(__name__) -EXTERNAL_SAMPLE_STORED_SUBJECT = "external_sample.storage_completed" class ExternalSampleTransferredEvent(BaseModel): - sample_internal_id: str = Field(alias="statusdb.sample_internal_id") + sample_internal_id: str = Field(alias=SAMPLE_INTERNAL_ID_FIELD) cluster_location: Path transfer_completed_at: datetime @@ -48,7 +48,7 @@ def handle(config: CGConfig, event_payload: dict) -> None: event_publisher.publish( nats_config=config.nats, event_name=EXTERNAL_SAMPLE_STORED_SUBJECT, - event_payload={"statusdb.sample_internal_id": event.sample_internal_id}, + event_payload={SAMPLE_INTERNAL_ID_FIELD: event.sample_internal_id}, ) diff --git a/cg/services/events/event_handlers/external_sample_uploaded_handler.py b/cg/services/events/event_handlers/external_sample_uploaded_handler.py index 2dde9a21a0..c1d766097a 100644 --- a/cg/services/events/event_handlers/external_sample_uploaded_handler.py +++ b/cg/services/events/event_handlers/external_sample_uploaded_handler.py @@ -4,6 +4,7 @@ from cg.models.cg_config import LOG, CGConfig from cg.services import transfer_to_cluster_service +from cg.services.events.constants import CUSTOMER_INTERNAL_ID_FIELD, SAMPLE_NAME_FIELD from cg.store.models import ( SAMPLE_NAME_MAXIMUM_LENGTH, SAMPLE_NAME_MINIMUM_LENGTH, @@ -14,13 +15,13 @@ class ExternalSampleUploadedEvent(BaseModel): - customer: str = Field(alias="statusdb.customer") + customer: str = Field(alias=CUSTOMER_INTERNAL_ID_FIELD) customer_uploaded_at: datetime sample_name: str = Field( pattern=SAMPLE_NAME_PATTERN, min_length=SAMPLE_NAME_MINIMUM_LENGTH, max_length=SAMPLE_NAME_MAXIMUM_LENGTH, - alias="statusdb.sample_name", + alias=SAMPLE_NAME_FIELD, ) diff --git a/cg/services/events/event_handlers/external_samples_ordered_handler.py b/cg/services/events/event_handlers/external_samples_ordered_handler.py index 3efc76d82c..4e5e747039 100644 --- a/cg/services/events/event_handlers/external_samples_ordered_handler.py +++ b/cg/services/events/event_handlers/external_samples_ordered_handler.py @@ -4,6 +4,7 @@ from cg.models.cg_config import CGConfig from cg.services import transfer_to_cluster_service +from cg.services.events.constants import CUSTOMER_INTERNAL_ID_FIELD, SAMPLE_NAME_ARRAY_FIELD from cg.store.models import Customer, Sample from cg.store.store import Store @@ -11,8 +12,8 @@ class ExternalSamplesOrderedEvent(BaseModel): - customer: str = Field(alias="status_db.customer") - sample_names: list[str] = Field(alias="status_db.sample_names") + customer: str = Field(alias=CUSTOMER_INTERNAL_ID_FIELD) + sample_names: list[str] = Field(alias=SAMPLE_NAME_ARRAY_FIELD) def handle(config: CGConfig, event_payload: dict): diff --git a/cg/services/transfer_to_cluster_service.py b/cg/services/transfer_to_cluster_service.py index 3df759d922..c9a6c6d329 100644 --- a/cg/services/transfer_to_cluster_service.py +++ b/cg/services/transfer_to_cluster_service.py @@ -11,10 +11,13 @@ RSYNC_CONTENTS_COMMAND, ) from cg.services.events import event_publisher +from cg.services.events.constants import ( + EXTERNAL_SAMPLE_TRANSFERRED_SUBJECT, + SAMPLE_INTERNAL_ID_FIELD, +) from cg.store.models import Sample LOG = logging.getLogger(__name__) -EXTERNAL_SAMPLE_TRANSFERRED_SUBJECT = "external_sample.transfer_completed" RSYNC_SBATCH_SCRIPT: str = "transfer_sample.sh" @@ -54,7 +57,7 @@ def _get_sbatch_command(cg_config: CGConfig, sample: Sample) -> str: destination_path.mkdir(parents=True, exist_ok=True) LOG.debug(f"Destination directory: {destination_path}") event_payload = { - "statusdb.sample_internal_id": sample.internal_id, + SAMPLE_INTERNAL_ID_FIELD: sample.internal_id, "transfer_completed_at": "$(date +%Y-%m-%dT%H:%M:%S)", "cluster_location": destination_path.as_posix(), } diff --git a/tests/services/events/event_handlers/test_external_sample_stored_handler.py b/tests/services/events/event_handlers/test_external_sample_stored_handler.py index 84f78d11cf..275bd319d0 100644 --- a/tests/services/events/event_handlers/test_external_sample_stored_handler.py +++ b/tests/services/events/event_handlers/test_external_sample_stored_handler.py @@ -9,6 +9,7 @@ from cg.models.cg_config import CGConfig from cg.services.analysis_starter.analysis_starter import AnalysisStarter from cg.services.analysis_starter.factories.starter_factory import AnalysisStarterFactory +from cg.services.events.constants import SAMPLE_INTERNAL_ID_FIELD from cg.services.events.event_handlers import external_sample_stored_handler from cg.store.models import Case, Sample from cg.store.store import Store @@ -17,7 +18,7 @@ def test_handle_starts_case(mocker: MockerFixture): # GIVEN a valid event payload with a sample id - event_payload: dict = {"status_db.sample_internal_id": "ACC123"} + event_payload: dict = {SAMPLE_INTERNAL_ID_FIELD: "ACC123"} # GIVEN that the sample belongs to a purely external case status_db: Store = create_autospec(Store) @@ -65,7 +66,7 @@ def test_handle_starts_case(mocker: MockerFixture): def test_handle_fails_with_no_case(): # GIVEN a valid event payload with a sample id - event_payload: dict = {"status_db.sample_internal_id": "ACC123"} + event_payload: dict = {SAMPLE_INTERNAL_ID_FIELD: "ACC123"} # GIVEN that the sample does not have a linked case that should deliver it status_db: Store = create_autospec(Store) @@ -85,7 +86,7 @@ def test_handle_fails_with_no_case(): def test_handle_ignores_case_with_internal_samples(mocker: MockerFixture): # GIVEN a valid event payload with a sample id - event_payload: dict = {"status_db.sample_internal_id": "ACC123"} + event_payload: dict = {SAMPLE_INTERNAL_ID_FIELD: "ACC123"} # GIVEN that the sample belongs to a case that delivers the sample status_db: Store = create_autospec(Store) @@ -127,7 +128,7 @@ def test_handle_ignores_case_with_internal_samples(mocker: MockerFixture): def test_handle_ignores_case_with_unstored_samples(mocker: MockerFixture): # GIVEN a valid event payload with a sample id - event_payload: dict = {"status_db.sample_internal_id": "ACC123"} + event_payload: dict = {SAMPLE_INTERNAL_ID_FIELD: "ACC123"} # GIVEN that the sample belongs to a purely external case status_db: Store = create_autospec(Store) @@ -173,7 +174,7 @@ def test_handle_ignores_case_with_unstored_samples(mocker: MockerFixture): def test_handle_ignores_case_with_undeliverable_samples(mocker: MockerFixture): # GIVEN a valid event payload with a sample id - event_payload: dict = {"status_db.sample_internal_id": "ACC123"} + event_payload: dict = {SAMPLE_INTERNAL_ID_FIELD: "ACC123"} # GIVEN that the sample belongs to a case that does not deliver all of its samples status_db: Store = create_autospec(Store) diff --git a/tests/services/events/event_handlers/test_external_sample_transferred_handler.py b/tests/services/events/event_handlers/test_external_sample_transferred_handler.py index 8101af673f..f0905a1918 100644 --- a/tests/services/events/event_handlers/test_external_sample_transferred_handler.py +++ b/tests/services/events/event_handlers/test_external_sample_transferred_handler.py @@ -7,6 +7,7 @@ from cg.apps.housekeeper.hk import HousekeeperAPI from cg.models.cg_config import CGConfig, NatsConfig +from cg.services.events.constants import EXTERNAL_SAMPLE_STORED_SUBJECT, SAMPLE_INTERNAL_ID_FIELD from cg.services.events.event_handlers import external_sample_transferred_handler from cg.store.models import Sample from cg.store.store import Store @@ -44,7 +45,7 @@ def test_handle_success(mocker: MockerFixture): # GIVEN a valid event payload event_payload = { - "statusdb.sample_internal_id": "ACC123", + SAMPLE_INTERNAL_ID_FIELD: "ACC123", "cluster_location": "/path/to/home", "transfer_completed_at": "2026-08-31T14:41:00", } @@ -77,6 +78,6 @@ def test_handle_success(mocker: MockerFixture): # THEN an event was published saying the sample was stored publish_mock.assert_called_once_with( nats_config=nats_config, - event_name="external_sample.storage_completed", - event_payload={"statusdb.sample_internal_id": "ACC123"}, + event_name=EXTERNAL_SAMPLE_STORED_SUBJECT, + event_payload={SAMPLE_INTERNAL_ID_FIELD: "ACC123"}, ) diff --git a/tests/services/events/event_handlers/test_external_sample_uploaded_handler.py b/tests/services/events/event_handlers/test_external_sample_uploaded_handler.py index d48c6e2b11..c8eb251dc1 100644 --- a/tests/services/events/event_handlers/test_external_sample_uploaded_handler.py +++ b/tests/services/events/event_handlers/test_external_sample_uploaded_handler.py @@ -6,6 +6,7 @@ from pytest_mock import MockerFixture from cg.models.cg_config import CGConfig +from cg.services.events.constants import CUSTOMER_INTERNAL_ID_FIELD, SAMPLE_NAME_FIELD from cg.services.events.event_handlers import external_sample_uploaded_handler from cg.services.events.event_handlers.external_sample_uploaded_handler import ( transfer_to_cluster_service, @@ -29,8 +30,8 @@ def test_handle_triggers_transfer(mocker: MockerFixture): # GIVEN some payload for an external sample upload event event_payload = { - "statusdb.customer": "cust000", - "statusdb.sample_name": "sample-name", + CUSTOMER_INTERNAL_ID_FIELD: "cust000", + SAMPLE_NAME_FIELD: "sample-name", "customer_uploaded_at": "2026-06-02T11:14:52", } @@ -73,8 +74,8 @@ def test_handle_not_trigger_transfer(mocker: MockerFixture): # GIVEN some payload for an external sample upload event event_payload = { - "statusdb.customer": "cust000", - "statusdb.sample_name": "sample-name", + CUSTOMER_INTERNAL_ID_FIELD: "cust000", + SAMPLE_NAME_FIELD: "sample-name", "customer_uploaded_at": "2026-06-02T11:14:52", } @@ -104,8 +105,8 @@ def test_handle_invalid_sample_name(): # GIVEN some event payload where the sample name contains illegal letters event_payload = { - "statusdb.customer": "cust000", - "statusdb.sample_name": "invalid_sample_name", + CUSTOMER_INTERNAL_ID_FIELD: "cust000", + SAMPLE_NAME_FIELD: "invalid_sample_name", "customer_uploaded_at": "2026-06-02T11:14:52", } @@ -121,8 +122,8 @@ def test_handle_invalid_date_format(): # GIVEN some event payload where the uploaded at is malformed event_payload = { - "statusdb.customer": "cust000", - "statusdb.sample_name": "sample-name", + CUSTOMER_INTERNAL_ID_FIELD: "cust000", + SAMPLE_NAME_FIELD: "sample-name", "customer_uploaded_at": "2026-06-02T11:14.52", } diff --git a/tests/services/events/event_handlers/test_external_samples_ordered_handler.py b/tests/services/events/event_handlers/test_external_samples_ordered_handler.py index a51f64679f..91ef0b24ac 100644 --- a/tests/services/events/event_handlers/test_external_samples_ordered_handler.py +++ b/tests/services/events/event_handlers/test_external_samples_ordered_handler.py @@ -3,6 +3,7 @@ from pytest_mock import MockerFixture from cg.models.cg_config import CGConfig +from cg.services.events.constants import CUSTOMER_INTERNAL_ID_FIELD, SAMPLE_NAME_ARRAY_FIELD from cg.services.events.event_handlers import external_samples_ordered_handler from cg.services.events.event_handlers.external_samples_ordered_handler import ( transfer_to_cluster_service, @@ -14,8 +15,8 @@ def test_handle_trigger_transfer_only_for_stored_sample(mocker: MockerFixture): # GIVEN a payload for an order with two external samples event_payload = { - "status_db.customer": "cust000", - "status_db.sample_names": ["sample-name-1", "sample-name-2"], + CUSTOMER_INTERNAL_ID_FIELD: "cust000", + SAMPLE_NAME_ARRAY_FIELD: ["sample-name-1", "sample-name-2"], } # GIVEN that one of the samples is in the ExternalSample table diff --git a/tests/services/test_transfer_to_cluster_service.py b/tests/services/test_transfer_to_cluster_service.py index de4c63c2c1..cff15b7e6c 100644 --- a/tests/services/test_transfer_to_cluster_service.py +++ b/tests/services/test_transfer_to_cluster_service.py @@ -44,7 +44,7 @@ def expected_sbatch_content() -> str: rsync -rvL /path/to/rome/cust000/sample-name/ /path/to/hasta/cust000/sample-name -/nats/binary pub --jetstream --server nats://server --tlsca /ca/cert --tlscert /client/cert --tlskey /client/key --token $(cat /token) cg-test.external_sample.transfer_completed "{\\"statusdb.sample_internal_id\\": \\"ACC1\\", \\"transfer_completed_at\\": \\"$(date +%Y-%m-%dT%H:%M:%S)\\", \\"cluster_location\\": \\"/path/to/hasta/cust000/sample-name\\"}" +/nats/binary pub --jetstream --server nats://server --tlsca /ca/cert --tlscert /client/cert --tlskey /client/key --token $(cat /token) cg-test.external.sample_transfer_completed "{\\"status_db.sample.internal_id\\": \\"ACC1\\", \\"transfer_completed_at\\": \\"$(date +%Y-%m-%dT%H:%M:%S)\\", \\"cluster_location\\": \\"/path/to/hasta/cust000/sample-name\\"}" """