Bug fixes

app/broadcasters/pubsub_broadcaster.py

@@ -1,4 +1,3 @@
-import json
 from typing import Protocol
 
 from sdx_base.services.pubsub import PubsubService
@@ -25,4 +24,4 @@ def __init__(
         self.pubsub_client = PubsubService()
 
     def broadcast(self, dataset_metadata: DatasetMetadata) -> None:
-        self.pubsub_client.publish_message(self.settings.publish_dataset_topic_id, json.dumps(dataset_metadata), {})
+        self.pubsub_client.publish_message(self.settings.publish_dataset_topic_id, dataset_metadata.model_dump_json(), {})

app/exceptions/dataset_broadcast_exception.py

@@ -0,0 +1,4 @@
+from app.exceptions import DatasetException
+
+
+class DatasetBroadcastException(DatasetException): ...

app/routes.py

@@ -137,7 +137,7 @@ async def delete_dataset(dataset_service: DatasetService = DEPS.depends(DatasetS
     try:
         dataset_service.delete_dataset()
     except NonCriticalException as e:
-        # Return a status 200 (non critical exception)
+        # Return a status 200 (non-critical exception)
         return JSONResponse(
             status_code=200,
             content={"success": True, "message": str(e)},

app/services/dataset_service.py

@@ -4,6 +4,7 @@
 
 from app import get_logger
 from app.enums.delete_status import DeleteStatus
+from app.exceptions.dataset_broadcast_exception import DatasetBroadcastException
 from app.exceptions.dataset_deletion_empty_exception import DatasetDeletionEmptyException
 from app.exceptions.dataset_deletion_exception import DatasetDeletionException
 from app.exceptions.dataset_not_found_exception import DatasetNotFoundException
@@ -225,8 +226,12 @@ def create_dataset(self):
         # Create a DatasetMetadata object
         dataset_metadata = DatasetMetadata(dataset_id=dataset_id, **new_dataset_metadata.model_dump())
 
-        # Broadcast the dataset has been created (pubsub)
-        self.broadcaster.broadcast(dataset_metadata)
+        try:
+            # Broadcast the dataset has been created (pubsub)
+            self.broadcaster.broadcast(dataset_metadata)
+        except Exception as e:
+            logger.error("Failed to broadcast new dataset")
+            raise DatasetBroadcastException from e
 
         logger.info(f"Cleaning up for: {dataset_id}")
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sds-loader"
-version = "0.1.2"
+version = "0.1.3"
 description = "A service for loading new SDS schemas and datasets"
 requires-python = "==3.13.*"
 dependencies = [
@@ -14,10 +14,6 @@ dependencies = [
     "requests>=2.32.3",
 ]
 
-[[tool.uv.index]]
-name = "sds-repo"
-url = "https://europe-west2-python.pkg.dev/ons-sds-ci/sds-python-packages/simple/"
-
 [dependency-groups]
 dev = [
     "pytest>=8.3.5",
@@ -34,17 +30,21 @@ version-check = [
     "tomlkit>=0.13.3",
 ]
 
-[tool.uv.sources]
-sdx-base = { index = "sdx-repo" }
+[[tool.uv.index]]
+name = "sds-repo"
+url = "https://europe-west2-python.pkg.dev/ons-sds-ci/sds-python-packages/simple/"
 
 [[tool.uv.index]]
 name = "sdx-repo"
 url = "https://europe-west2-python.pkg.dev/ons-sdx-ci/sdx-python-packages/simple/"
 
+[tool.uv.sources]
+sdx-base = { index = "sdx-repo" }
+sds-common = { index = "sds-repo" }
 
 [tool.ruff]
 line-length = 121
-exclude = ["lib", "__pycache__", ".venv", ]
+exclude = ["lib", "__pycache__", ".venv"]
 
 [tool.ruff.lint]
 extend-select = ["E303", "W391", "W292", "E302"]

tests/integration/test_create_dataset_endpoint.py

@@ -167,5 +167,72 @@ class MockSettings:
                 "/events/dataset/create",
             )
 
-            # Assert a 200
+            # Assert a 500
+            assert response.status_code == 500
+
+    def test_create_dataset_returns_500_if_the_broadcast_fails(
+        self,
+        test_app: FastAPI,
+        mock_dataset_source_repo,
+        mock_dataset_storage_repo,
+        mock_dataset_deletion_repo,
+        mock_broadcaster: MockBroadcaster,
+        raw_dataset_factory: RawDatasetFactory,
+
+    ):
+        """
+        Test that when the broadcast fails, a 500 error code is returned.
+        """
+
+        # Mock the broadcaster to raise an exception
+        mock_broadcaster.broadcast = lambda: Exception("Error broadcasting")
+
+        # Use predictable survey_id and period_id
+        survey_id = "123"
+        period_id = "456"
+
+        # ------------------------
+        # Source repository mocks
+        # ------------------------
+
+        # Mock the source repo to return a valid JSON filename
+        mock_dataset_source_repo.get_oldest_filename.return_value = "valid-filename.json"
+
+        # Mock the source repo to return valid RawDataset
+        mock_dataset_source_repo.get_raw_data.return_value = raw_dataset_factory.build(
+            survey_id=survey_id,
+            period_id=period_id,
+        )
+
+        # ------------------------
+        # Storage repository mocks
+        # ------------------------
+
+        # Mock the current storage repository (firestore) to force version 1
+        mock_dataset_storage_repo.get_latest_dataset_metadata.return_value = None
+
+        with DEPS.override_for_test() as test_container:
+            # For this test we set autodelete to True
+            class MockSettings:
+                autodelete_dataset: bool = True
+                retain_old_datasets = False
+
+            # Override the DatasetService dependencies with our mocks
+            test_container[DatasetService] = DatasetService(
+                dataset_source_repo=mock_dataset_source_repo,
+                dataset_storage_repo=mock_dataset_storage_repo,
+                dataset_deletion_repo=mock_dataset_deletion_repo,
+                broadcaster=mock_broadcaster,
+                settings=MockSettings(),
+            )
+
+            # Create a TestClient instance
+            client = TestClient(test_app)
+
+            # Make the get request to the endpoint
+            response = client.get(
+                "/events/dataset/create",
+            )
+
+            # Assert a 500
             assert response.status_code == 500