Skip to content
21 changes: 14 additions & 7 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -305,11 +305,15 @@ endpoints (`dds_documents/client.py`) are still assumptions to confirm with the

## Notebook magics for SDI and DDS (`%sdi` / `%dds`)

`%sdi` (`SdiSession`) reads SDI and `%dds` (`DdsSession`) pushes files to DDS. Both are built
on first use from the kernel environment like `%catalog`'s and `%ingest`'s sessions. `%dds
push` puts files in a folder of the dataset `%catalog use(...)` last selected; `path=` picks
another one, absolute or relative to that context, as any `%catalog` path. The folder names
can be typed bare on a `%dds` line (worked example: `debugger/sdi_session_example.ipynb`):
`%sdi` (`SdiSession`) reads SDI and `%dds` (`DdsSession`) pushes files to DDS, lists them and
gets them back. Both are built on first use from the kernel environment like `%catalog`'s and
`%ingest`'s sessions. `%dds push` puts files in a folder of the dataset `%catalog use(...)` last
selected, `%dds list` shows the file names already in one, and `%dds get` downloads one of them
into the notebook's folder (`output_dir=` for another; a missing file is an error). `path=`
picks another dataset, absolute or relative to that context, as any `%catalog` path. The folder
is compulsory and is one of DDS's eight document folders, typed bare on a `%dds` line; a
misspelt, quoted or missing folder is an error that lists the valid ones (worked example:
`debugger/sdi_session_example.ipynb`):

```python
import eea_datalakehouse.notebook # registers %catalog/%ingest/%sdi/%dds
Expand All @@ -318,6 +322,9 @@ import eea_datalakehouse.notebook # registers %catalog/%ingest/%sdi/%dds
%dds push(DELIVERABLES, "report.pdf", "annex/*.xlsx")
record = %sdi get_xml("070d9baa-448d-4168-8514-7dadb3ad876d")
%dds push(METADATA, record, overwrite=False)
%dds list(DELIVERABLES) # report.pdf and the annexes, sorted
%dds list(DELIVERABLES, "*.pdf") # filter: a wildcard, ".xlsx", "report", or a list of them
f = %dds get(DELIVERABLES, "report.pdf") # -> ./report.pdf; f is the local file
%dds target(ASSESSMENTS) # where a push would go
%dds help
```
Expand Down Expand Up @@ -510,7 +517,7 @@ context itself — raises `CatalogSessionError` if none is set yet:
| `dds_ingestion/progress.py` | tqdm progress bar with graceful fallback |
| `dds_ingestion/folder.py` | `FolderIngest` orchestration (scan/parallel/resume) |
| `dds_documents/client.py` | `DocumentsClient` — put/get/list plain files in DDS folders (no ingest) |
| `dds_documents/controller.py` | `DdsController` / `DdsFolder` — push files to a catalog node's DDS document folders (metadata, deliverables, ...) |
| `dds_documents/controller.py` | `DdsController` / `DdsFolder` — push, list and get files in a catalog node's DDS document folders (metadata, deliverables, ...) |
| `dds_documents/paths.py` | catalog (`a.b.c`) ⇄ storage (`a/b/c`) path conversion |
| `dds_documents/session.py` | `DdsSession` — what `%dds` dispatches onto |
| `geopackage/controller.py` | `GpkgController` — Dremio tables/views to a GeoPackage, streamed over parallel Flight channels |
Expand Down Expand Up @@ -602,7 +609,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.34"
%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.35"

# staging's latest release (early access)
%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.35-staging"
Expand Down
100 changes: 90 additions & 10 deletions debugger/test_catalog_magic.ipynb

Large diffs are not rendered by default.

11 changes: 10 additions & 1 deletion src/eea_datalakehouse/dds_documents/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@

with DdsController.from_env() as dds:
dds.push("catalog.water.bwd", DdsFolder.DELIVERABLES, ["report.pdf"])
dds.list("catalog.water.bwd", DdsFolder.DELIVERABLES)
report = dds.get("catalog.water.bwd", DdsFolder.DELIVERABLES, "report.pdf")

:class:`DocumentsClient` is the HTTP layer underneath, one method per DDS
endpoint; :class:`DdsSession` is what the `%dds` magic dispatches onto.
Expand All @@ -18,20 +20,26 @@
from .controller import (
DdsController,
DdsFile,
DdsFileNotFoundError,
DdsFilesExistError,
DdsFolder,
DdsListing,
DdsPushError,
PushReport,
folder_name,
valid_folders_hint,
)
from .paths import path_segments, to_catalog_path, to_storage_path
from .session import DdsSession, DdsSessionError
from .session import DdsDownload, DdsSession, DdsSessionError

__all__ = [
"DdsController",
"DdsDownload",
"DdsFile",
"DdsFileNotFoundError",
"DdsFilesExistError",
"DdsFolder",
"DdsListing",
"DdsPushError",
"DdsSession",
"DdsSessionError",
Expand All @@ -43,4 +51,5 @@
"path_segments",
"to_catalog_path",
"to_storage_path",
"valid_folders_hint",
]
115 changes: 93 additions & 22 deletions src/eea_datalakehouse/dds_documents/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,17 +10,19 @@
================================ ============================================
:meth:`DocumentsClient.put` ``PUT /api/v1/files/{path}``
:meth:`~DocumentsClient.get` ``GET /api/v1/files/{path}``
:meth:`~DocumentsClient.exists` ``GET /api/v1/files/{path}``
:meth:`~DocumentsClient.list` ``GET /api/v1/files?prefix={folder}``
:meth:`~DocumentsClient.exists` ``HEAD /api/v1/files/{path}``
:meth:`~DocumentsClient.list` ``GET /api/v1/files/{folder}/`` (paged by ``cursor``)
:meth:`~DocumentsClient.folders` ``GET /api/v1/folders``
================================ ============================================

``PUT`` and ``GET /api/v1/folders`` follow the Dremio Document Service's own
routes: the upload body is the raw file, and overwriting is asked for with the
``X-DDS-Overwrite: true`` header (an existing file otherwise answers 409). The
other shapes are still **interface assumptions** to confirm with the DDS team
(see ``docs/sdi-integration-plan.md``). Each one is a module constant so a
confirmed contract is a one-line change.
These are the Dremio Document Service's own routes (its ``/openapi.json``):
the upload body is the raw file, and overwriting is asked for with the
``X-DDS-Overwrite: true`` header (an existing file otherwise answers 409).
``GET /api/v1/files/{path}`` downloads a file, or — when the path ends in
``/`` — lists that folder, a page (``limit``, default 200) at a time. The
service does not publish the listing's JSON shape, so :meth:`list` reads the
usual ones (see :func:`_listing_entries`) and says which keys it found when a
response is none of them.
"""

from __future__ import annotations
Expand All @@ -34,11 +36,20 @@
from ..dds_ingestion.credentials import DremioCreds, load_base_url, load_creds

DEFAULT_TIMEOUT = 60.0
# Assumed DDS document contract — to confirm with the DDS team.
FILES_ENDPOINT = "/api/v1/files/{path}"
LIST_ENDPOINT = "/api/v1/files"
# A trailing slash asks DDS for the folder's listing instead of a download;
# without it, /api/v1/files redirects (307) to /api/v1/files/.
LIST_ENDPOINT = "/api/v1/files/{path}/"
FOLDERS_ENDPOINT = "/api/v1/folders"
OVERWRITE_HEADER = "X-DDS-Overwrite"
LIST_PAGE_SIZE = 200
# A listing longer than this many pages is a server that keeps handing back a cursor.
_MAX_LIST_PAGES = 10_000
# Where a listing page keeps its entries, its next-page cursor, and an entry its path.
_ENTRY_KEYS = ("items", "entries", "files", "children", "objects", "contents", "results", "data")
_CURSOR_KEYS = ("next_cursor", "nextCursor", "cursor", "next")
_PATH_KEYS = ("path", "key", "rel_path", "name", "filename")
_FOLDER_TYPES = {"dir", "directory", "folder", "prefix", "collection"}
# 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
Expand Down Expand Up @@ -146,21 +157,44 @@ def get(self, path: str) -> bytes | None:
return resp.content

def exists(self, path: str) -> bool:
"""Whether a document is stored at ``path``."""
return self.get(path) is not None
"""Whether a document is stored at ``path`` (a ``HEAD``: nothing is downloaded)."""
endpoint = self._file_endpoint(path)
resp = self._http.head(f"{self._base_url}{endpoint}", headers=self._auth_headers)
if resp.status_code == 404:
return False
_raise_for_status(resp, f"HEAD {endpoint}")
return True

def list(self, folder: str) -> list[str]:
"""Paths of the documents under ``folder``."""
prefix = _clean_path(folder)
resp = self._http.get(
f"{self._base_url}{LIST_ENDPOINT}",
params={"prefix": prefix},
headers=self._auth_headers,
"""The files in ``folder``, as DDS names them (a path or a bare name),
every page of them; subfolders are left out. Empty if DDS has no such
folder (404)."""
endpoint = LIST_ENDPOINT.format(path=quote(_clean_path(folder), safe="/"))
params: dict[str, str | int] = {"limit": LIST_PAGE_SIZE}
found: list[str] = []
for _ in range(_MAX_LIST_PAGES):
resp = self._http.get(
f"{self._base_url}{endpoint}", params=params, headers=self._auth_headers
)
if resp.status_code == 404:
return found
_raise_for_status(resp, f"GET {endpoint}")
try:
payload: Any = resp.json()
except ValueError as exc:
raise DocumentsApiError(
resp.status_code,
f"expected a JSON folder listing, got {resp.headers.get('content-type')!r}",
where=f"GET {endpoint}",
) from exc
entries, cursor = _listing_entries(payload, where=f"GET {endpoint}")
found.extend(_entry_path(entry) for entry in entries if not _is_folder_entry(entry))
if not cursor or cursor == params.get("cursor"):
return found
params["cursor"] = cursor
raise DocumentsApiError(
200, f"listing did not end after {_MAX_LIST_PAGES} pages", where=f"GET {endpoint}"
)
_raise_for_status(resp, f"GET {LIST_ENDPOINT}?prefix={prefix}")
payload: Any = resp.json()
rows = payload.get("files", []) if isinstance(payload, dict) else payload
return [row["path"] if isinstance(row, dict) else str(row) for row in rows]

def folders(self) -> list[str]:
"""The document folder names DDS shows under every catalog node.
Expand Down Expand Up @@ -190,6 +224,43 @@ def _clean_path(path: str) -> str:
return cleaned


def _listing_entries(payload: Any, *, where: str) -> tuple[list[Any], str | None]:
"""``(entries, next cursor)`` of one listing page: a bare list, or an object
keeping its entries under one of ``_ENTRY_KEYS`` and the next page's cursor
under one of ``_CURSOR_KEYS`` (or none on the last page)."""
if isinstance(payload, list):
return payload, None
if isinstance(payload, dict):
key = next((k for k in _ENTRY_KEYS if isinstance(payload.get(k), list)), None)
if key is not None:
cursor = next((payload[k] for k in _CURSOR_KEYS if payload.get(k)), None)
return payload[key], str(cursor) if cursor is not None else None
keys = ", ".join(map(str, payload)) if isinstance(payload, dict) else type(payload).__name__
raise DocumentsApiError(
200, f"unrecognised folder listing (top-level keys: {keys or 'none'})", where=where
)


def _entry_path(entry: Any) -> str:
if isinstance(entry, dict):
for key in _PATH_KEYS:
if isinstance(entry.get(key), str) and entry[key]:
return str(entry[key])
raise DocumentsApiError(
200, f"listing entry without a path or name (keys: {', '.join(map(str, entry))})"
)
return str(entry)


def _is_folder_entry(entry: Any) -> bool:
if isinstance(entry, dict):
kind = str(entry.get("type") or entry.get("kind") or "").lower()
if kind in _FOLDER_TYPES or entry.get("is_dir") is True or entry.get("isDir") is True:
return True
return any(isinstance(entry.get(k), str) and entry[k].endswith("/") for k in _PATH_KEYS)
return str(entry).endswith("/")


def _error_message(resp: httpx.Response) -> str:
"""The ``message`` of a DDS ``{error, message}`` body, else the capped text."""
message = resp.text
Expand Down
Loading
Loading