diff --git a/src/eea_datalakehouse/dds_ingestion/__init__.py b/src/eea_datalakehouse/dds_ingestion/__init__.py index 54d714c..af9f631 100644 --- a/src/eea_datalakehouse/dds_ingestion/__init__.py +++ b/src/eea_datalakehouse/dds_ingestion/__init__.py @@ -27,7 +27,12 @@ from __future__ import annotations -from .client import IngestApiError, IngestClient, S3UploadError +from .client import ( + IngestApiError, + IngestClient, + S3UploadError, + StorageUnavailableError, +) from .credentials import ( DremioCreds, MissingCredentialsError, @@ -77,6 +82,7 @@ "S3UploadError", "StageResult", "StatusResult", + "StorageUnavailableError", "UploadPart", "UploadTarget", "ingest_folder", diff --git a/src/eea_datalakehouse/dds_ingestion/client.py b/src/eea_datalakehouse/dds_ingestion/client.py index cc6b8b7..5499844 100644 --- a/src/eea_datalakehouse/dds_ingestion/client.py +++ b/src/eea_datalakehouse/dds_ingestion/client.py @@ -72,6 +72,31 @@ def __init__(self, status_code: int, message: str, *, where: str | None = None) self.where = where +class StorageUnavailableError(IngestApiError): + """DDS refused the transfer: it cannot write to its own S3 storage (503). + + Not a fault in the folder, the target or the caller's rights — the service + verified, with its own credentials, that the object storage behind the + catalog rejects writes, and said so instead of handing out upload targets + that every file would fail against. Nothing has been uploaded; the fix is an + administrator's (credentials, bucket policy, endpoint), and the transfer can + simply be re-run once storage is working. + + Raised only on DDS's own ``storage_unavailable`` answer, so a 503 from a + proxy or load balancer in front of the service stays a plain + :class:`IngestApiError` and is not mislabelled as a storage fault. + """ + + def __init__(self, status_code: int, message: str, *, where: str | None = None) -> None: + # The server's message is already written for the person reading this in + # a notebook; wrapping it in "DDS ingest API error 503 on POST ..." would + # bury the explanation behind plumbing. The class name carries the rest. + RuntimeError.__init__(self, message) + self.status_code = status_code + self.message = message + self.where = where + + class S3UploadError(IngestApiError): """A pre-signed upload to object storage failed — NOT a DDS API error. @@ -381,12 +406,24 @@ def _error_message(resp: httpx.Response) -> str: message = message[:_MAX_ERROR_CHARS] + "… (truncated)" return message or f"(empty {resp.status_code} response body)" + @staticmethod + def _error_slug(resp: httpx.Response) -> str: + """The machine-readable ``error`` slug of a DDS error body, or ``""``.""" + try: + payload = resp.json() + except ValueError: + return "" + return str(payload.get("error", "")) if isinstance(payload, dict) else "" + @classmethod def _raise_for_status(cls, resp: httpx.Response, where: str) -> None: """Raise :class:`IngestApiError` naming ``where`` unless the call succeeded.""" if resp.is_success: return - raise IngestApiError(resp.status_code, cls._error_message(resp), where=where) + message = cls._error_message(resp) + if cls._error_slug(resp) == "storage_unavailable": + raise StorageUnavailableError(resp.status_code, message, where=where) + raise IngestApiError(resp.status_code, message, where=where) @classmethod def _raise_for_upload( diff --git a/tests/dds_ingestion/test_client.py b/tests/dds_ingestion/test_client.py index 723aaaa..d0b7eb2 100644 --- a/tests/dds_ingestion/test_client.py +++ b/tests/dds_ingestion/test_client.py @@ -7,6 +7,7 @@ IngestApiError, IngestClient, S3UploadError, + StorageUnavailableError, ) from eea_datalakehouse.dds_ingestion.credentials import DremioCreds from eea_datalakehouse.dds_ingestion.models import FileSpec, UploadPart, UploadTarget @@ -303,3 +304,70 @@ def test_commit_result_carries_the_physical_location() -> None: {"session_id": "s2", "status": "done", "table_path": "x/y", "record_count": 1} ) assert staged.storage_path is None + + +# --- storage the service itself cannot write to (DDS 503) ------------------ + +_STORAGE_DOWN = ( + "this transfer cannot start: DDS cannot write to the S3 storage behind the " + "catalog, so every file would fail to upload. This is a dependency problem " + "between DDS and S3 that has to be resolved by an administrator — it is not " + "caused by your data or your permissions, and nothing has been uploaded. " + "Verified at 2026-08-25T14:02:11+00:00 — catalog bucket (ingest): write: " + "ClientError: AccessDenied (HTTP 403): Access Denied." +) + + +@respx.mock +def test_storage_unavailable_is_its_own_error(creds: DremioCreds) -> None: + """DDS refusing up front is not the caller's mistake, and says so. + + The server has already written the explanation for whoever is reading it in + a notebook, so it must arrive intact rather than wrapped in plumbing. + """ + respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock( + return_value=httpx.Response( + 503, + json={ + "error": "storage_unavailable", + "message": _STORAGE_DOWN, + "path": "bio.uploads", + }, + ) + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises( + StorageUnavailableError + ) as exc: + client.begin( + target_catalog_path="bio.uploads", + intent="read_only", + data_format="parquet", + conflict_mode="fail", + files=[FileSpec("a.parquet", 6)], + ) + + assert exc.value.status_code == 503 + assert str(exc.value) == _STORAGE_DOWN + assert "DDS ingest API error" not in str(exc.value) + assert exc.value.where == "POST /api/v1/ingest/begin" + # …and still an IngestApiError, so existing handling keeps working. + assert isinstance(exc.value, IngestApiError) + + +@respx.mock +def test_a_503_from_a_proxy_is_not_called_a_storage_fault(creds: DremioCreds) -> None: + """Only DDS's own slug means storage; a gateway's 503 means the gateway.""" + respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock( + return_value=httpx.Response(503, text="503 Service Unavailable") + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError) as exc: + client.begin( + target_catalog_path="bio.uploads", + intent="read_only", + data_format="parquet", + conflict_mode="fail", + files=[FileSpec("a.parquet", 6)], + ) + + assert not isinstance(exc.value, StorageUnavailableError) + assert "POST /api/v1/ingest/begin" in str(exc.value) diff --git a/tests/dds_ingestion/test_folder.py b/tests/dds_ingestion/test_folder.py index c2b6025..765a7b9 100644 --- a/tests/dds_ingestion/test_folder.py +++ b/tests/dds_ingestion/test_folder.py @@ -6,6 +6,7 @@ from pathlib import Path import pytest +from eea_datalakehouse.dds_ingestion.client import StorageUnavailableError from eea_datalakehouse.dds_ingestion.folder import ( FolderIngest, IngestStateError, @@ -296,3 +297,35 @@ def test_retry_still_starts_over_for_a_staged_transfer(tmp_path: Path) -> None: job._session_id = "sess-1" job.retry() assert len(client.begin_calls) == 1 + + +# --- storage DDS cannot write to ------------------------------------------- + + +class StorageDownClient(FakeClient): + """A server that refuses the transfer at ``begin``: its S3 rejects writes.""" + + def begin(self, *, files: list[FileSpec], **kwargs: object) -> BeginResult: + self.begin_calls.append({"files": files, **kwargs}) + raise StorageUnavailableError( + 503, + "this transfer cannot start: DDS cannot write to the S3 storage " + "behind the catalog … resolved by an administrator … nothing has " + "been uploaded.", + where="POST /api/v1/ingest/begin", + ) + + +def test_run_stops_before_uploading_when_dds_cannot_write_to_s3( + data_folder: Path, +) -> None: + """The point of the server-side pre-flight: nothing is transferred.""" + client = StorageDownClient() + job = _make_ingest(data_folder, client) + + with pytest.raises(StorageUnavailableError) as exc: + job.run() + + assert "administrator" in str(exc.value) + assert client.uploaded == [] + assert client.commit_calls == []