From a841a50c58f328b52627b1a54e9be16ea82885b0 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 13 Aug 2026 08:19:30 +0000 Subject: [PATCH 01/16] Update README install examples to v0.1.7 [skip ci] --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 6f36a1c..56e16f4 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ branch at a time, so pin to whatever the badge shows *now*, not a number copied ```bash # main's latest release (stable) — pin to the tag the "main" badge above shows -pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.6" +pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.7" # staging's latest release (early access) — pin to the tag the "staging" badge above shows pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.7-staging" @@ -198,7 +198,7 @@ Run this in a notebook cell (see the live badges under [Install](#install) for t ```python # main's latest release (stable) — recommended -%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.6" +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.7" # staging's latest release (early access) %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.7-staging" From b8e1d0ecf9f2ed5353d4ab08f0ef8822cc268dd4 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 14 Aug 2026 11:14:39 +0000 Subject: [PATCH 02/16] Bump version to 0.1.8 [skip ci] --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 19bf0dc..3ff4ffe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "EEADataLakehouse" -version = "0.1.7" +version = "0.1.8" description = "Data preparation and validation utilities for the EEA data lakehouse pipeline." readme = "README.md" requires-python = ">=3.11" From 18f878974a0100865f9a7929102db00fe3d1cd19 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 14 Aug 2026 11:14:40 +0000 Subject: [PATCH 03/16] Update README install examples to v0.1.8-staging [skip ci] --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 2155ad3..232db34 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@staging" ``` # staging's latest release (early access) — pin to the tag the "staging" badge above shows -pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.7-staging" +pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.8-staging" ``` ## Usage @@ -241,7 +241,7 @@ To pin to one specific release instead, use the exact tag the live badges under %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.6" # staging's latest release (early access) -%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.7-staging" +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.8-staging" ``` Use the `%pip` magic rather than `!pip` — it installs into the kernel the From b63b0aca2a954bc23f58944b39947b2044c129d3 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 14 Aug 2026 11:19:44 +0000 Subject: [PATCH 04/16] Bump version to 0.1.9 [skip ci] --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 3ff4ffe..e986b15 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "EEADataLakehouse" -version = "0.1.8" +version = "0.1.9" description = "Data preparation and validation utilities for the EEA data lakehouse pipeline." readme = "README.md" requires-python = ">=3.11" From 938c7802f120d9e38215760bdb33d17003354096 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 14 Aug 2026 11:19:45 +0000 Subject: [PATCH 05/16] Update README install examples to v0.1.9-staging [skip ci] --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 232db34..92eae5c 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@staging" ``` # staging's latest release (early access) — pin to the tag the "staging" badge above shows -pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.8-staging" +pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.9-staging" ``` ## Usage @@ -241,7 +241,7 @@ To pin to one specific release instead, use the exact tag the live badges under %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.6" # staging's latest release (early access) -%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.8-staging" +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.9-staging" ``` Use the `%pip` magic rather than `!pip` — it installs into the kernel the From 9b7bc880fb0e0c6293f14eb2237dbe484c5cecc0 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 14 Aug 2026 11:21:31 +0000 Subject: [PATCH 06/16] Update README install examples to v0.1.9 [skip ci] --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 92eae5c..cff913c 100644 --- a/README.md +++ b/README.md @@ -238,7 +238,7 @@ To pin to one specific release instead, use the exact tag the live badges under [Releasing a new version](#releasing-a-new-version)): ```python -%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.6" +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.9" # staging's latest release (early access) %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.9-staging" From adf63545c355a6faab6552f89de4ed819cf0ab31 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 14 Aug 2026 11:31:57 +0000 Subject: [PATCH 07/16] Bump version to 0.1.10 [skip ci] --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index e986b15..10e1c35 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "EEADataLakehouse" -version = "0.1.9" +version = "0.1.10" description = "Data preparation and validation utilities for the EEA data lakehouse pipeline." readme = "README.md" requires-python = ">=3.11" From 0673e90298bd3518c819d23be51400ce2c1fbee1 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 14 Aug 2026 11:31:58 +0000 Subject: [PATCH 08/16] Update README install examples to v0.1.10-staging [skip ci] --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 92eae5c..0f88719 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@staging" ``` # staging's latest release (early access) — pin to the tag the "staging" badge above shows -pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.9-staging" +pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.10-staging" ``` ## Usage @@ -241,7 +241,7 @@ To pin to one specific release instead, use the exact tag the live badges under %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.6" # staging's latest release (early access) -%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.9-staging" +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.10-staging" ``` Use the `%pip` magic rather than `!pip` — it installs into the kernel the From c07c7c2b20da4002e6fe0c54e348b0b73fd87f05 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 14 Aug 2026 11:33:24 +0000 Subject: [PATCH 09/16] Update README install examples to v0.1.10 [skip ci] --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 6de8c03..addb1fd 100644 --- a/README.md +++ b/README.md @@ -238,7 +238,7 @@ To pin to one specific release instead, use the exact tag the live badges under [Releasing a new version](#releasing-a-new-version)): ```python -%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.9" +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.10" # staging's latest release (early access) %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.10-staging" From b90763e35f1511cf3fb10a5f51226d403d4afd84 Mon Sep 17 00:00:00 2001 From: blikij Date: Fri, 7 Aug 2026 14:53:52 +0200 Subject: [PATCH 10/16] Tidy the dds_ingestion copy-over and restore the PAT redaction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The working tree held a verbatim copy of the old standalone `dds_ingest` package laid over the restructured one. That copy is the richer code — it carries the whole session-management surface (attach/status/estimate/retry/ cancel, context-manager support, IngestStateError, EstimateResult/StageResult, is_resumable/is_terminal) that the committed version lacks — so it is kept. What it dropped or reverted is fixed here instead. - Restore IngestClient.__repr__, which withholds the Bearer PAT. Without it a traceback, log handler or debugger renders the token. - Point the module references back at eea_datalakehouse.dds_ingestion, in the docstrings and in the logger name, which had gone back to "dds_ingest" and so was no longer filterable under the package it lives in. - Delete the copy-over debris: two pyproject.toml files inside the package declaring the old `dds-ingest` distribution on the hatchling backend, a duplicate README one level too high, and two identical in-package test trees that still imported `dds_ingest` and could not load. - Port the one thing those trees had that the tracked suite did not: test_session.py, 13 tests covering the session surface above. The suite goes 31 -> 46, all passing. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01G9tqi2Kd1f8gynmSYP4j6e --- src/eea_datalakehouse/dds_ingestion/README.md | 77 +++++ .../dds_ingestion/__init__.py | 9 +- src/eea_datalakehouse/dds_ingestion/client.py | 95 +++++- src/eea_datalakehouse/dds_ingestion/folder.py | 244 ++++++++++++--- src/eea_datalakehouse/dds_ingestion/models.py | 85 +++++- tests/dds_ingestion/test_session.py | 277 ++++++++++++++++++ 6 files changed, 744 insertions(+), 43 deletions(-) create mode 100644 src/eea_datalakehouse/dds_ingestion/README.md create mode 100644 tests/dds_ingestion/test_session.py diff --git a/src/eea_datalakehouse/dds_ingestion/README.md b/src/eea_datalakehouse/dds_ingestion/README.md new file mode 100644 index 0000000..2468700 --- /dev/null +++ b/src/eea_datalakehouse/dds_ingestion/README.md @@ -0,0 +1,77 @@ +# eea_datalakehouse.dds_ingestion + +Notebook-side client for the **Dremio Document Service (DDS) Ingest API** (DI-8.4 / DI-8.5). + +A generated Jupyter notebook imports `FolderIngest` to transfer a folder of data +files to S3 and register it as a Dremio table — coordinated entirely through the +DDS REST API. The **only** object-storage access is the presigned URLs issued by +DDS; this package never uses an S3 SDK or S3 credentials. + +## Install + +This is a subpackage of the **EEADataLakehouse** library, not a distribution of +its own — install the library from the repository root: + +```bash +pip install -e ".[dev]" # dev extra adds ruff / mypy / pytest / respx +``` + +In a notebook kernel, pinned to a ref that carries this subpackage: + +```python +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@development" +``` + +## Usage (inside a notebook kernel) + +Dremio credentials are read from the injected kernel env vars `_DREMIO_USER` / +`_DREMIO_PWD`, and the service URL from `DDS_BASE_URL`. None of these are ever +logged or printed. + +```python +from eea_datalakehouse.dds_ingestion import FolderIngest + +outcome = FolderIngest( + folder="./my_data", + target_catalog_path="biodiversity.uploads", + data_format="parquet", # one of parquet | csv | json + intent="read_only", # or "editable" + conflict_mode="fail", + parallelism=4, # concurrent uploads (default 4) +).run() + +print(outcome.commit.table_path, outcome.commit.record_count) +``` + +`run()` performs `begin → upload(all files) → commit`, and leaves the session +handle available afterwards. + +## Managing the transfer + +A transfer is a server-side session, so it can be inspected and resumed: + +```python +with FolderIngest(folder="./my_data", target_catalog_path="catalog/theme/sub", + data_format="parquet") as job: + try: + job.run() + except Exception: + if job.status().is_resumable: + job.retry() # re-runs only the step that failed — no re-upload +``` + +`job.session_id` · `status()` · `estimate()` · `retry()` · `cancel()` · +`close()` · `FolderIngest.attach(session_id, ...)` to pick a transfer up in a +later kernel. + +**Full class reference:** [`docs/python-client-guide.md`](../../docs/python-client-guide.md). + +## Layout + +| Path | Purpose | +|---|---| +| `dds_ingestion/credentials.py` | env-var creds + redacted `DremioCreds` | +| `dds_ingestion/models.py` | typed request/response models for the contract | +| `dds_ingestion/client.py` | thin, unit-testable HTTP client (`IngestClient`) | +| `dds_ingestion/progress.py` | tqdm progress bar with graceful fallback | +| `dds_ingestion/folder.py` | `FolderIngest` orchestration (scan/parallel/session) | diff --git a/src/eea_datalakehouse/dds_ingestion/__init__.py b/src/eea_datalakehouse/dds_ingestion/__init__.py index 6a57109..e88210b 100644 --- a/src/eea_datalakehouse/dds_ingestion/__init__.py +++ b/src/eea_datalakehouse/dds_ingestion/__init__.py @@ -1,4 +1,5 @@ -"""Ingestion into DDS (Dremio Document Service), notebook-side (DI-8.4/8.5). +"""eea_datalakehouse.dds_ingestion — notebook-side client for the Dremio +Document Service Ingest API. A generated Jupyter notebook imports :class:`FolderIngest` to transfer a folder of data files to S3 (via DDS-issued presigned URLs only) and register it as a @@ -31,6 +32,7 @@ DEFAULT_PARALLELISM, FolderIngest, IngestOutcome, + IngestStateError, ingest_folder, scan_folder, ) @@ -38,10 +40,12 @@ BeginResult, CommitResult, DataFormat, + EstimateResult, FileSpec, Intent, Progress, S3Plan, + StageResult, StatusResult, UploadPart, UploadTarget, @@ -53,15 +57,18 @@ "CommitResult", "DataFormat", "DremioCreds", + "EstimateResult", "FileSpec", "FolderIngest", "IngestApiError", "IngestClient", "IngestOutcome", + "IngestStateError", "Intent", "MissingCredentialsError", "Progress", "S3Plan", + "StageResult", "StatusResult", "UploadPart", "UploadTarget", diff --git a/src/eea_datalakehouse/dds_ingestion/client.py b/src/eea_datalakehouse/dds_ingestion/client.py index 1e42d79..58977f9 100644 --- a/src/eea_datalakehouse/dds_ingestion/client.py +++ b/src/eea_datalakehouse/dds_ingestion/client.py @@ -1,9 +1,25 @@ """Thin, unit-testable HTTP client for the DDS Ingest API (v0.1). This layer knows nothing about local folders, parallelism or progress bars: it -only translates the four ingest endpoints to/from the typed models in -:mod:`eea_datalakehouse.dds_ingestion.models`. The orchestration logic lives in -:mod:`eea_datalakehouse.dds_ingestion.folder`. +only translates the ingest endpoints to/from the typed models in +:mod:`eea_datalakehouse.dds_ingestion.models`. The orchestration logic +lives in :mod:`eea_datalakehouse.dds_ingestion.folder`. + +One method per endpoint: + +=========================== ========================================== +:meth:`IngestClient.begin` ``POST /api/v1/ingest/begin`` +:meth:`~IngestClient.commit` ``POST /api/v1/ingest/commit`` +:meth:`~IngestClient.get_status` ``GET /api/v1/ingest/{id}`` +:meth:`~IngestClient.list_sessions` ``GET /api/v1/ingest`` +:meth:`~IngestClient.retry` ``POST /api/v1/ingest/{id}/retry`` +:meth:`~IngestClient.cancel` ``DELETE /api/v1/ingest/{id}`` +:meth:`~IngestClient.estimate` ``POST /api/v1/ingest/estimate`` +:meth:`~IngestClient.stage` ``POST /api/v1/ingest/stage`` +=========================== ========================================== + +plus :meth:`~IngestClient.upload_file` / :meth:`~IngestClient.upload_file_multipart`, +which talk to S3 through the presigned targets ``begin`` issued. Authentication uses a Bearer token (the Dremio PAT, which is the kernel's ``_DREMIO_PWD``) sent **only on the DDS API calls** — never on the presigned S3 @@ -23,8 +39,10 @@ BeginResult, CommitResult, DataFormat, + EstimateResult, FileSpec, Intent, + StageResult, StatusResult, UploadTarget, ) @@ -132,12 +150,83 @@ def commit( return CommitResult.from_json(data) def get_status(self, session_id: str) -> StatusResult: + """Current state of one session (``GET /api/v1/ingest/{id}``).""" resp = self._http.get( f"{self._base_url}/api/v1/ingest/{session_id}", headers=self._auth_headers ) self._raise_for_status(resp) return StatusResult.from_json(resp.json()) + def list_sessions(self, state: str | None = None) -> list[StatusResult]: + """Your ingest sessions (``GET /api/v1/ingest``). + + ``state`` filters server-side: ``"active"`` (still running), + ``"failed"``, or ``"all"``. Omit for the server's default view. + """ + params = {"state": state} if state else None + resp = self._http.get( + f"{self._base_url}/api/v1/ingest", params=params, headers=self._auth_headers + ) + self._raise_for_status(resp) + payload = resp.json() + rows = payload if isinstance(payload, list) else payload.get("sessions", []) + return [StatusResult.from_json(row) for row in rows] + + def retry(self, session_id: str) -> StatusResult: + """Re-run a failed session's last stage (``POST /api/v1/ingest/{id}/retry``). + + Only a ``failed`` session can be retried, and only the Dremio load stage + is resumable — the server kept the staged files for exactly that case + (see :attr:`StatusResult.is_resumable`). A session whose upload never + completed, or whose failure a re-run cannot fix, raises + :class:`IngestApiError` telling you to upload again. + """ + resp = self._http.post( + f"{self._base_url}/api/v1/ingest/{session_id}/retry", + headers=self._auth_headers, + ) + self._raise_for_status(resp) + return StatusResult.from_json(resp.json()) + + def cancel(self, session_id: str) -> None: + """Discard a session and its staged data (``DELETE /api/v1/ingest/{id}``). + + Use it to abandon a transfer: the staged objects are deleted and the + session is dropped. It does NOT remove a table that a previous commit + already created. + """ + resp = self._http.delete( + f"{self._base_url}/api/v1/ingest/{session_id}", headers=self._auth_headers + ) + self._raise_for_status(resp) + + def estimate(self, session_id: str) -> EstimateResult: + """Row-count + size class for the staged data (``POST .../estimate``). + + A pre-flight check between upload and commit. On a managed catalog the + count comes from the staged Parquet footers (CSV/JSON have none and + report 0). + """ + data = self._post("/api/v1/ingest/estimate", {"session_id": session_id}) + return EstimateResult.from_json(data) + + def stage(self, session_id: str, rel_path: str, data: bytes) -> StageResult: + """Upload one file THROUGH the server (``POST /api/v1/ingest/stage``). + + The fallback when the S3 endpoint is not reachable from this kernel: DDS + writes the bytes with its own catalog credentials instead of handing back + a presigned URL. Slower — the bytes transit the server — so prefer + :meth:`upload_file`, and use this when that fails to connect. + """ + resp = self._http.post( + f"{self._base_url}/api/v1/ingest/stage", + data={"session_id": session_id, "rel_path": rel_path}, + files={"file": (rel_path.rsplit("/", 1)[-1], data)}, + headers=self._auth_headers, + ) + self._raise_for_status(resp) + return StageResult.from_json(resp.json()) + def upload_file(self, target: UploadTarget, data: bytes) -> str | None: """Upload one file's bytes to its presigned target. diff --git a/src/eea_datalakehouse/dds_ingestion/folder.py b/src/eea_datalakehouse/dds_ingestion/folder.py index 2f71dad..5d8a470 100644 --- a/src/eea_datalakehouse/dds_ingestion/folder.py +++ b/src/eea_datalakehouse/dds_ingestion/folder.py @@ -9,8 +9,9 @@ * show a progress bar (tqdm, degrading gracefully if absent); * resume — re-running skips files already uploaded for the session. -The class depends only on :class:`~eea_datalakehouse.dds_ingestion.client.IngestClient`, -so the HTTP layer can be mocked or swapped in tests. +The class depends only on +:class:`~eea_datalakehouse.dds_ingestion.client.IngestClient`, so the HTTP +layer can be mocked or swapped in tests. """ from __future__ import annotations @@ -19,6 +20,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from pathlib import Path +from typing import Any from .client import IngestClient from .credentials import DremioCreds, load_base_url, load_creds @@ -26,8 +28,10 @@ BeginResult, CommitResult, DataFormat, + EstimateResult, FileSpec, Intent, + StatusResult, UploadTarget, ) from .progress import make_progress_bar @@ -42,15 +46,29 @@ } +class IngestStateError(RuntimeError): + """An operation was asked for in a state that cannot support it. + + Raised locally, before any HTTP call — e.g. committing a transfer that never + began, or retrying one that is still running. + """ + + @dataclass(slots=True) class IngestOutcome: - """Summary returned by :meth:`FolderIngest.run`.""" + """Summary returned by :meth:`FolderIngest.run` and :meth:`FolderIngest.retry`. - begin: BeginResult + ``begin`` is ``None`` when the transfer was resumed rather than started here + (:meth:`FolderIngest.attach` had no ``begin`` of its own to report), and + ``resumed`` says which happened. + """ + + begin: BeginResult | None commit: CommitResult files_uploaded: int files_skipped: int etags: dict[str, str] = field(default_factory=dict) + resumed: bool = False def scan_folder(folder: Path, data_format: DataFormat) -> list[FileSpec]: @@ -108,6 +126,11 @@ def __init__( self.multipart = multipart self.show_progress = show_progress + # Set once ``begin`` runs (or by ``attach``); the handle every session + # command below works from. + self._session_id: str | None = None + self._begin: BeginResult | None = None + # The client owns the credentials; if the caller did not inject one we # build it from the kernel environment. Creds never leave the client. if client is not None: @@ -119,51 +142,192 @@ def __init__( self._client = IngestClient(resolved_url, resolved_creds) self._owns_client = True + # -- session handle --------------------------------------------------- + + @property + def session_id(self) -> str | None: + """Id of the transfer this object is driving, once ``begin`` has run.""" + return self._session_id + + @classmethod + def attach( + cls, + session_id: str, + folder: str | Path, + target_catalog_path: str, + *, + data_format: DataFormat, + **kwargs: Any, + ) -> FolderIngest: + """Bind to an EXISTING session instead of starting a new one. + + For picking a transfer back up in a later kernel — inspect it with + :meth:`status`, resume it with :meth:`retry`, or abandon it with + :meth:`cancel`. ``folder`` still has to point at the same local data, so + a re-upload is possible if the staged copy is gone. + """ + job = cls(folder, target_catalog_path, data_format=data_format, **kwargs) + job._session_id = session_id + return job + + def close(self) -> None: + """Release the HTTP client, if this object created it.""" + if self._owns_client: + self._client.close() + + def __enter__(self) -> FolderIngest: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + self.close() + # -- orchestration ---------------------------------------------------- def run(self) -> IngestOutcome: - """Execute the full begin → upload → commit flow.""" + """Execute the full begin → upload → commit flow. - try: - files = scan_folder(self.folder, self.data_format) - if not files: - raise FileNotFoundError( - f"no {self.data_format} files found under {self.folder}" - ) - begin = self._client.begin( - target_catalog_path=self.target_catalog_path, - intent=self.intent, - data_format=self.data_format, - conflict_mode=self.conflict_mode, - files=files, - table_name=self.table_name, - idempotency_key=self.idempotency_key, - multipart=self.multipart, + The client is left open afterwards so the same object can still + :meth:`status`, :meth:`retry` or :meth:`cancel` the transfer. Use it as a + context manager (or call :meth:`close`) to release it. + """ + begin = self.begin() + etags, multipart_etags, uploaded, skipped = self._upload_all(begin) + commit = self.commit(multipart_etags=multipart_etags or None) + return IngestOutcome( + begin=begin, + commit=commit, + files_uploaded=uploaded, + files_skipped=skipped, + etags=etags, + ) + + def begin(self) -> BeginResult: + """Open the transfer and get the presigned upload targets (step 1 of 3). + + The server validates the target here — an unauthorised or non-existent + catalog folder fails now, before a single byte is uploaded. + """ + files = scan_folder(self.folder, self.data_format) + if not files: + raise FileNotFoundError( + f"no {self.data_format} files found under {self.folder}" ) - etags, multipart_etags, uploaded, skipped = self._upload_all(begin) - commit = self._client.commit( - session_id=begin.session_id, - multipart_etags=multipart_etags or None, + begin = self._client.begin( + target_catalog_path=self.target_catalog_path, + intent=self.intent, + data_format=self.data_format, + conflict_mode=self.conflict_mode, + files=files, + table_name=self.table_name, + idempotency_key=self.idempotency_key, + multipart=self.multipart, + ) + self._session_id = begin.session_id + self._begin = begin + return begin + + def commit( + self, *, multipart_etags: list[dict[str, object]] | None = None + ) -> CommitResult: + """Finalise the transfer: load the staged files into the table (step 3). + + This is where the Dremio work happens (``CREATE TABLE`` + ``COPY INTO`` + on a managed catalog), so it is the step most likely to fail — see + :meth:`retry`. + """ + if self._session_id is None: + raise IngestStateError("no session to commit — call begin() or run() first") + return self._client.commit( + session_id=self._session_id, multipart_etags=multipart_etags + ) + + # -- session management ----------------------------------------------- + + def status(self) -> StatusResult: + """Current server-side state of this transfer.""" + if self._session_id is None: + raise IngestStateError("no session yet — call begin() or run() first") + return self._client.get_status(self._session_id) + + def estimate(self) -> EstimateResult: + """Row count + size class of the staged data, between upload and commit.""" + if self._session_id is None: + raise IngestStateError("no session yet — call begin() or run() first") + return self._client.estimate(self._session_id) + + def cancel(self) -> None: + """Abandon this transfer and delete its staged data. + + Does not drop a table an earlier successful commit already created. + """ + if self._session_id is None: + raise IngestStateError("no session to cancel") + self._client.cancel(self._session_id) + self._session_id = None + + def retry(self) -> IngestOutcome: + """Re-run this transfer from the step that failed. + + Two cases, decided by the server's own report of where it broke: + + * **The load failed** (``CREATE TABLE`` / ``COPY INTO``) and the staged + files were kept — the common case, e.g. Dremio was briefly unavailable + or the target folder was fixed afterwards. Only that step re-runs; the + upload is NOT repeated. + * **Anything else** — the upload never finished, or the failure was one a + re-run cannot fix (an unresolved collision, an unreadable file), so the + staged copy is gone. A fresh session is opened and the folder is + uploaded again. Any ``idempotency_key`` is dropped for that attempt, or + the server would just replay the failed session. + + Raises :class:`IngestStateError` if there is nothing to retry — no + session, one that is still running, or one that already succeeded. + """ + if self._session_id is None: + raise IngestStateError("no session to retry — call run() first") + status = self._client.get_status(self._session_id) + if status.status == "done": + raise IngestStateError( + f"transfer {self._session_id} already succeeded; nothing to retry" ) + if status.status in ("pending", "uploading", "committing"): + raise IngestStateError( + f"transfer {self._session_id} is {status.status}; wait for it to " + "finish before retrying" + ) + if status.is_resumable: + logger.info("resuming the load step of %s", self._session_id) + resumed = self._client.retry(self._session_id) return IngestOutcome( - begin=begin, - commit=commit, - files_uploaded=uploaded, - files_skipped=skipped, - etags=etags, + begin=self._begin, + commit=CommitResult.from_json(resumed.raw), + files_uploaded=0, + files_skipped=len(self._begin.s3.uploads) if self._begin else 0, + resumed=True, ) - finally: - if self._owns_client: - self._client.close() + # Nothing to resume server-side: start over with a new session, which + # means dropping the idempotency key that would replay the failed one. + logger.info( + "transfer %s cannot be resumed (%s); re-uploading under a new session", + self._session_id, + status.raw.get("failed_stage") or status.status, + ) + self._session_id = None + self._begin = None + self.idempotency_key = None + return self.run() # -- upload phase ----------------------------------------------------- def _already_done(self, session_id: str) -> set[str]: """Return rel_paths already uploaded for this session (resume support). - Queries the session status; the server reports ``files_done`` and may - list completed rel_paths under ``raw["uploaded"]``. We only skip files - the server explicitly names, so resume never wrongly drops a file. + Queries the session status and skips only the files the server + explicitly names under ``raw["uploaded"]``, so resume never wrongly drops + a file. NOTE: the server populates that list for **server-run** transfers + (``POST /ingest/folder``); a client-uploaded session reports nothing + there, so re-running one uploads every file again (an S3 overwrite of the + identical key, which is harmless). """ try: @@ -242,11 +406,15 @@ def ingest_folder( data_format: DataFormat, **kwargs: object, ) -> IngestOutcome: - """Convenience wrapper: build a :class:`FolderIngest` and run it.""" + """Convenience wrapper: build a :class:`FolderIngest`, run it, close it. - return FolderIngest( + One-shot. If the transfer might need a :meth:`FolderIngest.retry`, drive the + class directly (ideally as a context manager) so the session handle survives. + """ + with FolderIngest( folder, target_catalog_path, data_format=data_format, **kwargs, # type: ignore[arg-type] - ).run() + ) as job: + return job.run() diff --git a/src/eea_datalakehouse/dds_ingestion/models.py b/src/eea_datalakehouse/dds_ingestion/models.py index 1ed1da9..7dddb4e 100644 --- a/src/eea_datalakehouse/dds_ingestion/models.py +++ b/src/eea_datalakehouse/dds_ingestion/models.py @@ -147,7 +147,12 @@ def from_json(cls, data: dict[str, Any]) -> Progress: @dataclass(frozen=True, slots=True) class StatusResult: - """Response from ``GET /api/v1/ingest/{session_id}``.""" + """Response from ``GET /api/v1/ingest/{session_id}`` (and the list/retry calls). + + ``status`` is the lifecycle state — ``pending`` → ``uploading`` → + ``committing`` → ``done``, or ``failed`` / ``cancelled``. ``raw`` keeps the + whole payload so fields the client does not model yet stay reachable. + """ status: str progress: Progress @@ -160,3 +165,81 @@ def from_json(cls, data: dict[str, Any]) -> StatusResult: progress=Progress.from_json(data.get("progress") or {}), raw=data, ) + + @property + def session_id(self) -> str | None: + value = self.raw.get("session_id") + return str(value) if value is not None else None + + @property + def table_path(self) -> str | None: + value = self.raw.get("table_path") + return str(value) if value is not None else None + + @property + def record_count(self) -> int | None: + value = self.raw.get("record_count") + return int(value) if value is not None else None + + @property + def error(self) -> str | None: + """Why the transfer failed, including the stage — e.g. ``"Dremio load + failed: ..."`` or ``"S3 upload failed: ..."``. ``None`` unless failed.""" + value = self.raw.get("error") + return str(value) if value is not None else None + + @property + def is_terminal(self) -> bool: + """Whether the session has finished, one way or another.""" + return self.status in ("done", "failed", "cancelled") + + @property + def is_resumable(self) -> bool: + """Whether :meth:`FolderIngest.retry` can re-run this transfer. + + A failed transfer is resumable when the server kept its staged files — + which it does when the upload landed and only the Dremio load failed on + a managed catalog. Anything else has to be uploaded again. + + This reads the server's own ``resumable`` verdict rather than re-deriving + it from ``failed_stage``: a load-stage failure the server could not hold + the bytes for reports ``failed_stage="load"`` too, and inferring from that + sent :meth:`~eea_datalakehouse.dds_ingestion.folder.FolderIngest.retry` + down the resume path to be told the staged data was gone — instead of + simply re-uploading. A server that does not send the field reads as not + resumable, which is the safe direction: the transfer is re-uploaded + under a new session. + """ + return self.status == "failed" and bool(self.raw.get("resumable")) + + +@dataclass(frozen=True, slots=True) +class EstimateResult: + """Response from ``POST /api/v1/ingest/estimate`` (pre-flight sizing).""" + + record_count: int + size_class: str + + @classmethod + def from_json(cls, data: dict[str, Any]) -> EstimateResult: + return cls( + record_count=int(data.get("record_count", 0)), + size_class=str(data.get("size_class", "")), + ) + + +@dataclass(frozen=True, slots=True) +class StageResult: + """Response from ``POST /api/v1/ingest/stage`` (server-proxied upload).""" + + rel_path: str + key: str + bytes_written: int | None = None + + @classmethod + def from_json(cls, data: dict[str, Any]) -> StageResult: + return cls( + rel_path=str(data["rel_path"]), + key=str(data["key"]), + bytes_written=data.get("bytes"), + ) diff --git a/tests/dds_ingestion/test_session.py b/tests/dds_ingestion/test_session.py new file mode 100644 index 0000000..5acb601 --- /dev/null +++ b/tests/dds_ingestion/test_session.py @@ -0,0 +1,277 @@ +"""Session management on :class:`FolderIngest` — status, retry, cancel, attach. + +The important behaviour is :meth:`FolderIngest.retry`, which resumes at the step +that failed. The server tells it which step that was; these tests pin both +branches: + +* the load failed and the staged files were kept → re-run ONLY the load, no + re-upload; +* anything else → the staged copy is gone, so upload again under a NEW session + (and drop the idempotency key, or the server would replay the failed one). +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from eea_datalakehouse.dds_ingestion.folder import FolderIngest, IngestStateError +from eea_datalakehouse.dds_ingestion.models import ( + BeginResult, + CommitResult, + S3Plan, + StatusResult, + UploadTarget, +) + + +class SessionClient: + """Fake IngestClient that serves a scripted sequence of session states.""" + + def __init__(self, *statuses: dict[str, Any]) -> None: + self._statuses = list(statuses) + self.begin_calls: list[dict[str, Any]] = [] + self.commit_calls: list[str] = [] + self.retry_calls: list[str] = [] + self.cancel_calls: list[str] = [] + self.uploaded: list[str] = [] + self.closed = False + self._next_session = 0 + + # -- endpoints used by FolderIngest -- + def begin(self, *, files: list[Any], **kwargs: Any) -> BeginResult: + self.begin_calls.append({"files": files, **kwargs}) + self._next_session += 1 + return BeginResult( + session_id=f"sess-{self._next_session}", + status="uploading", + s3=S3Plan( + bucket="b", + key_prefix="p/", + uploads=tuple( + UploadTarget(rel_path=f.rel_path, url=f"https://s3.test/{f.rel_path}") + for f in files + ), + ), + ) + + def get_status(self, session_id: str) -> StatusResult: + payload = self._statuses.pop(0) if self._statuses else {"status": "done"} + return StatusResult.from_json({"session_id": session_id, **payload}) + + def upload_file(self, target: UploadTarget, data: bytes) -> str | None: + self.uploaded.append(target.rel_path) + return f'"etag-{target.rel_path}"' + + def commit(self, *, session_id: str, **kwargs: Any) -> CommitResult: + self.commit_calls.append(session_id) + return CommitResult( + session_id=session_id, status="done", table_path="t", record_count=2 + ) + + def retry(self, session_id: str) -> StatusResult: + self.retry_calls.append(session_id) + return StatusResult.from_json( + { + "session_id": session_id, + "status": "done", + "table_path": "bio.uploads.t", + "record_count": 7, + } + ) + + def cancel(self, session_id: str) -> None: + self.cancel_calls.append(session_id) + + def close(self) -> None: + self.closed = True + + +def _job(folder: Path, client: SessionClient, **kwargs: Any) -> FolderIngest: + return FolderIngest( + folder, + "bio.uploads", + data_format="parquet", + show_progress=False, + client=client, # type: ignore[arg-type] + **kwargs, + ) + + +# --- the session handle ------------------------------------------------------ + + +def test_session_id_is_exposed_after_begin(data_folder: Path) -> None: + client = SessionClient() + job = _job(data_folder, client) + assert job.session_id is None + job.begin() + assert job.session_id == "sess-1" + + +def test_attach_binds_to_an_existing_session(data_folder: Path) -> None: + client = SessionClient({"status": "failed", "failed_stage": "load"}) + job = FolderIngest.attach( + "sess-earlier", + data_folder, + "bio.uploads", + data_format="parquet", + show_progress=False, + client=client, + ) + assert job.session_id == "sess-earlier" + assert job.status().status == "failed" + + +def test_commit_before_begin_is_a_state_error(data_folder: Path) -> None: + job = _job(data_folder, SessionClient()) + with pytest.raises(IngestStateError): + job.commit() + + +def test_cancel_discards_the_session(data_folder: Path) -> None: + client = SessionClient() + job = _job(data_folder, client) + job.begin() + job.cancel() + assert client.cancel_calls == ["sess-1"] + assert job.session_id is None + + +# --- retry ------------------------------------------------------------------- + + +def test_retry_resumes_the_load_without_re_uploading(data_folder: Path) -> None: + # The stranded-transfer case: the upload landed, Dremio failed, the server + # kept the staged files. Only the load re-runs. + client = SessionClient( + { + "status": "failed", + "failed_stage": "load", + "resumable": True, + "error": "Dremio load failed: boom", + } + ) + job = _job(data_folder, client) + job.begin() + client.uploaded.clear() + + outcome = job.retry() + + assert client.retry_calls == ["sess-1"] + assert client.uploaded == [] # nothing re-uploaded + assert client.begin_calls == [client.begin_calls[0]] # no new session + assert outcome.resumed is True + assert outcome.commit.record_count == 7 + + +def test_retry_re_uploads_a_load_failure_that_kept_nothing(data_folder: Path) -> None: + # A load-stage failure the server could NOT hold the bytes for still reports + # failed_stage="load". Inferring resumability from that sent retry down the + # resume path to be told the staged data was gone; the server's own verdict + # sends it to re-upload instead. + client = SessionClient( + { + "status": "failed", + "failed_stage": "load", + "resumable": False, + "error": "Dremio load failed: 'x.parquet' could not be read as Parquet", + } + ) + job = _job(data_folder, client) + job.begin() + client.uploaded.clear() + + outcome = job.retry() + + assert client.retry_calls == [] # server retry not attempted + assert len(client.begin_calls) == 2 # a second session + assert sorted(client.uploaded) == ["a.parquet", "sub/b.parquet"] + assert outcome.resumed is False + + +def test_retry_after_an_upload_failure_starts_a_fresh_session(data_folder: Path) -> None: + # The staged files were cleaned up, so there is nothing to resume: upload + # again under a new session. + client = SessionClient( + {"status": "failed", "failed_stage": "upload", "error": "S3 upload failed"} + ) + job = _job(data_folder, client) + job.begin() + client.uploaded.clear() + + outcome = job.retry() + + assert client.retry_calls == [] # server retry not attempted + assert len(client.begin_calls) == 2 # a second session + assert sorted(client.uploaded) == ["a.parquet", "sub/b.parquet"] + assert outcome.resumed is False + assert job.session_id == "sess-2" + + +def test_retry_drops_the_idempotency_key_when_re_uploading(data_folder: Path) -> None: + # Reusing the key would make the server replay the SAME failed session, which + # is exactly the loop that made a retry look like it did nothing. + client = SessionClient({"status": "failed", "failed_stage": "upload"}) + job = _job(data_folder, client, idempotency_key="key-1") + job.begin() + assert client.begin_calls[0]["idempotency_key"] == "key-1" + + job.retry() + + assert client.begin_calls[1]["idempotency_key"] is None + + +@pytest.mark.parametrize("state", ["pending", "uploading", "committing"]) +def test_retry_refuses_while_still_running(data_folder: Path, state: str) -> None: + client = SessionClient({"status": state}) + job = _job(data_folder, client) + job.begin() + with pytest.raises(IngestStateError, match=state): + job.retry() + + +def test_retry_refuses_a_finished_transfer(data_folder: Path) -> None: + client = SessionClient({"status": "done"}) + job = _job(data_folder, client) + job.begin() + with pytest.raises(IngestStateError, match="already succeeded"): + job.retry() + + +def test_retry_without_a_session_is_a_state_error(data_folder: Path) -> None: + job = _job(data_folder, SessionClient()) + with pytest.raises(IngestStateError): + job.retry() + + +# --- status flags ------------------------------------------------------------ + + +def test_is_resumable_follows_the_servers_verdict() -> None: + kept = StatusResult.from_json( + {"status": "failed", "failed_stage": "load", "resumable": True} + ) + # Same stage, but the server cleaned up: NOT resumable. Deriving this from + # failed_stage alone is what made retry ask to resume data that was gone. + cleaned = StatusResult.from_json( + {"status": "failed", "failed_stage": "load", "resumable": False} + ) + swept = StatusResult.from_json({"status": "failed", "failed_stage": "upload"}) + running = StatusResult.from_json({"status": "committing", "resumable": True}) + # A server too old to send the field reads as not resumable — the safe way to + # be wrong, since the transfer is simply re-uploaded. + silent = StatusResult.from_json({"status": "failed", "failed_stage": "load"}) + assert kept.is_resumable is True + assert cleaned.is_resumable is False + assert swept.is_resumable is False + assert running.is_resumable is False + assert silent.is_resumable is False + + +def test_is_terminal_covers_every_end_state() -> None: + for state in ("done", "failed", "cancelled"): + assert StatusResult.from_json({"status": state}).is_terminal is True + assert StatusResult.from_json({"status": "uploading"}).is_terminal is False From 0cd06872befa0046a51bbdc46a98b55b88006073 Mon Sep 17 00:00:00 2001 From: blikij Date: Tue, 18 Aug 2026 15:47:42 +0200 Subject: [PATCH 11/16] Add docs/dev-notes.md scratchpad of findings and proposals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Observations and ideas collected about eea_datalakehouse while the development team is away. Nothing here has been applied to the code — every entry is a proposal to discuss when the team is back. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K9VStQfVW6N2nH3Mn6cSid --- docs/dev-notes.md | 805 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 805 insertions(+) create mode 100644 docs/dev-notes.md diff --git a/docs/dev-notes.md b/docs/dev-notes.md new file mode 100644 index 0000000..f017594 --- /dev/null +++ b/docs/dev-notes.md @@ -0,0 +1,805 @@ +# Dev notes — findings & ideas + +Scratchpad for observations and ideas about `eea_datalakehouse`, collected while +the development team is away. **Nothing here has been applied to the code** — +every entry is a proposal to discuss when the team is back. + +Started: 2026-08-18 · Branch at time of writing: `chore/tidy-dds-ingestion-copyover` + +## How to use this file + +- One entry per finding, newest at the bottom of its section. +- Keep entries small and self-contained: what was observed, why it matters, what + we'd suggest. No entry implies a decision has been made. +- Status: `open` (needs discussion) · `agreed` (team said yes, not yet done) · + `rejected` (discussed, deliberately not doing) · `done` (implemented elsewhere). + +### Entry template + +```markdown +### + +- **Status:** open +- **Where:** `path/to/file.py:42` +- **Observation:** what is actually there today. +- **Why it matters:** the concrete consequence. +- **Suggestion:** the smallest change that would address it. +- **Open question:** anything we need the team to decide. +``` + +## Findings + +### Dangling link to `docs/python-client-guide.md` + +- **Status:** open +- **Where:** `src/eea_datalakehouse/dds_ingestion/README.md` (last line of + "Managing the transfer") +- **Observation:** the README points at + [`docs/python-client-guide.md`](../../docs/python-client-guide.md) as the + "Full class reference". That file does not exist in the repository, and + `git log` shows it was never committed and later deleted — it has simply never + been there. +- **Why it matters:** the link is the only pointer to a full API reference for + `FolderIngest`, so a notebook author following the README hits a 404 on + GitHub. It also reads as if documentation exists that nobody can find. +- **Suggestion:** either write the guide, or drop the link and inline the short + method list that is already in that README. +- **Open question:** was the guide drafted somewhere outside this repo (wiki, + Confluence, the DDS repo) and just needs copying over? + +### The API is shaped for a developer, not for a custodian writing a one-off script + +- **Status:** open +- **Where:** `src/eea_datalakehouse/catalog/` (on `development`/`main` — **not** on + `chore/tidy-dds-ingestion-copyover`, so all line numbers below are as of + `development`), plus `pyproject.toml` +- **Premise:** the people writing this code are data custodians, working in a + command-line / batch-scripting style — short scripts, run once, per dataflow. + The library currently asks them to work like application developers instead. + +The concrete things that force the developer style, numbered below. Each is +small on its own; together they are why a five-line job takes a page of Python. +Sub-sections `1a`, `2a` and `7` are proposals rather than observations. + +#### 1. Credentials are threaded through every call site — and don't need to be + +`Catalog.__init__` (`catalog/client.py:125-136`) takes `base_url` and `token` as +required positional arguments, plus an optional `username`. Nothing resolves +them: every caller supplies all three, on every construction. + +The library's own debug harness shows the cost — `debugger/debug_run.py` repeats + +```python +catalog = Catalog(DREMIO_BASE_URL, DREMIO_TOKEN, username=DREMIO_USERNAME) +``` + +at **20+ separate call sites** (lines 143, 159, 186, 210, 220, 235, 245, 252, +259, 280, 290, 311, 330, 341, 353, 368, 387, 401, 414, …). If the authors' own +script looks like that, a custodian's will too. + +**The machinery to avoid this already exists in this same library, twice:** + +- `dds_ingestion/folder.py:136-143` — `FolderIngest` resolves its own + credentials from the environment when the caller doesn't inject a client, via + `load_creds()` / `load_base_url()` (`dds_ingestion/credentials.py:48-76`). + So one half of the library auto-resolves and the other half refuses to. +- `dds_ingestion/common/dremio_identity.py` — a full identity resolver + (`resolve()`, `endpoint()`, `dds_credentials()`) that already knows the + precedence a custodian needs: the JupyterLab "Dremio Catalog" settings panel + first, then whatever `%init` bound into the kernel, then the process + environment including the local `_DREMIO_USER` / `_DREMIO_PWD` / + `DREMIO_BASE_URL` aliases (`dremio_identity.py:177-205`, `242-260`, + `115-119`). It resolves exactly the four values `Catalog` demands: + `DREMIO_USERNAME`, `DREMIO_TOKEN`, `DREMIO_URL`, `DDS_BASE_URL`. + +So: **no, credentials do not need to be passed.** `Catalog` simply never calls +the resolver that the package next door already ships. + +- **Suggestion:** make every argument to `Catalog()` optional and fall back to + `dremio_identity.resolve()`, so `Catalog()` with no arguments works in a Hub + kernel, in the local stack, and in a cron job — with explicit arguments still + winning when someone needs to override. Same treatment for the `DREMIO_URL` + vs `DDS_BASE_URL` split, which a custodian should never have to think about. +- **Open question:** is `dremio_identity` deliberately kept inside + `dds_ingestion.common` (i.e. ingest-only), or should it move up to + `eea_datalakehouse.common` and become the one identity path for ingest, + catalog and anything later? + +#### 1a. Answering it directly: yes, `%init` is enough — with two gaps + +A real `%init` in this environment exposes: + +``` +DREMIO_USERNAME adm_bliki +DREMIO_TOKEN set (hidden, 64 chars) +DREMIO_URL https://dremio.eea.europa.eu:9047 +DREMIO_PASSWORD (not defined) +DDS_BASE_URL http://localhost:8000/ +SCHEDULER_URL http://gpu02.pdmz.eea:8080/scheduler +``` + +Lined up against what `Catalog(base_url, token, username=...)` demands: +`DREMIO_URL` → `base_url`, `DREMIO_TOKEN` → `token`, `DREMIO_USERNAME` → +`username`. **All three are already there.** Nothing needs to be typed, and +`dremio_identity.resolve()` already returns exactly this dict +(`BOUND_VARS = SECRET_VARS + SERVICE_VARS`, `dremio_identity.py:95-109`). + +`DREMIO_PASSWORD (not defined)` is not a blocker: the Flight handshake uses +`authenticate_basic_token(username, token)` — the PAT as the password, not +`DREMIO_PASSWORD` (`catalog/sql.py:255-262`, `319-321`). And the trailing slash +on `DDS_BASE_URL` is harmless — `RestSqlExecutor`, `CatalogRestClient` and +`IngestClient` all `rstrip("/")` their base URL. + +Two things do need deciding before an argument-less `Catalog()` would work +everywhere: + +**Gap 1 — `resolve()` does not read the kernel globals `%init` binds.** +`endpoint()` looks in three places: the settings panel, then `ip.user_ns` (what +`%init` bound), then the environment (`dremio_identity.py:187-201`). +`resolve()` looks in only two: the panel and the environment +(`dremio_identity.py:242-260`) — no `ip.user_ns`. It works today because the Hub +injects those variables into the *environment* and `%init` merely re-exports +them as globals, so the environment lookup happens to find the same values. But +anything that exists only as a kernel global — a value the extension binds +without exporting, or one a user reassigns in a cell — is invisible to +`resolve()` while being visible to `endpoint()`. Two resolvers, two different +precedence chains. + +**Gap 2 — `%init` exposes no Flight location, and `datacopy` is Flight-only.** +`datacopy` / `datamove` always run over Arrow Flight (`catalog/client.py:49-51`, +`205-245`), which needs a `grpc://host:port`, not the REST URL. None of the six +`%init` variables carries one. With nothing given, `_default_flight_location` +derives it from `DREMIO_URL`: `https://dremio.eea.europa.eu:9047` → +`grpc+tls://dremio.eea.europa.eu:32010` (`catalog/sql.py:69-87`) — and that +code's own comment says port 32010 is Dremio's documented default, +"unverified against this project's actual deployment". So making credentials +automatic would make `datacopy` silently depend on a guess; if the guess is +wrong it fails as `Socket closed`, which reads like a network fault rather than +a misconfiguration. + +- **Suggestion:** one resolver, used by everything. Fold `endpoint()`'s + three-source precedence into `resolve()` so both agree, then let `Catalog()` + and `FolderIngest()` default every connection argument from it — explicit + arguments still win. Add `DREMIO_FLIGHT_LOCATION` + (`catalog/sql.py:43`) to whatever `%init` exports, or verify 32010 for this + deployment and drop the "unverified" caveat. +- **Open question:** is `%init` ours to extend? If the variable list is owned by + the `jupyter_dremio` Hub extension, adding the Flight location is a + cross-repo change, and until then the library needs a documented default. + +#### 2. `idempotency_key` is mandatory everywhere and reaches nothing + +Every one of the 19 `Catalog` methods declares `*, idempotency_key: str` with +**no default** (`catalog/client.py:174-332`) — including the pure reads: +`gettablesfrom`, `gettableitemsfrom`, `getwikifrom`, `gettagsfrom`, +`getmetafromwiki`. + +Two observations about what that key actually does: + +- **It is never sent to Dremio.** The REST executor posts `{"sql": sql}` and + nothing else (`catalog/sql.py:165`); no header carries it. Its only job is to + be a lookup key in a local JSON file at + `~/.cache/eea_datalakehouse/catalog_retry_state.json` + (`catalog/retry_state.py:23-26`), written only when a call raises + `EngineStartingError`. +- **The executor layer already treats it as optional** — + `SqlExecutor.execute(sql, *, idempotency_key: str | None = None)` + (`catalog/sql.py:62`). The layer users touch is stricter than the layer that + consumes it. + +The predictable result is in the example you sent: `idempotency_key="343242dsrew"`. +A keyboard mash is the rational response to a required argument whose purpose +isn't visible at the call site — but it is also the handle you would need to +type back into `catalog.retry_pending(...)` later, so the mash quietly +forfeits the one feature the argument exists for. + +The name also over-promises. For `datacopy`, re-running with the same key does +**not** make the operation a no-op: with the default `overwrite=False` the second +run raises `CatalogOperationError: target ... already exists — pass +overwrite=True to replace it` (`catalog/operations.py:402-408`). "Idempotency +key" reads as "safe to re-run"; it isn't. + +- **Suggestion:** make it optional and fill it in behind the scenes — see 2a. +- **Open question:** was a server-side idempotency contract ever intended (a + header Dremio or DDS would honour), or is local retry bookkeeping the whole + design? + +#### 2a. Proposed: make `idempotency_key` optional and prefill it + +**Position to put to the team** (proposed here, not yet discussed with them): +the idempotency key is a sound concept for larger, multi-step transactions — +that is where knowing "this is the same unit of work as before" earns its +keep — but it is overkill for one-time execution code. It should be an +**optional** parameter, filled in behind the scenes when the caller doesn't +supply one. + +The signature change is backwards compatible and stops at the operations layer: +`idempotency_key: str | None = None` on the 19 `Catalog` methods and their +`operations.*` counterparts. Nothing below needs touching — `SqlExecutor` +already declares it optional (`catalog/sql.py:62`), and every existing caller +that passes a key keeps working unchanged. + +**How to prefill — the one real design choice.** + +*Option A: a fresh random key per call (uuid4).* Simplest to implement, but the +key is the handle you need for `catalog.retry_pending(key)`, and with a random +one the custodian never sees it. It would have to be surfaced somehow — returned +on the result, exposed as `catalog.last_idempotency_key`, or `retry_pending()` +with no argument meaning "the most recent pending one". Every stalled run also +leaves a *new* orphan entry in the retry-state file, which nothing prunes. + +*Option B (recommended): derive it deterministically* from the operation name +plus the arguments that identify the work — for `datacopy`, the resolved source +and target paths. Re-running the same script regenerates the same key, which +matters because **re-running the script is what a custodian actually does** after +a cold-engine stall. They never have to learn `retry_pending` exists: + +- first run stalls → `retry_state.record(...)` writes the derived key, attempts 1 + (`catalog/retry_state.py:82-91`); +- the custodian re-runs the same script → same derived key → attempts 2 under the + same entry, not a second orphan; +- it succeeds → `retry_state.clear(key)` removes it + (`catalog/operations.py`, end of `_run_steps` / `_run_actions`). + +The retry-state file stays bounded by the number of distinct operations, not by +the number of attempts. + +One caution worth stating so nobody trips on it later: with a derived key, two +genuinely separate runs with identical arguments (copy, delete the target, copy +again) share a key. Since the key is only local bookkeeping and is cleared on +success, that is harmless — but it is a real consequence of B and should be a +deliberate choice, not a surprise. + +**Where the parameter should stay explicit** + +- The read-only operations should not take one at all — `gettablesfrom`, + `gettableitemsfrom`, `getwikifrom`, `gettagsfrom`, `getmetafromwiki`. There is + nothing to resume; today `_fetch_step` records them anyway. +- Genuinely multi-step operations keep the explicit parameter available, because + that is the case the concept was designed for — notably `datamove`, whose own + module docstring flags the CREATE-succeeded-then-DROP-stalled gap that a blind + retry cannot resolve (`catalog/operations.py:30-37`). +- Anything a scheduler drives, where the caller wants a stable handle it chose + itself rather than one the library derived. + +- **Open question:** should `retry_state` gain an age-out or a cap regardless? + `list_pending()` exists (`catalog/retry_state.py:114`) but nothing ever prunes + the file, and with auto-filled keys entries will be created more often than + they are today. + +#### 3. There is no command line + +No `[project.scripts]` in `pyproject.toml` (neither branch), no `__main__.py`, +and no `argparse` / `click` / `typer` anywhere under `src/`. "Command-line / +batch scripting style" therefore means: write a Python file, import the right +class, construct it with credentials, call a method, print the result. + +The good news is the groundwork is already right: `catalog/operations.py` +exposes every operation as a plain module-level function taking an executor +(`operations.table2view(executor, ...)`, `datacopy(executor, ...)`, …), and +`Catalog`'s methods are thin delegations to them (`client.py:174-332`). A CLI +would be an adapter over that layer, not a rewrite. + +- **Suggestion:** one console entry point, one subcommand per verb, arguments + in the same order as the Python functions: + + ```bash + eea-catalog datacopy SRC DST --create-target-folder + eea-catalog gettablesfrom bwd --json + eea-ingest ./my_data biodiversity.uploads --format parquet --parallelism 4 + ``` + + with credentials resolved per #1 (nothing on the command line), a non-zero + exit code on failure, and `--json` for anything a shell script needs to parse. +- **Open question:** do custodians run these on their own machine, in the + project container, or as a scheduled Hub notebook? That decides whether the + CLI or a still-simpler Python one-liner is the primary surface. + +#### 4. The verbs that match the custodians' vocabulary are the unimplemented ones + +`draft2version` and `publishversion` both raise `NotImplementedError` +(`catalog/operations.py:326` and `:340`). Those are precisely the two +domain-level actions in the draft → version → publish workflow. + +Your example is that exact workflow — promoting +`…bwd.draft.bw_assessment.assessments` to +`…bwd.versions.2025_6.bw_assessment.assessments` — done by hand with the +generic `datacopy` and two fully spelled-out paths. The custodian ends up +performing the plumbing that the domain verb was meant to hide. + +- **Suggestion:** treat `draft2version` / `publishversion` as the priority, and + let them own the path arithmetic (see #5): `draft2version("bw_assessment", + version="2025_6")` rather than two 7-segment strings. +- **Open question:** what is the intended semantics — is a version a physical + copy (today's `datacopy` behaviour), or a view repoint? `publishversion`'s + docstring says the consumer view repoint, but the draft→version step is + undecided in code. + +#### 5. Catalog paths are opaque strings, though the taxonomy is known + +`datacopy` takes two free-form dotted strings. In your example they are seven +segments long and differ only in the middle: + +``` +catalog.water_management_resources.bathing_water.bwd.draft.bw_assessment.assessments +catalog.water_management_resources.bathing_water.bwd.versions.2025_6.bw_assessment.assessments +``` + +Everything before and after `draft` / `versions.2025_6` is repeated by hand, +and nothing validates the shape — `_quote_path` just splits on `.` and quotes +each segment (`catalog/operations.py:90-96`). A typo in segment 2 surfaces as +a Dremio error, not a local one. + +Meanwhile the library already encodes this taxonomy elsewhere: +`dds_ingestion/common/catalog.py` documents the path as +`{source}/{domain}/{subdomain}/{dataflow}/...` and hard-codes +`DATAFLOW_DEPTH = 3`, refusing to create anything above that level. + +- **Suggestion:** a small path type or builder shared by both packages, so the + custodian names the dataflow and the stage and the library assembles the + string — and so a wrong number of segments fails locally, before any call. +- **Open question:** is `lakehouse_structure.md` (referenced from + `dds_ingestion/common/catalog.py`) the authoritative taxonomy? If so it could + drive validation directly. + +#### 6. Batch-unfriendly output + +- `logger = logging.getLogger("eea_datalakehouse.dds_ingestion")` + (`dds_ingestion/folder.py:39`) — no handler is configured anywhere in the + package, so in a plain script every `logger.info` about resuming or + re-uploading goes nowhere unless the custodian configures logging first. +- Progress is a tqdm bar (`dds_ingestion/progress.py:37-43`), sensible in a + notebook; in a cron log it writes carriage returns, and the no-tqdm fallback + prints one line per file. +- Results come back as Python objects (`IngestOutcome`, `SqlResult`, + `TableInfo`), so reporting what happened requires more Python. + +- **Suggestion:** as part of the CLI in #3 — a default log line per step to + stderr, results to stdout as JSON on request, and progress that detects a + non-TTY and degrades to periodic lines. + +#### 7. Proposed: a session context, so paths stop being absolute + +**Position to put to the team** (proposed here, not yet discussed with them): +let the user set the session "context" once — the SQL `USE ` idea — so +the library has a default schema to resolve short paths against. This is the +direct fix for #5, and it is what makes the command-line mental model work: +`cd` once, then use relative names. + +On your own example it collapses two 7-segment strings into two short ones: + +```python +catalog.use("catalog.water_management_resources.bathing_water.bwd") +catalog.datacopy("draft.bw_assessment.assessments", + "versions.2025_6.bw_assessment.assessments") +``` + +Everything the two paths had in common moves into the `use()` line, said once. +What is left is exactly what differs — which is the part the custodian is +actually thinking about. + +**The important finding: this cannot be Dremio's own context.** + +Dremio's `POST /api/v3/sql` does accept a `context` array alongside `sql`, and +the client currently sends only `sql` (`catalog/sql.py:165`) — so passing it +through looks like the obvious one-line implementation. It isn't, because only +one of the three ways this library names things would be affected by it: + +1. **SQL DDL** — `CREATE TABLE {target} AS SELECT * FROM {source}` + (`catalog/operations.py:419-424`). A server-side context *would* apply here. +2. **INFORMATION_SCHEMA lookups** — `_entry_exists` and `_entry_kind` + (`catalog/operations.py:190-232`) and `gettablesfrom` + (`catalog/operations.py:613-650`) compare `"TABLE_SCHEMA"` against the path + as a **string literal**. `TABLE_SCHEMA` always holds the full dotted path, so + a session context does not rewrite it — a relative path simply fails to + match, `_entry_exists` returns `False`, and `datacopy` reports + `source ... does not exist` for a source that is right there. Silent and + confusing. +3. **Dremio's catalog REST API** — `_lookup_by_path` does + `"/".join(path.split("."))` against `/api/v3/catalog/by-path/…` + (`catalog/rest.py:55-60`) and `create_folder` posts `path.split(".")` + (`catalog/rest.py:261`). That API has no context concept at all; it needs the + absolute path. Folders, wiki and tags all go through it. + +So the context has to be **resolved client-side** — joined onto the front of a +relative path before anything else happens — which is also the better answer: +it keeps one code path for all three mechanisms, it works identically over REST +and Flight, and it lets a wrong path fail locally instead of as a Dremio error. +There is already precedent for the library resolving paths on the caller's +behalf: `_resolve_target_path` implements `cp source dest/` semantics +(`catalog/operations.py:234-249`). + +**Where the context should come from** — the same three-source pattern as the +credentials in #1, so a script can set it without touching code: + +- `catalog.use("…")` — mirrors the SQL the custodians already know, and can be + called more than once in a script that touches two dataflows; +- `Catalog(context="…")` — for the one-shot case; +- an environment variable (say `EEA_CATALOG_CONTEXT`), so a batch job sets it + alongside the credentials and the script body carries no paths at all. Worth + asking whether `%init` should export it too. + +**Where to put the boundary.** `dds_ingestion/common/catalog.py` already draws +one: `DATAFLOW_DEPTH = 3` — segments 0-2 (`source` / `domain` / `subdomain`) +must already exist and are never created, and from index 3 down is the +dataflow's own space. That gives a principled default context of the first +three segments (`catalog.water_management_resources.bathing_water`), leaving +`bwd.draft.…` relative. Your example suggests the more useful day-to-day +boundary is one deeper — the dataflow itself (`….bathing_water.bwd`) — since +that is what the two paths actually share. Both are defensible; the team should +pick one rather than leaving it to each script. + +**The one thing that must not be fudged: relative vs absolute.** Guessing — +"try it relative, fall back to absolute" — is the wrong answer here, because +the failure mode is silent misresolution into a real but wrong location. Most +operations would merely error, but `deletefolder(cascade=True)` +(`catalog/rest.py:340`) deletes a subtree depth-first, and `createfolder` +would quietly build a new one in the wrong place. Suggested rule: with a +context set, **every path is relative to it**, and an absolute path is written +with an explicit marker (a leading `.`, or an `absolute("…")` helper) — the +same unambiguous split a shell makes between `foo` and `/foo`. Whatever the +rule, destructive operations should echo the fully resolved path before acting. + +- **Open question:** should the context also apply to + `FolderIngest(target_catalog_path=…)`, so ingest and catalog take paths the + same way? They are the two halves of the same job and currently share no path + handling at all. +- **Open question:** one context, or a source/target pair? A draft→version + promotion has a common prefix; a copy between two dataflows does not, and + would need one of the two paths spelled out in full. + +#### 8. Proposed: `datacopy` should accept a folder and copy everything in it + +**Position to put to the team** (proposed here, not yet discussed with them): +a custodian should be able to point `datacopy` at a dataset folder and have +every table inside it copied, rather than making one call per table. + +**Today it doesn't just refuse — it refuses with the wrong reason.** +`datacopy`'s first step is `check_source_exists`, which calls `_entry_exists` +(`catalog/operations.py:190-206`): an `INFORMATION_SCHEMA."TABLES"` lookup for +`TABLE_SCHEMA = ` and `TABLE_NAME = `. A folder has no row there, +so the call fails with + +``` +CatalogOperationError: source '…bwd.draft.bw_assessment' does not exist +``` + +for a folder that plainly does exist. The library can already tell the +difference — `CatalogRestClient.is_folder()` (`catalog/rest.py:76-97`) — and it +already uses it, but only on the *other* side: `_resolve_target_path` asks +`is_folder` about the **target** to implement `cp source dest/` semantics +(`catalog/operations.py:234-249`). Folder-aware on the target, blind on the +source. Even without the feature, the message should say "is a folder" rather +than "does not exist". + +**Every building block for the real thing already exists:** + +| Need | Already there | +|---|---| +| detect a folder source | `CatalogRestClient.is_folder` (`rest.py:76`) | +| enumerate its contents | `gettablesfrom` — every table/view under a schema, **at any depth, in one query** (`operations.py:613-650`) | +| create target subfolders | `ensure_folder_path` — creates missing levels, tolerates already-there (`rest.py:275-312`) | +| tell a table from a view | `_entry_kind` (`operations.py:209-232`) | +| land inside a folder | `_resolve_target_path` (`operations.py:234-249`) | + +So this is wiring existing parts together, not new machinery. + +**And it is what `draft2version` needs anyway.** Per the taxonomy in +`dds_ingestion/common/catalog.py`, the dataset folder is +`…bwd.draft.bw_assessment` and `assessments` is a table inside it. "Copy a +dataset folder" and "promote a draft to a version" are the same operation at +the same level — so #8 is the mechanism #4 is missing, not a separate feature. + +**What the team has to decide** — none of these have an obvious default: + +- **Views.** `gettablesfrom` returns tables *and* views, but deliberately does + not select `TABLE_TYPE` (its own docstring says so). Two problems follow. + Copying a view with `CREATE TABLE … AS SELECT *` silently **materialises it + into a table** — the copy is a different kind of thing from the original. And + copying it *as* a view leaves its definition pointing at the original source + tables, so the copied folder isn't self-contained. `datamove` already treats + table/view confusion as a bug class worth auto-detecting + (`operations.py:447-457`). Cheap first step: add `TABLE_TYPE` to + `gettablesfrom`'s SELECT, so a folder copy knows per entry without N extra + round-trips. +- **Depth.** `gettablesfrom` is already recursive — its `LIKE 'schema.%'` + matches across dots, so it returns the whole subtree. That may be exactly + right for a dataset folder, but it should be a stated choice (`cp -r`), not + inherited by accident from the helper. +- **Partial failure.** One CTAS becomes N. `_run_actions` records + `step k/N` and `retry_pending` re-dispatches the operation **from the + beginning** (`operations.py:1167-1195`) — so with the default + `overwrite=False`, a resumed folder copy fails on every table that already + made it. It needs skip-what-exists, or a per-table record of what succeeded. + There is a precedent in this same library: `FolderIngest._already_done()` + skips files the server reports as already uploaded + (`dds_ingestion/folder.py:322-341`), and `retry_state` already persists a + free-form `params` dict that could carry a done-list + (`catalog/retry_state.py:40`). +- **`overwrite=True` at folder scale.** Does it mean "replace each colliding + table" or "replace the target folder"? Very different blast radius, and the + current wording doesn't extend cleanly. +- **Parallelism.** `FolderIngest` copies 4-way with a `ThreadPoolExecutor` + (`dds_ingestion/folder.py:363`). Tempting here, but `datacopy` is Flight-only + and `FlightSqlExecutor` holds one `FlightClient` with a lazily-cached + `_auth_header` — that lazy initialisation is unguarded, and this repo has not + verified concurrent statement execution on one Flight client. N concurrent + CTAS may also just queue on the Dremio engine. Worth measuring before + assuming it helps. +- **Return type.** `datacopy` returns one `SqlResult`. A folder copy needs a + per-table summary — the ingest side already has the shape for this in + `IngestOutcome` (`copied` / `skipped` / per-item results, + `dds_ingestion/folder.py:57-71`). + +- **Caveat for whoever implements it:** `is_folder` is best-effort by design — + it returns `False` on *any* lookup failure, because some Dremio source types + (the internal Arctic/Nessie-backed ones) answer a nested by-path lookup with + a 400 rather than a 404 (`catalog/rest.py:84-96`). On such a source, folder + detection silently degrades back to today's "does not exist". That needs a + decision too: fail loudly, or fall back to treating the path as a table. +- **Open question:** should `datamove` get the same treatment? It carries the + identical single-entry assumption, and moving a dataset folder is an equally + natural request — but copy-then-drop across N entries has a much worse + partial-failure story, which the module docstring already flags for the + single-entry case (`operations.py:30-37`). + +#### 8a. Proposed: split it into two verbs rather than overloading `datacopy` + +**Position to put to the team:** instead of `datacopy` detecting what it was +pointed at, have two explicit verbs — one for a single table, one for a whole +folder. + +This is the better shape, and not only on taste. Four concrete reasons: + +- **Detection is best-effort, so overloading is a silent-behaviour-change + risk.** `is_folder` returns `False` on any lookup failure, including the 400 + that Arctic/Nessie-backed sources answer with (`catalog/rest.py:84-96`). An + overloaded `datacopy` would quietly run the *single-table* path on a + false negative and fail with the misleading "does not exist". With the intent + declared in the verb, the same false negative becomes a clear error — "this + is not a folder" — because the library knows what the caller meant. +- **The return types genuinely differ.** A single copy returns `SqlResult`; a + folder copy needs a per-entry summary (see #8). Overloading forces a union + return that every caller has to narrow. +- **The parameters don't overlap.** Depth/recursion, per-entry parallelism and + a done-list only make sense for the folder verb; `overwrite` means different + things at the two scales. Overloading means parameters that are silently + ignored half the time — the same trap `Catalog(username=…)` already sets, + where the argument matters only if you happen to call a Flight operation. +- **It matches the naming already in use.** The existing verbs are lowercase + concatenations — `gettablesfrom`, `gettableitemsfrom`, `draft2version`, + `createfolder`. `tablecopy` / `datasetcopy` sit in that style without + introducing a new convention. + +**The one thing to settle first: "dataset" is an overloaded word here.** + +- In this project's taxonomy it means the folder — `common/catalog.py` calls + `bw_assessment` "the dataset folder", one level below `draft`. +- In **Dremio's** own vocabulary a *dataset* is a table or a view — that is what + PDS (physical dataset) and VDS (virtual dataset) stand for. + +So to a Dremio-fluent reader `datasetcopy()` would suggest copying exactly one +table, which is the opposite of the intent. Three ways out, in the order I'd +rank them: + +1. `tablecopy()` / `foldercopy()` — "folder" is already this library's own word + for the thing (`createfolder`, `deletefolder`, `ensure_folder_path`, + `is_folder`), so it stays internally consistent and collides with nothing. +2. `tablecopy()` / `datasetcopy()` as suggested, with the EEA meaning stated + loudly in the docstring — fine if the custodians' vocabulary is what matters + and Dremio's is not. +3. Keep one `datacopy` with an explicit `recursive=True` — the `cp -r` model, + which fits the command-line framing of #7, but keeps the union return type. + +- **Open question:** if this splits, what happens to `datacopy`/`datamove`? + Both are exported (`catalog/__init__.py:29-30`), used throughout + `debugger/debug_run.py`, and documented in the README — so it's an alias or a + deprecation, not a rename. And does `datamove` split the same way + (`tablemove` / `foldermove`), or stay single-entry given its worse + partial-failure story? + +#### 9. Proposed: settle the naming convention — `tableToView`, `datasetToViews` + +**Position to put to the team:** replace the `2`-as-"to" names with spelled-out +word separation — `table2view` → `tableToView` — and add the folder-level +counterpart `datasetToViews()`, matching the `tablecopy`/`datasetcopy` split in +#8a. + +**The underlying problem is bigger than the `2`: this module currently uses +three naming styles at once.** + +| Style | Names | +|---|---| +| lowercase run-together | `datacopy`, `datamove`, `deleteview`, `createfolder`, `deletefolder`, `deletetags`, `deletewiki`, `gettablesfrom`, `gettableitemsfrom`, `gettagsfrom`, `getwikifrom`, `getmetafromwiki`, `settagsto`, `setwikito`, `publishversion` | +| digit-for-word | `table2view`, `draft2version`, `setmeta2wiki` | +| snake_case | `retry_pending` | + +All 19 are public, all in the same `__all__` (`catalog/__init__.py:56-88`) and +the same class — including `retry_pending`, which is snake_case sitting beside +`gettablesfrom` in `Catalog` (`catalog/client.py:250`, `:327`). + +And the layer immediately below is *entirely* snake_case. `Catalog.createfolder` +→ `operations.createfolder` → `catalog_rest.create_folder` and +`ensure_folder_path` (`catalog/operations.py:1104-1109`). The same concept is +spelled two different ways one line apart. Same for `deletefolder` → +`delete_folder`, and for `is_folder` / `is_table_or_view` / `get_wiki` / +`set_wiki` / `get_tags` / `set_tags`, plus `load_creds`, `load_base_url`, +`scan_folder`, `ingest_folder`, `resolve_executor`, `list_pending` elsewhere in +the library. + +So this is one decision covering all 19 names, not a fix for three of them. + +**Which convention.** The instinct is right either way — `gettableitemsfrom` is +hard to read and `2` is not a word. The two candidates: + +- **camelCase** (`tableToView`, `datasetToViews`, `getTablesFrom`) — as + proposed. Reads well, and it matches the JSON world these operations talk to: + Dremio's own fields are camelCase (`entityType`, and the settings keys + `ddsServerUrl` / `accessToken` in `dremio_identity.py:85-90`). +- **snake_case** (`table_to_view`, `dataset_to_views`, `get_tables_from`) — + PEP 8, and it matches the ~20 functions in this library that already use it, + including the exact methods these verbs delegate to. It is also the only + option that can be *enforced*: ruff's `N` (pep8-naming) rules would keep it + from drifting again, and `select` currently omits `N` + (`pyproject.toml:53`) so nothing catches the drift today. Choosing camelCase + means `N` can never be turned on without a wall of `noqa`. + +My recommendation is snake_case, for the enforceability and for consistency +with the layer underneath — but the important thing is that one of them is +chosen for all 19 rather than the styles continuing to mix. Either beats what +is there now. + +**The verb matrix, if #8a and this both go ahead** (camelCase spelling shown +as proposed; substitute snake_case throughout if that wins): + +| single entry | whole folder | +|---|---| +| `tableToView` | `datasetToViews` | +| `tableCopy` | `datasetCopy` | +| `tableMove` | `datasetMove` | + +- Note the plural in `datasetToViews` — deliberate, since the operation yields + many views, but it makes the set irregular next to `datasetCopy`. Worth + deciding whether plurals track the output or the names stay uniform. +- `draft2version` and `setmeta2wiki` are the other two `2` names and should move + in the same sweep (`draftToVersion`, `setMetaToWiki`). +- The "dataset" vs Dremio's PDS/VDS meaning of the word, raised in #8a, applies + here too: `datasetToViews` reads to a Dremio-fluent user as "turn one dataset + into several views". + +**Migration.** These names are already released — tags up to `v0.1.10`, and the +README tells notebooks to install from `@main` / `@staging`, so custodians' +existing scripts import them. They are also used throughout +`debugger/debug_run.py` and documented in both READMEs. So: rename in one sweep, +keep the old names as thin aliases emitting a `DeprecationWarning` for one +release, then drop them. Worth doing now precisely because adoption is still +small — the cost only grows. + +- **Open question:** does the rename extend to the `Catalog` class's own + parameters (`target_catalog_path`, `conflict_mode`) and to `dds_ingestion`'s + public surface, or is it scoped to the catalog verbs? + +#### 10. Proposed: `createfolder` / `deletefolder` should take more than one folder + +**Position to put to the team:** both should accept a set of folders, not a +single path — `mkdir -p a b c` and `rm -r a b c` take multiple operands, and +that is the shell model these verbs are already imitating (their own docstrings +say "`rmdir` vs `rm -r`", `catalog/operations.py:1131`). + +Today both are strictly single-path: `createfolder(catalog_rest, path, *, +create_parents=False, …)` (`catalog/operations.py:1070`) and +`deletefolder(catalog_rest, path, *, cascade=False, …)` +(`catalog/operations.py:1119`). Setting up a dataflow means one call per folder. + +**This is the cheapest of the batch features to add** (cf. #8), because both +underlying operations are already idempotent: + +- `create_folder` tolerates a 409 and reports whether it newly created the + folder (`catalog/rest.py:255-273`); +- `delete_folder` returns early when the folder is already gone + (`catalog/rest.py:352-354`). + +So a partially-completed batch can simply be re-run — which matters, because a +batch shares one idempotency key and `retry_pending` re-dispatches the operation +**from the start** (`catalog/operations.py:1167-1195`). For #8's copy that is a +real problem; here it is harmless. + +**What the team has to decide:** + +- **Ordering is not the order the user typed, and it differs by verb.** + - Deleting without `cascade` must go **deepest-first**, or removing a parent + before its child fails with "folder is not empty" + (`catalog/rest.py:359-364`). + - Creating without `create_parents` must go **shallowest-first**, or creating + `a.b.c` before `a.b` fails with "parent does not exist" + (`catalog/operations.py:1098-1102`). + + Opposite sorts, and a naive `for p in paths:` loop gets both wrong. Sorting by + depth inside the batch verb makes a list of related folders "just work" — + which is most of the value of the feature. +- **Report what happened.** Both currently return `None`, and + `createfolder(create_parents=True)` **throws away information it already + has**: `ensure_folder_path` returns the list of levels it actually created + (`catalog/rest.py:275-312`) and `create_folder` returns whether the folder was + new — then `createfolder` discards both. A batch needs a per-path result + (created / already existed / failed) and the pieces are already there. +- **Fail-fast or continue-on-error?** For creates, continuing and reporting is + probably right. For deletes it is not obviously right, and the default should + be a deliberate choice rather than whatever falls out of the loop. +- **Echo the resolved paths before deleting.** Especially once #7's context + makes the written paths relative — a batch `deletefolder(cascade=True)` is the + single most destructive call in this library. + +**On API shape — this one can safely accept both, unlike #8a.** The argument +against overloading `datacopy` was that folder-vs-table detection is +best-effort and can silently pick the wrong path. Here there is no detection: +`str` vs `Sequence[str]` is unambiguous at runtime, so `path: str | +Sequence[str]` keeps every existing call working and needs no new verb. If the +team prefers explicit plural verbs anyway (`createfolders` / `deletefolders`), +that is a naming call under #9, not a correctness one. + +- **Note for whoever implements it:** `deletefolder` is the only operation that + bypasses the shared `_run_steps` / `_run_actions` helpers and hand-rolls its + own record-and-clear (`catalog/operations.py:1135-1142`). A batch version + should go through the shared helper so step numbering and retry recording + behave like everything else. +- **Note:** the `_OPERATIONS` registry that `retry_pending` dispatches through + (`catalog/operations.py:1145-1163`) is keyed by operation name — any new or + renamed verb from #9 or #8a has to be added there too, or its retries break + with a `KeyError`. + +#### What "simple" could look like + +Sketch to argue about, not a proposal to implement — the same job as your +example, with #1, #2, #4, #5 and #7 addressed. Each line that disappears is one +of the findings above: + +```python +from eea_datalakehouse.catalog import Catalog + +# no URL, no token, no username — resolved from the panel / %init / env (#1) +with Catalog() as catalog: + catalog.use("catalog.water_management_resources.bathing_water.bwd") # (#7) + catalog.datacopy("draft.bw_assessment.assessments", # (#2: no key) + "versions.2025_6.bw_assessment.assessments") +``` + +and once the domain verb exists (#4), with the context supplying everything +above the dataflow: + +```python +with Catalog() as catalog: + catalog.use("catalog.water_management_resources.bathing_water.bwd") + catalog.draft2version("bw_assessment", version="2025_6") +``` + +or, with #3, nothing in Python at all: + +```bash +export EEA_CATALOG_CONTEXT=catalog.water_management_resources.bathing_water.bwd +eea-catalog draft2version bw_assessment --version 2025_6 +``` + +Compare against what the same job takes today — full credentials, a hand-made +idempotency key, and the taxonomy typed twice: + +```python +with Catalog(DREMIO_URL, DREMIO_TOKEN, username=DREMIO_USERNAME) as catalog: + catalog.datacopy( + "catalog.water_management_resources.bathing_water.bwd.draft.bw_assessment.assessments", + "catalog.water_management_resources.bathing_water.bwd.versions.2025_6.bw_assessment.assessments", + create_target_folder=True, + idempotency_key="343242dsrew", + ) +``` + +## Ideas / parking lot + +_Larger or fuzzier things that aren't defects — worth a conversation, not a ticket yet._ + +## Questions for the team + +_Things we can't resolve from the code alone._ + +- Is `docs/` intended to become a real documentation folder in this repo, or is + documentation hosted elsewhere? From af9dcab59254d8636eaa69a7803bfb1336aed0b7 Mon Sep 17 00:00:00 2001 From: blikij Date: Tue, 18 Aug 2026 16:35:07 +0200 Subject: [PATCH 12/16] Add dev notes on docstring coverage and the Jupyter completion surface Three proposals, all still open for discussion: - half the public surface has no docstring (35% on this branch, 49% with catalog/ on development), which is exactly what Shift+Tab and `obj?` render in a notebook; - Catalog's methods are thin delegates that drop the docstrings operations.* already has, so none of that text reaches the object a notebook holds; - nothing in the repo pins the JupyterLab 4 environment the library is written for, and py.typed covers only dds_ingestion/. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K9VStQfVW6N2nH3Mn6cSid --- docs/dev-notes.md | 118 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 118 insertions(+) diff --git a/docs/dev-notes.md b/docs/dev-notes.md index f017594..04f01d5 100644 --- a/docs/dev-notes.md +++ b/docs/dev-notes.md @@ -793,6 +793,120 @@ with Catalog(DREMIO_URL, DREMIO_TOKEN, username=DREMIO_USERNAME) as catalog: ) ``` +### Half the public surface has no docstring — and a notebook is where that shows + +- **Status:** open +- **Where:** the whole package; worst in `dds_ingestion/models.py`, + `dds_ingestion/progress.py`, `dds_ingestion/client.py` (this branch) and + `catalog/client.py` (on `development`) +- **Observation:** counting public classes/functions/methods (`__init__` + included, `_private` names excluded): + + | tree | public defs | no docstring | + | --- | --- | --- | + | this branch (`data_preparation` + `dds_ingestion`) | 82 | 29 (35%) | + | `development` (adds `catalog/`) | 152 | 75 (49%) | + + Note what is *not* missing: type information. `mypy` runs `strict = true` + (`pyproject.toml:56`) and a scan of both trees finds no unannotated parameter + and one unannotated return in total. Only the prose is absent, and unevenly: + + - `catalog/client.py` — 20 of 22 public members, including every verb + (`table2view`, `draft2version`, `publishversion`, `datacopy`, `datamove`, + …). Its own class docstring is thorough; the methods under it have none. + See the next entry — those docstrings do exist, one layer down. + - `dds_ingestion/client.py` — `IngestClient.begin:107`, `.commit:136`, + `.close:90`, and both `__init__`s. + - `dds_ingestion/models.py` — 15 of 27, i.e. every `from_json`/`as_payload` + and the `Progress:134` class itself. + - `dds_ingestion/progress.py` — the `ProgressBar:13` protocol's `update`/ + `close`, though the module docstring does spell the contract out. + - `data_preparation/` — classes and methods are documented well; but none of + the four modules, nor the package `__init__.py`, has a module docstring. +- **Why it matters:** this library is driven from a notebook by design — + `dds_ingestion/__init__.py` opens with "A generated Jupyter notebook imports + :class:`FolderIngest`…". In JupyterLab, `Shift+Tab` inside a call and `obj?` + in a cell both render the same two things and nothing else: the signature and + the docstring. Where the docstring is empty a custodian gets a bare signature + and has to leave the notebook and open the source to learn what `commit()` + returns or what `Progress` carries. We have documentation — in READMEs — but + it is not reachable from where people are typing, which is the one place the + question actually gets asked. +- **Suggestion:** a one-line summary on everything named in an `__all__`, in the + NumPy style `data_preparation/transformation.py:19` already uses, plus module + docstrings for the four `data_preparation` modules. To stop it drifting back, + turn on ruff's pydocstyle rules for `src/`: `pyproject.toml:50` currently + selects `E,F,I,UP,B,C4,SIM`, so nothing has ever checked for a docstring — + adding `"D"` with `[tool.ruff.lint.pydocstyle] convention = "numpy"` would. +- **Open question:** do we want this on the `from_json`/`as_payload` wire + helpers too, or only on the surface a notebook touches? A `D` rule set covers + both unless we add a per-file-ignore for `models.py`. + +### `Catalog`'s methods drop the docstrings that `operations` already wrote + +- **Status:** open +- **Where:** `src/eea_datalakehouse/catalog/client.py:174-249` vs + `catalog/operations.py` (on `development`) +- **Observation:** all 19 public verbs in `operations.py` carry a real + docstring — `datacopy:343` "Copy data from `source_path` into `target_path`.", + `draft2version:315` "Promote the draft table at `draft_path` into a permanent + `version_path`.", and so on down the list. `Catalog`'s methods are thin + delegates onto them (`client.py:174` `table2view` → `operations.table2view`, + `:205` `datacopy` → `operations.datacopy`, same for the rest) and **not one + delegate has a docstring**, so none of that text reaches the object a notebook + actually holds. +- **Why it matters:** the writing is already done and paid for; it is just one + layer below where anyone looks. `Shift+Tab` on `catalog.datacopy(` — the + natural way to check argument order mid-cell — shows the signature and + nothing else, while `operations.datacopy?` shows the paragraph that would have + answered the question. `help(Catalog)` has the same hole. +- **Suggestion:** the cheapest fix that cannot drift is to copy the text at + class-definition time instead of retyping it — `datacopy.__doc__ = + operations.datacopy.__doc__` after the class body, or a tiny + `_delegates_to(operations.datacopy)` decorator that assigns `__doc__` **only**. + Deliberately not `functools.wraps`: that also rebinds `__name__`, + `__wrapped__` and `__signature__`, so the inspector would start showing the + module function's signature — including the leading `executor` argument that + `Catalog` supplies itself, which is exactly the confusion we'd be trying to + remove. If we'd rather stay explicit and boring, a one-line summary plus "See + `operations.datacopy`." is 19 lines of typing, once. +- **Open question:** is `Catalog` the documented surface and `operations` the + internals, or are both public? `catalog/__init__.py` exports both and its + module docstring points at `operations` for callers who want to manage the + executor themselves — which argues both, and so argues for docstrings that + read correctly in either place. + +### Nothing pins the notebook environment the library is written for + +- **Status:** open +- **Where:** `pyproject.toml:13-32` (`dependencies`) and `:34-41` + (`[project.optional-dependencies]`) +- **Observation:** the package is notebook-facing by design, but nothing in the + repo installs, pins, or exercises a notebook: no `jupyterlab` or `ipykernel` + in the runtime dependencies or in the `dev` extra, no `.ipynb` anywhere in the + tree, and `tqdm` — the bar `dds_ingestion/progress.py` drives — comes in as + the plain terminal build rather than its notebook widget variant. +- **Why it matters:** if we want to tell custodians "the API reference is + `Shift+Tab`", we need to be able to say *in which Jupyter*. Completion + behaviour is version-dependent (JupyterLab 4 is where the inline completer and + the LSP integration became first-class), and "it completes for me, not for + you" is unanswerable without a reproducible environment. It also means nobody + has ever run the notebook path in CI. +- **Suggestion:** add a `notebook` extra — `jupyterlab>=4,<5` and `ipykernel`, + optionally `jupyterlab-lsp` + `python-lsp-server` if we want static completion + as well as the runtime kind — so `pip install -e ".[notebook]"` reproduces + what we document against. Two related loose ends worth folding in: `py.typed` + exists only under `dds_ingestion/`, not in `catalog/`, `data_preparation/` or + the package root, so a type checker or an LSP treats the rest of the package + as untyped despite `mypy strict`; and `pyproject.toml` declares no package + data at all, so it is worth confirming the marker actually lands in a built + wheel (`python -m build && unzip -l dist/*.whl | grep py.typed`). +- **Open question:** do custodians run JupyterLab themselves, or in a hosted + environment where we control neither the Lab version nor the installed + extensions? That decides whether static/LSP completion is worth chasing at + all — if it isn't, everything rests on runtime introspection, which is to say + on the docstrings in the two entries above. + ## Ideas / parking lot _Larger or fuzzier things that aren't defects — worth a conversation, not a ticket yet._ @@ -803,3 +917,7 @@ _Things we can't resolve from the code alone._ - Is `docs/` intended to become a real documentation folder in this repo, or is documentation hosted elsewhere? +- Which Jupyter do custodians actually run — JupyterLab 4 locally, an older + Notebook, or a hosted environment we don't control? If we intend tab-completion + and `Shift+Tab` to be the day-to-day API reference, that version is a + supported-platform decision, not a preference. From d82bad046a5e16ab91ae27bee51183d8aaab8e2f Mon Sep 17 00:00:00 2001 From: blikij Date: Fri, 21 Aug 2026 15:13:46 +0200 Subject: [PATCH 13/16] docs: brief the client half of read-only permanent ingest (DI-11) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The design lives with the server work it depends on, in the DDS repo's devplans; this records what is built HERE and when — the sub_path parameter, storage_path, the resume wording that a permanent folder changes, and the two things noted but deliberately not built: deleting a read-only table with its backing data, and whether the client-side ensure_catalog_path is now redundant. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VV1gKHekMUx69LJKSjP8Q3 --- docs/read-only-ingest-client-plan.md | 109 +++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 docs/read-only-ingest-client-plan.md diff --git a/docs/read-only-ingest-client-plan.md b/docs/read-only-ingest-client-plan.md new file mode 100644 index 0000000..5e6837a --- /dev/null +++ b/docs/read-only-ingest-client-plan.md @@ -0,0 +1,109 @@ +# Read-only ingest — the client half (DI-11.9) + +Companion note for **this** repository. The full design lives with the server +work it depends on: + +``` +EEALakeHouse/Components/DremioDocumentService/devplans/2026-08-21/ + Development Plan- Read-Only Ingest to Permanent S3 (DI-11).md +``` + +Read that first — this file only records what is built **here**, and when. + +## The short version + +`FolderIngest` already has the attribute: `intent="read_only" | "editable"` +(`src/eea_datalakehouse/dds_ingestion/folder.py:103`, `models.py:13`), sent to +`POST /ingest/begin` (`client.py:121`). On EEA production's managed catalog the +server currently ignores it for placement — every ingest stages to a temp prefix, +is copied into the catalog's Iceberg storage, and the upload is deleted. + +DI-11 makes `read_only` mean *store the files permanently* under +`local_s3/dh-prod-data/read///` and register them. +**Where the bytes land is chosen by the server**, in `begin`'s presigned targets, +so this package needs no new parameter and no new call. + +## What changes here + +0. **`sub_path` — the one new parameter** (DI-11.12). `FolderIngest(..., + sub_path="2026")`, passed straight through to `begin`; the server validates, + normalises and scopes on it. It is what lets a read-only table accumulate: + + ```python + FolderIngest(folder="./bw_2026", target_catalog_path="…/bwd/reference", + data_format="parquet", intent="read_only", + table_name="water_temperature", sub_path="2026").run() + # → read/…/water_temperature/2026/*.parquet, alongside 2024/ and 2025/ + ``` + + A custodian whose local folder *already* has the structure needs nothing — + `scan_folder` (`folder.py:80-92`) preserves sub-folders and the server keeps + the original `rel_path`. `sub_path` is for filing a **flat** folder under a + name they choose. Worth a worked example in `README.md`, and note that + `replace` with a `sub_path` clears only that year. +1. **Docs and docstrings** — `folder.py` class docstring + the `intent` + parameter, `dds_ingestion/README.md:38`, `__init__.py:16`: say what each + intent now does (permanent raw folder + catalog view vs. Iceberg copy). +2. **`CommitResult.storage_path`** (`models.py`) — optional, additive; the + physical Dremio path the server reports for a permanent ingest, so a notebook + can print where the files actually are. Parse permissively as everywhere else + (absent ⇒ `None`), so it works against an older server. +3. **Resume/retry note** (`folder.py:305-322`, `retry()`): `_already_done`'s + docstring says a re-run is "an S3 overwrite of the identical key, which is + harmless". Against a permanent folder in `append` mode that is only true if + the server reuses the keys instead of numbering them. Align the wording with + whatever DI-11.7 settles, and steer users to `attach()` + `retry()` rather + than re-running `run()`. +4. **Tests** — `tests/dds_ingestion/`: `storage_path` parsing, and a + respx-mocked `begin` whose `key_prefix` is a permanent `read/...` prefix, + asserting the client uploads to the issued targets unchanged. + +## What does not change + +- No new constructor argument, no new method, no new endpoint. +- `editable` behaviour, the upload/parallelism/progress machinery, credentials. +- Nothing has to change for the client to keep working while the server runs + with `DDS_INGEST_READ_MODE=staged` (its default). + +## When + +DI-11 **step 4** ("Prove") in the rollout table — after the server feature exists +behind its flag, and it is the first client pointed at a `permanent` server. The +JupyterLab extension is deliberately left alone until step 5. + +## Noted for later: deleting a read-only table + +Because read-only files are permanent, dropping the view or dataset from the +catalog leaves the objects in S3. The agreed home for that cleanup is a +**delete-table operation in this package** — `eea_datalakehouse.catalog` has +`deleteview` and `deletefolder` today, but nothing that removes a table together +with its backing data. **Not implemented as part of DI-11**, recorded here (and +in §8.5 of the DI-11 plan) so it is a known gap rather than a surprise. + +When it is picked up: this package holds no S3 credentials by design, so the +delete needs a DDS-side endpoint for the read folder to call — not boto3 in a +notebook. + +## Related: `common/catalog.py` moves server-side + +`dds_ingestion/common/catalog.py` (on `development`) walks a catalog path and +creates the missing dataflow levels, refusing to invent a domain or subdomain. +That walk is exactly what a read-only ingest needs before its view can be +created, so **DI-11.2 ports it into DDS**, where the extension and any REST +client get it too and the taxonomy rule is enforced in one place. Once that +lands, the client-side copy is either a thin convenience wrapper over +`POST /api/v1/catalog/{path}/folder` or redundant — decide when DI-11.2 is done, +not before. + +This is a *catalog* concern only: the `read/` prefixes in S3 need no creating at +all — see §3.6 of the DI-11 plan. + +## One thing to settle first + +There are two copies of this client: this repo +(`src/eea_datalakehouse/dds_ingestion/`) and a snapshot in the DDS repo +(`clients/dds_ingest/dds_ingest/`). They have already drifted — `folder.py`, +`client.py` and `models.py` differ; `credentials.py` and `progress.py` are +identical. **This repo is where the client is developed.** Whether the DDS-side +copy is re-synced, thinned to a test fixture, or dropped is a separate decision, +not part of DI-11 — but decide it before editing both by hand again. From 4f0159f507e12b199baef74b93ead7d55572a143 Mon Sep 17 00:00:00 2001 From: blikij Date: Fri, 21 Aug 2026 15:14:00 +0200 Subject: [PATCH 14/16] feat(dds_ingestion): name a sub-folder, and never re-upload permanent data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two client-side halves of DI-11, where the server now stores a read-only ingest's files permanently instead of staging and deleting them. `FolderIngest(..., sub_path="2026")` files an upload under a named sub-folder of the table, so a read-only dataset can accumulate a year at a time. It is passed straight through to begin: validating, normalising and scoping it are the server's job, or the client becomes a second opinion about where data lives. A folder that already has the structure locally needs nothing — scan_folder keeps sub-folders and the server preserves them. `retry()` no longer falls back to "open a fresh session and upload again" for a transfer whose files are stored permanently. That fallback is right when the staged copy was discarded and actively harmful here: the server numbers an incoming name that already exists, precisely so an append cannot overwrite live data, which turns a silent re-run into duplicated rows. It now raises and says what to do instead — conflict_mode="replace" to redo it, or a sub_path for data that belongs beside what is there. StatusResult gains `placement` / `stores_permanently` to tell the two apart. A server without DI-11 omits the field and reads as staged, so nothing changes against one. Verified: mypy clean, 51 tests pass. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VV1gKHekMUx69LJKSjP8Q3 --- src/eea_datalakehouse/dds_ingestion/README.md | 26 +++++ src/eea_datalakehouse/dds_ingestion/client.py | 3 + src/eea_datalakehouse/dds_ingestion/folder.py | 35 ++++++- src/eea_datalakehouse/dds_ingestion/models.py | 17 ++++ tests/dds_ingestion/test_client.py | 35 +++++++ tests/dds_ingestion/test_folder.py | 94 ++++++++++++++++++- 6 files changed, 208 insertions(+), 2 deletions(-) diff --git a/src/eea_datalakehouse/dds_ingestion/README.md b/src/eea_datalakehouse/dds_ingestion/README.md index 2468700..6603410 100644 --- a/src/eea_datalakehouse/dds_ingestion/README.md +++ b/src/eea_datalakehouse/dds_ingestion/README.md @@ -46,6 +46,32 @@ print(outcome.commit.table_path, outcome.commit.record_count) `run()` performs `begin → upload(all files) → commit`, and leaves the session handle available afterwards. +## Read-only data that grows + +`intent="read_only"` on a server configured for permanent storage keeps your +files as uploaded and registers them as the table — nothing is copied into the +catalog. `sub_path` files each upload under a named sub-folder, so one table can +accumulate: + +```python +FolderIngest( + folder="./bw_2026", # a flat folder of parquet files + target_catalog_path="water_management_resources/bathing_water/bwd/reference", + data_format="parquet", + intent="read_only", + table_name="water_temperature", + sub_path="2026", # → .../water_temperature/2026/ + conflict_mode="fail", # refuses if 2026 is already there +).run() +``` + +Next year, the same call with `sub_path="2027"` adds to the same table; +`conflict_mode="replace"` with a `sub_path` re-does **that year only** and leaves +the others standing. If your local folder already has the structure +(`2026/*.parquet`), you do not need `sub_path` — the layout is preserved as-is. +The name is normalised server-side to the EEA convention (lowercase, +underscores), so `"Q1 2026"` becomes `q1_2026`. + ## Managing the transfer A transfer is a server-side session, so it can be inspected and resumed: diff --git a/src/eea_datalakehouse/dds_ingestion/client.py b/src/eea_datalakehouse/dds_ingestion/client.py index 58977f9..a6a200e 100644 --- a/src/eea_datalakehouse/dds_ingestion/client.py +++ b/src/eea_datalakehouse/dds_ingestion/client.py @@ -113,6 +113,7 @@ def begin( conflict_mode: str, files: list[FileSpec], table_name: str | None = None, + sub_path: str | None = None, idempotency_key: str | None = None, multipart: bool | None = None, ) -> BeginResult: @@ -126,6 +127,8 @@ def begin( } if table_name is not None: body["table_name"] = table_name + if sub_path: + body["sub_path"] = sub_path if idempotency_key is not None: body["idempotency_key"] = idempotency_key if multipart is not None: diff --git a/src/eea_datalakehouse/dds_ingestion/folder.py b/src/eea_datalakehouse/dds_ingestion/folder.py index 5d8a470..0afee08 100644 --- a/src/eea_datalakehouse/dds_ingestion/folder.py +++ b/src/eea_datalakehouse/dds_ingestion/folder.py @@ -92,7 +92,20 @@ def scan_folder(folder: Path, data_format: DataFormat) -> list[FileSpec]: class FolderIngest: - """Orchestrate ingest of a local folder into a Dremio table via DDS.""" + """Orchestrate ingest of a local folder into a Dremio table via DDS. + + ``sub_path`` files this upload under a named sub-folder of the table — the + accumulating-dataset shape, one year at a time:: + + FolderIngest(folder="./bw_2026", target_catalog_path=..., data_format="parquet", + intent="read_only", table_name="water_temperature", + sub_path="2026").run() + + It applies to a **read-only** ingest whose files are stored permanently; the + server refuses it otherwise rather than filing the data somewhere else. A + folder that already has the structure locally needs nothing: ``scan_folder`` + keeps sub-folders and the server preserves them. + """ def __init__( self, @@ -103,6 +116,7 @@ def __init__( intent: Intent = "read_only", conflict_mode: str = "fail", table_name: str | None = None, + sub_path: str | None = None, parallelism: int = DEFAULT_PARALLELISM, idempotency_key: str | None = None, multipart: bool | None = None, @@ -121,6 +135,7 @@ def __init__( self.intent = intent self.conflict_mode = conflict_mode self.table_name = table_name + self.sub_path = sub_path self.parallelism = parallelism self.idempotency_key = idempotency_key self.multipart = multipart @@ -219,6 +234,7 @@ def begin(self) -> BeginResult: conflict_mode=self.conflict_mode, files=files, table_name=self.table_name, + sub_path=self.sub_path, idempotency_key=self.idempotency_key, multipart=self.multipart, ) @@ -280,6 +296,15 @@ def retry(self) -> IngestOutcome: uploaded again. Any ``idempotency_key`` is dropped for that attempt, or the server would just replay the failed session. + **A transfer whose files are stored permanently is never re-uploaded + blindly.** There the upload landed in the table's own folder and stayed, + so a fresh session would add a *second* copy — the server numbers an + incoming name that already exists, precisely so an append can never + overwrite live data, and that protection turns a silent re-run into + duplicated rows. Such a transfer is resumable server-side by design, so + the first branch handles it; if the server says it is not, this raises + rather than guessing. + Raises :class:`IngestStateError` if there is nothing to retry — no session, one that is still running, or one that already succeeded. """ @@ -305,6 +330,14 @@ def retry(self) -> IngestOutcome: files_skipped=len(self._begin.s3.uploads) if self._begin else 0, resumed=True, ) + if status.stores_permanently: + raise IngestStateError( + f"transfer {self._session_id} stored its files permanently and " + "cannot be resumed, so re-running it would upload a second copy " + "alongside the first. Inspect the table, then re-ingest " + "deliberately — with conflict_mode='replace' to redo it, or a " + "sub_path for data that belongs beside what is already there." + ) # Nothing to resume server-side: start over with a new session, which # means dropping the idempotency key that would replay the failed one. logger.info( diff --git a/src/eea_datalakehouse/dds_ingestion/models.py b/src/eea_datalakehouse/dds_ingestion/models.py index 7dddb4e..ef14ba0 100644 --- a/src/eea_datalakehouse/dds_ingestion/models.py +++ b/src/eea_datalakehouse/dds_ingestion/models.py @@ -181,6 +181,23 @@ def record_count(self) -> int | None: value = self.raw.get("record_count") return int(value) if value is not None else None + @property + def placement(self) -> str: + """Where this transfer's files live: ``staged`` or ``read_permanent``. + + ``staged`` (the default, and what a server without DI-11 reports by + omitting the field) means the upload was copied into the catalog and + deleted. ``read_permanent`` means the files were stored where the table + lives and kept — so re-uploading them is not a safe way to recover. + """ + value = self.raw.get("placement") + return str(value) if value else "staged" + + @property + def stores_permanently(self) -> bool: + """Whether this transfer's uploaded files ARE the table (DI-11).""" + return self.placement == "read_permanent" + @property def error(self) -> str | None: """Why the transfer failed, including the stage — e.g. ``"Dremio load diff --git a/tests/dds_ingestion/test_client.py b/tests/dds_ingestion/test_client.py index b260448..3201e73 100644 --- a/tests/dds_ingestion/test_client.py +++ b/tests/dds_ingestion/test_client.py @@ -175,3 +175,38 @@ def test_error_response_raises_with_message(creds: DremioCreds) -> None: ) assert exc.value.status_code == 409 assert "exists" in str(exc.value) + + +@respx.mock +def test_begin_sends_sub_path_only_when_given(creds: DremioCreds) -> None: + """DI-11.12: the named sub-folder rides on ``begin`` and nowhere else. + + Omitted when unset, so a client that never uses it sends the body it always + sent — an older server sees no new field. + """ + import json + + route = respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock( + return_value=httpx.Response( + 200, + json={ + "session_id": "sess-1", + "status": "open", + "s3": {"bucket": "b", "key_prefix": "p/", "uploads": []}, + "collision": None, + }, + ) + ) + kwargs: dict[str, object] = { + "target_catalog_path": "bio.uploads", + "intent": "read_only", + "data_format": "parquet", + "conflict_mode": "append", + "files": [FileSpec("a.parquet", 6)], + } + with IngestClient(BASE_URL, creds) as client: + client.begin(**kwargs) # type: ignore[arg-type] + assert "sub_path" not in json.loads(route.calls.last.request.read()) + + client.begin(sub_path="2026", **kwargs) # type: ignore[arg-type] + assert json.loads(route.calls.last.request.read())["sub_path"] == "2026" diff --git a/tests/dds_ingestion/test_folder.py b/tests/dds_ingestion/test_folder.py index 99fcacd..c2b6025 100644 --- a/tests/dds_ingestion/test_folder.py +++ b/tests/dds_ingestion/test_folder.py @@ -6,7 +6,11 @@ from pathlib import Path import pytest -from eea_datalakehouse.dds_ingestion.folder import FolderIngest, scan_folder +from eea_datalakehouse.dds_ingestion.folder import ( + FolderIngest, + IngestStateError, + scan_folder, +) from eea_datalakehouse.dds_ingestion.models import ( BeginResult, CommitResult, @@ -28,6 +32,9 @@ def __init__(self, *, already_uploaded: list[str] | None = None) -> None: self.commit_calls: list[dict[str, object]] = [] self.closed = False self._already_uploaded = already_uploaded or [] + # Set to drive ``get_status`` from a test (e.g. a failed, unresumable + # session); ``None`` keeps the default "open" shape. + self.status_payload: dict[str, object] | None = None # concurrency instrumentation self._lock = threading.Lock() self._active = 0 @@ -46,6 +53,8 @@ def begin(self, *, files: list[FileSpec], **kwargs: object) -> BeginResult: ) def get_status(self, session_id: str) -> StatusResult: + if self.status_payload is not None: + return StatusResult.from_json(self.status_payload) return StatusResult.from_json( { "status": "open", @@ -204,3 +213,86 @@ def test_multipart_etags_threaded_into_commit(data_folder: Path) -> None: entry = by_rel["a.parquet"] assert entry["upload_id"] == "up-a.parquet" assert entry["parts"] == [{"part_number": 1, "etag": '"etag-a.parquet"'}] + + +def test_sub_path_is_passed_through_to_begin(tmp_path: Path) -> None: + """DI-11.12: the class carries the custodian's chosen sub-folder, nothing more. + + Normalising and scoping it is the server's job — the client would otherwise + be a second opinion about where the data lives. + """ + (tmp_path / "a.parquet").write_bytes(b"PAR1") + client = FakeClient() + job = FolderIngest( + tmp_path, + "catalog/water/bwd", + data_format="parquet", + table_name="water_temperature", + sub_path="2026", + client=client, + show_progress=False, + ) + job.run() + assert client.begin_calls[0]["sub_path"] == "2026" + + +def test_sub_path_defaults_to_none(tmp_path: Path) -> None: + (tmp_path / "a.parquet").write_bytes(b"PAR1") + client = FakeClient() + FolderIngest( + tmp_path, "catalog/water/bwd", data_format="parquet", + client=client, show_progress=False, + ).run() + assert client.begin_calls[0]["sub_path"] is None + + +def test_retry_refuses_to_re_upload_a_permanently_stored_transfer( + tmp_path: Path, +) -> None: + """DI-11.7: a blind re-run there adds a second copy, so it must not happen. + + The server numbers an incoming name that already exists — precisely so an + append can never overwrite live data — which turns a silent "start over" + into duplicated rows rather than a clean retry. + """ + (tmp_path / "a.parquet").write_bytes(b"PAR1") + client = FakeClient() + client.status_payload = { + "status": "failed", + "progress": {}, + "failed_stage": "load", + "resumable": False, + "placement": "read_permanent", + } + job = FolderIngest( + tmp_path, "catalog/water/bwd", data_format="parquet", + intent="read_only", client=client, show_progress=False, + ) + job._session_id = "sess-1" + + with pytest.raises(IngestStateError) as excinfo: + job.retry() + assert "second copy" in str(excinfo.value) + # And nothing was re-uploaded behind the user's back. + assert client.begin_calls == [] + assert client.uploaded == [] + + +def test_retry_still_starts_over_for_a_staged_transfer(tmp_path: Path) -> None: + # The staged path is unchanged: its files were discarded, so re-uploading is + # the only way forward and is safe. + (tmp_path / "a.parquet").write_bytes(b"PAR1") + client = FakeClient() + client.status_payload = { + "status": "failed", + "progress": {}, + "failed_stage": "upload", + "resumable": False, + } + job = FolderIngest( + tmp_path, "catalog/water/bwd", data_format="parquet", + client=client, show_progress=False, + ) + job._session_id = "sess-1" + job.retry() + assert len(client.begin_calls) == 1 From bfa2f809292d4dc21fa3b4803bcd22da5ab1b8c4 Mon Sep 17 00:00:00 2001 From: blikij Date: Fri, 21 Aug 2026 15:54:18 +0200 Subject: [PATCH 15/16] docs(dds_ingestion): say what intent decides, and report storage_path (DI-11.9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `CommitResult` parses `storage_path` — where a permanently-stored read-only table's files physically are, against `table_path`, which stays where the table is queried. Absent parses as None, so an older server is unaffected. The docs described `intent` as a choice of table shape, which is what it meant before the server could keep the upload. The class docstring, the README and the package docstring now make the same distinction in the space each has: read-only keeps your files and registers them; editable copies them in and deletes the upload; and permanent storage is a server setting, so where it is off both intents stage as before. The resume note is the one that was actively misleading. "An S3 overwrite of the identical key, which is harmless" is true of re-running the SAME session and false of starting a new one against permanent storage — the server numbers a colliding name to protect live data, so the re-run leaves a second copy instead of overwriting. It now says so and points at attach() + retry(). Verified: mypy clean, 52 tests pass. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VV1gKHekMUx69LJKSjP8Q3 --- docs/read-only-ingest-client-plan.md | 18 +++++++----- src/eea_datalakehouse/dds_ingestion/README.md | 15 +++++++++- .../dds_ingestion/__init__.py | 8 +++++- src/eea_datalakehouse/dds_ingestion/folder.py | 28 +++++++++++++++++-- src/eea_datalakehouse/dds_ingestion/models.py | 11 +++++++- tests/dds_ingestion/test_client.py | 28 +++++++++++++++++++ 6 files changed, 96 insertions(+), 12 deletions(-) diff --git a/docs/read-only-ingest-client-plan.md b/docs/read-only-ingest-client-plan.md index 5e6837a..6befe8f 100644 --- a/docs/read-only-ingest-client-plan.md +++ b/docs/read-only-ingest-client-plan.md @@ -1,4 +1,8 @@ -# Read-only ingest — the client half (DI-11.9) +# Read-only ingest — the client half (DI-11.9) — **DONE** + +> Landed: `sub_path` (DI-11.12), the retry guard and `placement` (DI-11.7), +> `CommitResult.storage_path` and the docs below (DI-11.9). mypy clean, 52 tests +> pass. What is still open is recorded at the end of this file. Companion note for **this** repository. The full design lives with the server work it depends on: @@ -23,9 +27,9 @@ DI-11 makes `read_only` mean *store the files permanently* under **Where the bytes land is chosen by the server**, in `begin`'s presigned targets, so this package needs no new parameter and no new call. -## What changes here +## What changed here -0. **`sub_path` — the one new parameter** (DI-11.12). `FolderIngest(..., +0. ✅ **`sub_path` — the one new parameter** (DI-11.12). `FolderIngest(..., sub_path="2026")`, passed straight through to `begin`; the server validates, normalises and scopes on it. It is what lets a read-only table accumulate: @@ -41,20 +45,20 @@ so this package needs no new parameter and no new call. the original `rel_path`. `sub_path` is for filing a **flat** folder under a name they choose. Worth a worked example in `README.md`, and note that `replace` with a `sub_path` clears only that year. -1. **Docs and docstrings** — `folder.py` class docstring + the `intent` +1. ✅ **Docs and docstrings** — `folder.py` class docstring + the `intent` parameter, `dds_ingestion/README.md:38`, `__init__.py:16`: say what each intent now does (permanent raw folder + catalog view vs. Iceberg copy). -2. **`CommitResult.storage_path`** (`models.py`) — optional, additive; the +2. ✅ **`CommitResult.storage_path`** (`models.py`) — optional, additive; the physical Dremio path the server reports for a permanent ingest, so a notebook can print where the files actually are. Parse permissively as everywhere else (absent ⇒ `None`), so it works against an older server. -3. **Resume/retry note** (`folder.py:305-322`, `retry()`): `_already_done`'s +3. ✅ **Resume/retry note** (`folder.py:305-322`, `retry()`): `_already_done`'s docstring says a re-run is "an S3 overwrite of the identical key, which is harmless". Against a permanent folder in `append` mode that is only true if the server reuses the keys instead of numbering them. Align the wording with whatever DI-11.7 settles, and steer users to `attach()` + `retry()` rather than re-running `run()`. -4. **Tests** — `tests/dds_ingestion/`: `storage_path` parsing, and a +4. ✅ **Tests** — `tests/dds_ingestion/`: `storage_path` parsing, and a respx-mocked `begin` whose `key_prefix` is a permanent `read/...` prefix, asserting the client uploads to the issued targets unchanged. diff --git a/src/eea_datalakehouse/dds_ingestion/README.md b/src/eea_datalakehouse/dds_ingestion/README.md index 6603410..cc8417e 100644 --- a/src/eea_datalakehouse/dds_ingestion/README.md +++ b/src/eea_datalakehouse/dds_ingestion/README.md @@ -35,14 +35,27 @@ outcome = FolderIngest( folder="./my_data", target_catalog_path="biodiversity.uploads", data_format="parquet", # one of parquet | csv | json - intent="read_only", # or "editable" + intent="read_only", # or "editable" — see below conflict_mode="fail", parallelism=4, # concurrent uploads (default 4) ).run() print(outcome.commit.table_path, outcome.commit.record_count) +print(outcome.commit.storage_path) # where the files are, when they are kept ``` +`intent` decides what the transfer leaves behind: + +| | `read_only` | `editable` | +|---|---|---| +| your uploaded files | kept — they **are** the table | copied in, then deleted | +| the table | the folder, registered, with a view at the catalog path | an Iceberg table | +| good for | published data, data that accumulates | a table you will write to | + +(Storing read-only files permanently is a server setting. Where it is not +enabled, both intents stage and load as before and only the table's shape +differs — `storage_path` is then `None`.) + `run()` performs `begin → upload(all files) → commit`, and leaves the session handle available afterwards. diff --git a/src/eea_datalakehouse/dds_ingestion/__init__.py b/src/eea_datalakehouse/dds_ingestion/__init__.py index e88210b..9caa5f1 100644 --- a/src/eea_datalakehouse/dds_ingestion/__init__.py +++ b/src/eea_datalakehouse/dds_ingestion/__init__.py @@ -13,10 +13,16 @@ folder="./my_data", target_catalog_path="biodiversity.uploads", data_format="parquet", - intent="read_only", + intent="read_only", # keep the files as the table | "editable" → Iceberg parallelism=4, ).run() print(outcome.commit.table_path, outcome.commit.record_count) + print(outcome.commit.storage_path) # where the files are, if kept + +``intent`` decides what the transfer leaves behind. On a server configured for +permanent read-only storage, ``"read_only"`` keeps the uploaded files where the +table lives and registers them; ``"editable"`` loads them into an Iceberg table +and deletes the upload. See :class:`FolderIngest` for the full rule. """ from __future__ import annotations diff --git a/src/eea_datalakehouse/dds_ingestion/folder.py b/src/eea_datalakehouse/dds_ingestion/folder.py index 0afee08..fb852b7 100644 --- a/src/eea_datalakehouse/dds_ingestion/folder.py +++ b/src/eea_datalakehouse/dds_ingestion/folder.py @@ -94,6 +94,23 @@ def scan_folder(folder: Path, data_format: DataFormat) -> list[FileSpec]: class FolderIngest: """Orchestrate ingest of a local folder into a Dremio table via DDS. + ``intent`` decides what the transfer leaves behind, and on a server + configured for permanent read-only storage that includes **where the files + end up**: + + * ``"read_only"`` — the upload is stored where the table lives and kept. The + folder is registered as the dataset (which is also what builds Dremio's + metadata: schema, file listing, Parquet statistics) and a view at the + catalog path points at it. Nothing is copied, and the files keep the shape + they were exported in. Use it for data that is published, not edited. + * ``"editable"`` — the upload is staged, loaded into an Iceberg table in the + catalog, and then deleted. Use it for a table that will be written to. + + Against a server that has not enabled permanent storage, both intents stage + and load as before; the difference is then only the table's shape. Either way + the destination is the server's to choose — this class uploads to the targets + ``begin`` issues. + ``sub_path`` files this upload under a named sub-folder of the table — the accumulating-dataset shape, one year at a time:: @@ -359,8 +376,15 @@ def _already_done(self, session_id: str) -> set[str]: explicitly names under ``raw["uploaded"]``, so resume never wrongly drops a file. NOTE: the server populates that list for **server-run** transfers (``POST /ingest/folder``); a client-uploaded session reports nothing - there, so re-running one uploads every file again (an S3 overwrite of the - identical key, which is harmless). + there, so re-running one uploads every file again. + + Re-running the SAME session is harmless — the keys are identical, so the + second upload overwrites the first. Starting a NEW session is a different + matter where the files are stored permanently: the server numbers an + incoming name that already exists, precisely so an append cannot + overwrite live data, and that turns a re-run into a second copy rather + than an overwrite. :meth:`retry` refuses exactly that case; prefer + :meth:`FolderIngest.attach` + :meth:`retry` over re-running :meth:`run`. """ try: diff --git a/src/eea_datalakehouse/dds_ingestion/models.py b/src/eea_datalakehouse/dds_ingestion/models.py index ef14ba0..04ea0ab 100644 --- a/src/eea_datalakehouse/dds_ingestion/models.py +++ b/src/eea_datalakehouse/dds_ingestion/models.py @@ -113,12 +113,20 @@ def from_json(cls, data: dict[str, Any]) -> BeginResult: @dataclass(frozen=True, slots=True) class CommitResult: - """Response from ``POST /api/v1/ingest/commit``.""" + """Response from ``POST /api/v1/ingest/commit``. + + ``table_path`` is where the table is **queried** — the catalog path. For a + read-only ingest whose files are stored permanently, ``storage_path`` is + where those files physically **are** (the Dremio path of the promoted + folder); it is ``None`` for a staged ingest, whose upload was copied into the + catalog's own storage and deleted, and against any server predating it. + """ session_id: str status: str table_path: str | None = None record_count: int | None = None + storage_path: str | None = None @classmethod def from_json(cls, data: dict[str, Any]) -> CommitResult: @@ -127,6 +135,7 @@ def from_json(cls, data: dict[str, Any]) -> CommitResult: status=data["status"], table_path=data.get("table_path"), record_count=data.get("record_count"), + storage_path=data.get("storage_path"), ) diff --git a/tests/dds_ingestion/test_client.py b/tests/dds_ingestion/test_client.py index 3201e73..4cdb5ee 100644 --- a/tests/dds_ingestion/test_client.py +++ b/tests/dds_ingestion/test_client.py @@ -210,3 +210,31 @@ def test_begin_sends_sub_path_only_when_given(creds: DremioCreds) -> None: client.begin(sub_path="2026", **kwargs) # type: ignore[arg-type] assert json.loads(route.calls.last.request.read())["sub_path"] == "2026" + + +def test_commit_result_carries_the_physical_location() -> None: + """DI-11.9: ``storage_path`` says where permanently-stored files actually are. + + ``table_path`` is where the table is queried; the two differ for a read-only + ingest that keeps its files. Absent (a staged ingest, or a server predating + the field) parses as ``None`` rather than failing — the models are + deliberately permissive about keys they do not know. + """ + from eea_datalakehouse.dds_ingestion.models import CommitResult + + kept = CommitResult.from_json( + { + "session_id": "s1", + "status": "done", + "table_path": "catalog/water/bwd/assessments", + "record_count": 12, + "storage_path": "local_s3/dh-prod-data/read/water/bwd/assessments", + } + ) + assert kept.table_path == "catalog/water/bwd/assessments" + assert kept.storage_path == "local_s3/dh-prod-data/read/water/bwd/assessments" + + staged = CommitResult.from_json( + {"session_id": "s2", "status": "done", "table_path": "x/y", "record_count": 1} + ) + assert staged.storage_path is None From e429482f2d22cad92673dbb44c299323765e722b Mon Sep 17 00:00:00 2001 From: blikij Date: Tue, 25 Aug 2026 11:30:31 +0200 Subject: [PATCH 16/16] feat(dds_ingestion): name the call that failed, and separate S3 faults MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ported from the DDS repo (EEALakeHouse@904c1a4), where a folder ingest failed with "DDS ingest API error 500: Internal Server Error" and nothing said more. A transfer makes several calls to two different systems, so a status code with no call name does not identify the broken one. * IngestApiError carries ``where`` — the method and path of the call that failed — and puts it in the message; status_code/message stay available. * S3UploadError (a subclass, so existing handling still catches it) is raised when a pre-signed upload fails: those bytes go straight to object storage, and the message names the file and the URL that actually answered rather than blaming the Document Service. * Error bodies are capped at 500 chars: a proxy's HTML error page ran to kilobytes and buried the status line it came with. An empty body now reads as "(empty 500 response body)" instead of nothing at all. Ported, not copied: this package keeps its own divergences (the PAT-redacting __repr__ and begin(sub_path=...)), which the DDS copy does not have. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Qh3Bnb8mdNjtFbmZxfmGEZ --- .../dds_ingestion/__init__.py | 3 +- src/eea_datalakehouse/dds_ingestion/client.py | 94 ++++++++++++++++--- tests/dds_ingestion/test_client.py | 67 ++++++++++++- 3 files changed, 147 insertions(+), 17 deletions(-) diff --git a/src/eea_datalakehouse/dds_ingestion/__init__.py b/src/eea_datalakehouse/dds_ingestion/__init__.py index 9caa5f1..54d714c 100644 --- a/src/eea_datalakehouse/dds_ingestion/__init__.py +++ b/src/eea_datalakehouse/dds_ingestion/__init__.py @@ -27,7 +27,7 @@ from __future__ import annotations -from .client import IngestApiError, IngestClient +from .client import IngestApiError, IngestClient, S3UploadError from .credentials import ( DremioCreds, MissingCredentialsError, @@ -74,6 +74,7 @@ "MissingCredentialsError", "Progress", "S3Plan", + "S3UploadError", "StageResult", "StatusResult", "UploadPart", diff --git a/src/eea_datalakehouse/dds_ingestion/client.py b/src/eea_datalakehouse/dds_ingestion/client.py index a6a200e..cc6b8b7 100644 --- a/src/eea_datalakehouse/dds_ingestion/client.py +++ b/src/eea_datalakehouse/dds_ingestion/client.py @@ -48,15 +48,53 @@ ) DEFAULT_TIMEOUT = 60.0 +# Error bodies are quoted back to the caller; cap them so an HTML error page +# does not bury the status line it came with. +_MAX_ERROR_CHARS = 500 class IngestApiError(RuntimeError): - """A DDS ingest API call returned a non-success status.""" + """A DDS ingest API call returned a non-success status. - def __init__(self, status_code: int, message: str) -> None: - super().__init__(f"DDS ingest API error {status_code}: {message}") + ``where`` names the call that failed (e.g. ``POST /api/v1/ingest/begin``). + It is part of the message because a transfer makes several different calls + to two different systems, and "API error 500" alone does not say which one + broke — see :class:`S3UploadError` for the other system. + """ + + def __init__(self, status_code: int, message: str, *, where: str | None = None) -> None: + super().__init__( + f"DDS ingest API error {status_code}" + f"{f' on {where}' if where else ''}: {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. + + The bytes go straight from here to S3 using the target ``begin`` issued, so + this failure is the object storage's (or whatever proxy sits in front of + it): a rejected policy, an unreachable endpoint, a gateway error. Reported + as its own class so a caller can tell "your upload never landed" from "the + Document Service refused the request", and so the message names the file and + the URL that actually answered. + """ + + def __init__( + self, status_code: int, message: str, *, rel_path: str, url: str + ) -> None: + RuntimeError.__init__( + self, + f"S3 upload of {rel_path!r} failed: HTTP {status_code} from {url}: {message}", + ) self.status_code = status_code self.message = message + self.where = url + self.rel_path = rel_path + self.url = url class IngestClient: @@ -157,7 +195,7 @@ def get_status(self, session_id: str) -> StatusResult: resp = self._http.get( f"{self._base_url}/api/v1/ingest/{session_id}", headers=self._auth_headers ) - self._raise_for_status(resp) + self._raise_for_status(resp, f"GET /api/v1/ingest/{session_id}") return StatusResult.from_json(resp.json()) def list_sessions(self, state: str | None = None) -> list[StatusResult]: @@ -170,7 +208,7 @@ def list_sessions(self, state: str | None = None) -> list[StatusResult]: resp = self._http.get( f"{self._base_url}/api/v1/ingest", params=params, headers=self._auth_headers ) - self._raise_for_status(resp) + self._raise_for_status(resp, "GET /api/v1/ingest") payload = resp.json() rows = payload if isinstance(payload, list) else payload.get("sessions", []) return [StatusResult.from_json(row) for row in rows] @@ -188,7 +226,7 @@ def retry(self, session_id: str) -> StatusResult: f"{self._base_url}/api/v1/ingest/{session_id}/retry", headers=self._auth_headers, ) - self._raise_for_status(resp) + self._raise_for_status(resp, f"POST /api/v1/ingest/{session_id}/retry") return StatusResult.from_json(resp.json()) def cancel(self, session_id: str) -> None: @@ -201,7 +239,7 @@ def cancel(self, session_id: str) -> None: resp = self._http.delete( f"{self._base_url}/api/v1/ingest/{session_id}", headers=self._auth_headers ) - self._raise_for_status(resp) + self._raise_for_status(resp, f"DELETE /api/v1/ingest/{session_id}") def estimate(self, session_id: str) -> EstimateResult: """Row-count + size class for the staged data (``POST .../estimate``). @@ -227,7 +265,7 @@ def stage(self, session_id: str, rel_path: str, data: bytes) -> StageResult: files={"file": (rel_path.rsplit("/", 1)[-1], data)}, headers=self._auth_headers, ) - self._raise_for_status(resp) + self._raise_for_status(resp, "POST /api/v1/ingest/stage") return StageResult.from_json(resp.json()) def upload_file(self, target: UploadTarget, data: bytes) -> str | None: @@ -263,7 +301,7 @@ def upload_file(self, target: UploadTarget, data: bytes) -> str | None: resp = self._http.request( target.method, target.url, content=data, headers=target.headers ) - self._raise_for_status(resp) + self._raise_for_upload(resp, rel_path=target.rel_path, url=target.url) return resp.headers.get("ETag") def upload_file_multipart( @@ -301,7 +339,7 @@ def upload_file_multipart( start = index * chunk body = data[start : start + chunk] if chunk else b"" resp = self._http.put(part.url, content=body) - self._raise_for_status(resp) + self._raise_for_upload(resp, rel_path=target.rel_path, url=part.url) etag = resp.headers.get("ETag") if etag is None: raise IngestApiError( @@ -318,14 +356,19 @@ def _post(self, path: str, body: dict[str, Any]) -> dict[str, Any]: resp = self._http.post( f"{self._base_url}{path}", json=body, headers=self._auth_headers ) - self._raise_for_status(resp) + self._raise_for_status(resp, f"POST {path}") result: dict[str, Any] = resp.json() return result @staticmethod - def _raise_for_status(resp: httpx.Response) -> None: - if resp.is_success: - return + def _error_message(resp: httpx.Response) -> str: + """The most useful text an error response carries. + + A DDS error is ``{error, message, path}`` JSON, so quote its ``message``. + Anything else — an S3 XML fault, a proxy's HTML error page, Starlette's + bare ``Internal Server Error`` — is quoted verbatim but capped: those + pages run to kilobytes and only the first lines identify the sender. + """ message = resp.text try: payload = resp.json() @@ -333,4 +376,25 @@ def _raise_for_status(resp: httpx.Response) -> None: message = str(payload.get("message") or payload.get("error") or message) except ValueError: pass - raise IngestApiError(resp.status_code, message) + message = message.strip() + if len(message) > _MAX_ERROR_CHARS: + message = message[:_MAX_ERROR_CHARS] + "… (truncated)" + return message or f"(empty {resp.status_code} response body)" + + @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) + + @classmethod + def _raise_for_upload( + cls, resp: httpx.Response, *, rel_path: str, url: str + ) -> None: + """Raise :class:`S3UploadError` unless the pre-signed upload succeeded.""" + if resp.is_success: + return + raise S3UploadError( + resp.status_code, cls._error_message(resp), rel_path=rel_path, url=url + ) diff --git a/tests/dds_ingestion/test_client.py b/tests/dds_ingestion/test_client.py index 4cdb5ee..723aaaa 100644 --- a/tests/dds_ingestion/test_client.py +++ b/tests/dds_ingestion/test_client.py @@ -3,7 +3,11 @@ import httpx import pytest import respx -from eea_datalakehouse.dds_ingestion.client import IngestApiError, IngestClient +from eea_datalakehouse.dds_ingestion.client import ( + IngestApiError, + IngestClient, + S3UploadError, +) from eea_datalakehouse.dds_ingestion.credentials import DremioCreds from eea_datalakehouse.dds_ingestion.models import FileSpec, UploadPart, UploadTarget @@ -177,6 +181,67 @@ def test_error_response_raises_with_message(creds: DremioCreds) -> None: assert "exists" in str(exc.value) +@respx.mock +def test_dds_error_names_the_call_that_failed(creds: DremioCreds) -> None: + """A 500 with Starlette's bare body must still say WHICH call broke. + + The message an ingest reported was "DDS ingest API error 500: Internal + Server Error" — true, and useless: a transfer calls begin, upload and commit + against two different systems. + """ + respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock( + return_value=httpx.Response(500, text="Internal Server Error") + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError) as exc: + client.begin( + target_catalog_path="x", + intent="read_only", + data_format="parquet", + conflict_mode="fail", + files=[FileSpec("a.parquet", 6)], + ) + + assert exc.value.where == "POST /api/v1/ingest/begin" + assert "POST /api/v1/ingest/begin" in str(exc.value) + assert "Internal Server Error" in str(exc.value) + + +@respx.mock +def test_failed_presigned_upload_is_not_reported_as_a_dds_error( + creds: DremioCreds, +) -> None: + """The upload goes straight to object storage; its failures are its own.""" + respx.post("https://s3.example/bucket").mock( + return_value=httpx.Response(500, text="gateway said no") + ) + target = UploadTarget( + rel_path="a.parquet", url="https://s3.example/bucket", method="POST" + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(S3UploadError) as exc: + client.upload_file(target, b"PAR1") + + assert exc.value.rel_path == "a.parquet" + assert exc.value.url == "https://s3.example/bucket" + message = str(exc.value) + assert message.startswith("S3 upload of 'a.parquet' failed: HTTP 500") + assert "https://s3.example/bucket" in message + assert "DDS ingest API error" not in message + # …and it is still an IngestApiError, so existing handling keeps working. + assert isinstance(exc.value, IngestApiError) + + +@respx.mock +def test_long_error_bodies_are_truncated(creds: DremioCreds) -> None: + respx.post(f"{BASE_URL}/api/v1/ingest/commit").mock( + return_value=httpx.Response(502, text="x" * 5000) + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError) as exc: + client.commit(session_id="sess-1") + + assert exc.value.message.endswith("… (truncated)") + assert len(exc.value.message) < 600 + + @respx.mock def test_begin_sends_sub_path_only_when_given(creds: DremioCreds) -> None: """DI-11.12: the named sub-folder rides on ``begin`` and nowhere else.