Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 16 additions & 9 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.34-staging"
pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.35-staging"
```

## Usage
Expand Down 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,10 +609,10 @@ 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.34-staging"
%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.35-staging"
```

Use the `%pip` magic rather than `!pip` — it installs into the kernel the
Expand Down
100 changes: 90 additions & 10 deletions debugger/test_catalog_magic.ipynb

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "EEADataLakehouse"
version = "0.1.34"
version = "0.1.35"
description = "Data preparation and validation utilities for the EEA data lakehouse pipeline."
readme = "README.md"
requires-python = ">=3.11"
Expand Down
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