Standardise event and field names

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"

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.
     """

cg/services/events/event_handlers/external_sample_stored_handler.py

@@ -7,14 +7,15 @@
 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
 
 LOG = logging.getLogger(__name__)
 
 
 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:

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},
     )
 
 

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,
     )
 
 

cg/services/events/event_handlers/external_samples_ordered_handler.py

@@ -4,15 +4,16 @@
 
 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
 
 LOG = logging.getLogger(__name__)
 
 
 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):

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(),
     }

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)

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"},
     )

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",
     }
 

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

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\\"}"
 
 """