Hide private attrs in responses

api/app/models/common/analysis.py

@@ -69,6 +69,13 @@ def model_dump_json(self, **kwargs):
         result["_input_uris"] = self._input_uris
         return json.dumps(result)
 
+    def public_metadata(self) -> dict:
+        """Return public metadata
+
+        Bypasses the private field injection in the overridden model_dump* methods
+        """
+        return super().model_dump()
+
     def thumbprint(self) -> uuid.UUID:
         """Generate a deterministic UUID thumbprint."""
         if self._input_uris is None:

api/app/routers/common_analytics.py

@@ -106,7 +106,7 @@ async def get_analysis(
         status=analysis.status,
         message=message,
         result=analysis.result,
-        metadata=analysis.metadata,
+        metadata={k: v for k, v in analysis.metadata.items() if not k.startswith("_")},
     )
 
 

api/test/unit/models/common/test_analysis.py

@@ -1,10 +1,18 @@
+import uuid
+from unittest.mock import AsyncMock, MagicMock
+
 import pytest
+from fastapi import HTTPException
+from fastapi.responses import Response
 
+from app.domain.models.analysis import Analysis
 from app.domain.models.dataset import Dataset
 from app.domain.models.environment import Environment
 from app.domain.repositories.zarr_dataset_repository import ZarrDatasetRepository
+from app.models.common.analysis import AnalysisStatus
 from app.models.common.areas_of_interest import AdminAreaOfInterest
 from app.models.land_change.tree_cover_loss import TreeCoverLossAnalyticsIn
+from app.routers.common_analytics import get_analysis
 
 
 def _make_analytics_in(**kwargs) -> TreeCoverLossAnalyticsIn:
@@ -20,6 +28,36 @@ def _make_analytics_in(**kwargs) -> TreeCoverLossAnalyticsIn:
     return analytics_in
 
 
+def _make_repository(analysis: Analysis) -> AsyncMock:
+    repo = AsyncMock()
+    repo.load_analysis.return_value = analysis
+    return repo
+
+
+def _make_response() -> MagicMock:
+    response = MagicMock(spec=Response)
+    response.headers = {}
+    return response
+
+
+RESOURCE_ID = uuid.uuid4()
+
+# Metadata as it would be stored internally — includes all private fields.
+_STORED_METADATA = {
+    "aoi": {"type": "admin", "ids": ["BRA.1"], "provider": "gadm", "version": "4.1"},
+    "start_year": "2020",
+    "end_year": "2021",
+    "canopy_cover": 30,
+    "intersections": [],
+    "_version": "20250912",
+    "_analytics_name": "tree_cover_loss",
+    "_input_uris": '["s3://gfw-data-lake/umd_tree_cover_loss/v1.12/raster/epsg-4326/zarr/year.zarr"]',
+}
+
+# The public subset clients should see.
+_PUBLIC_METADATA = {k: v for k, v in _STORED_METADATA.items() if not k.startswith("_")}
+
+
 class TestThumbprint:
     def test_thumbprint_without_set_input_uris_raises(self):
         defaults = dict(
@@ -89,3 +127,118 @@ def test_different_request_params_different_thumbprints(self):
         a.set_input_uris(Environment.production)
         b.set_input_uris(Environment.production)
         assert a.thumbprint() != b.thumbprint()
+
+
+class TestGetAnalysisPrivateFieldFiltering:
+    @pytest.mark.asyncio
+    async def test_private_fields_are_absent_from_response_metadata(self):
+        repo = _make_repository(
+            Analysis(
+                result=None, metadata=_STORED_METADATA, status=AnalysisStatus.saved
+            )
+        )
+        result = await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert "_version" not in result.metadata
+        assert "_analytics_name" not in result.metadata
+        assert "_input_uris" not in result.metadata
+
+    @pytest.mark.asyncio
+    async def test_public_fields_are_preserved_in_response_metadata(self):
+        repo = _make_repository(
+            Analysis(
+                result=None, metadata=_STORED_METADATA, status=AnalysisStatus.saved
+            )
+        )
+        result = await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert result.metadata == _PUBLIC_METADATA
+
+    @pytest.mark.asyncio
+    async def test_filtering_applies_for_pending_status(self):
+        repo = _make_repository(
+            Analysis(
+                result=None, metadata=_STORED_METADATA, status=AnalysisStatus.pending
+            )
+        )
+        result = await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert "_input_uris" not in result.metadata
+        assert result.metadata == _PUBLIC_METADATA
+
+    @pytest.mark.asyncio
+    async def test_filtering_applies_for_failed_status(self):
+        repo = _make_repository(
+            Analysis(
+                result={"error": "something went wrong"},
+                metadata=_STORED_METADATA,
+                status=AnalysisStatus.failed,
+            )
+        )
+        result = await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert "_input_uris" not in result.metadata
+        assert result.metadata == _PUBLIC_METADATA
+
+    @pytest.mark.asyncio
+    async def test_metadata_with_no_private_fields_is_returned_unchanged(self):
+        """Handles legacy stored analyses that pre-date the private fields."""
+        repo = _make_repository(
+            Analysis(
+                result=None, metadata=_PUBLIC_METADATA, status=AnalysisStatus.saved
+            )
+        )
+        result = await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert result.metadata == _PUBLIC_METADATA
+
+
+class TestGetAnalysisStatusBehaviour:
+    """Basic status tests."""
+
+    @pytest.mark.asyncio
+    async def test_saved_status_returns_success_message(self):
+        repo = _make_repository(
+            Analysis(
+                result={"data": 1},
+                metadata=_STORED_METADATA,
+                status=AnalysisStatus.saved,
+            )
+        )
+        result = await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert result.status == AnalysisStatus.saved
+        assert "completed" in result.message.lower()
+
+    @pytest.mark.asyncio
+    async def test_pending_status_sets_retry_after_header(self):
+        response = _make_response()
+        repo = _make_repository(
+            Analysis(
+                result=None, metadata=_STORED_METADATA, status=AnalysisStatus.pending
+            )
+        )
+        await get_analysis(RESOURCE_ID, repo, response)
+        assert response.headers.get("Retry-After") == "1"
+
+    @pytest.mark.asyncio
+    async def test_failed_status_with_error_result_uses_error_as_message(self):
+        error_msg = "AOI is too small."
+        repo = _make_repository(
+            Analysis(
+                result={"error": error_msg},
+                metadata=_STORED_METADATA,
+                status=AnalysisStatus.failed,
+            )
+        )
+        result = await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert result.message == error_msg
+
+    @pytest.mark.asyncio
+    async def test_missing_metadata_raises_404(self):
+        repo = _make_repository(Analysis(result=None, metadata=None, status=None))
+        with pytest.raises(HTTPException) as exc_info:
+            await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert exc_info.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_repository_exception_raises_500(self):
+        repo = AsyncMock()
+        repo.load_analysis.side_effect = Exception("connection refused")
+        with pytest.raises(HTTPException) as exc_info:
+            await get_analysis(RESOURCE_ID, repo, _make_response())
+        assert exc_info.value.status_code == 500