From f46ceea4cb6df1ccdaece88fe6a41bc7bdf4d0da Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 27 Aug 2026 11:54:47 +0000 Subject: [PATCH 1/2] Update README install examples to v0.1.12 [skip ci] --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 70e18ea..91df0f4 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.11" +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.12" # staging's latest release (early access) %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.12-staging" From a4c9b3c47bdb4f2c95a5f0096b5e7bcf16c7c0b7 Mon Sep 17 00:00:00 2001 From: oskaresparza Date: Mon, 7 Sep 2026 13:09:35 +0200 Subject: [PATCH 2/2] Add a notebook-facing facade: CatalogSession/IngestSession + %catalog/%ingest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A queue-then-commit layer over the developer-oriented catalog/ingestion APIs, aimed at data custodians in JupyterLab rather than application developers — see docs/notebook-facade-for-data-scientists.md for the full design discussion this implements. - CatalogSession: queue copy/move/tag/untag/set_wiki/delete_wiki/set_meta/ create_folder/delete_folder, then commit(retry=True) all-or-nothing — every reversible verb rolls back on failure via a real compensating action; delete_folder and overwrite=True copies/moves are explicitly flagged as unresolvable rather than silently claimed clean. - IngestSession: queue FolderIngest runs, commit(retry=True) in order — no rollback, by design (no delete-with-backing-data op exists yet; see docs/read-only-ingest-client-plan.md), so a partial failure reports what already landed instead of pretending it can be undone. - idempotency_key is fully encapsulated in both — never a parameter a custodian sees or passes. - CatalogSession context: a leading '.' on a path resolves against a "current" prefix that ordinary use already keeps up to date on its own (no set_context() call needed for normal sequential use). - %catalog/%ingest (eea_datalakehouse.notebook.magics): thin dispatch onto one session per kernel, friendly error printing instead of tracebacks. Registered automatically by `import eea_datalakehouse.notebook` (no %load_ext needed, though it still works and is safe to combine). - A Jupyter Comm channel (eea_datalakehouse.catalog_context) lets an integrated frontend push a selected catalog-tree leaf's path into the context invisibly — the receiving half of a future JupyterLab extension integration; the emitting half lives in the separate eeadata/EEALakeHouse repo and isn't built here. Adds a `notebook` extra (ipython) and two worked-example notebooks under docs/notebooks/. 285 tests passing. Co-Authored-By: Claude Sonnet 5 --- docs/notebook-facade-for-data-scientists.md | 331 +++++++++++ docs/notebooks/catalog_session_example.ipynb | 156 ++++++ docs/notebooks/ingest_session_example.ipynb | 144 +++++ pyproject.toml | 4 + src/eea_datalakehouse/catalog/__init__.py | 7 +- src/eea_datalakehouse/catalog/session.py | 524 ++++++++++++++++++ .../dds_ingestion/__init__.py | 5 + .../dds_ingestion/session.py | 206 +++++++ src/eea_datalakehouse/notebook/__init__.py | 36 ++ src/eea_datalakehouse/notebook/magics.py | 196 +++++++ tests/catalog/test_session.py | 141 +++++ tests/catalog/test_session_context.py | 109 ++++ tests/dds_ingestion/test_ingest_session.py | 168 ++++++ tests/notebook/__init__.py | 0 tests/notebook/conftest.py | 19 + tests/notebook/test_autoregister.py | 64 +++ tests/notebook/test_context_comm.py | 161 ++++++ tests/notebook/test_magics.py | 136 +++++ 18 files changed, 2406 insertions(+), 1 deletion(-) create mode 100644 docs/notebook-facade-for-data-scientists.md create mode 100644 docs/notebooks/catalog_session_example.ipynb create mode 100644 docs/notebooks/ingest_session_example.ipynb create mode 100644 src/eea_datalakehouse/catalog/session.py create mode 100644 src/eea_datalakehouse/dds_ingestion/session.py create mode 100644 src/eea_datalakehouse/notebook/__init__.py create mode 100644 src/eea_datalakehouse/notebook/magics.py create mode 100644 tests/catalog/test_session.py create mode 100644 tests/catalog/test_session_context.py create mode 100644 tests/dds_ingestion/test_ingest_session.py create mode 100644 tests/notebook/__init__.py create mode 100644 tests/notebook/conftest.py create mode 100644 tests/notebook/test_autoregister.py create mode 100644 tests/notebook/test_context_comm.py create mode 100644 tests/notebook/test_magics.py diff --git a/docs/notebook-facade-for-data-scientists.md b/docs/notebook-facade-for-data-scientists.md new file mode 100644 index 0000000..5748fae --- /dev/null +++ b/docs/notebook-facade-for-data-scientists.md @@ -0,0 +1,331 @@ +# Making the catalog API approachable for data scientists in JupyterLab + +Idea note, not a plan — captures a discussion, nothing here is agreed or scheduled. + +**Prototype exists.** `CatalogSession`/`IngestSession` +(`src/eea_datalakehouse/catalog/session.py`, +`src/eea_datalakehouse/dds_ingestion/session.py`) implement "Queue calls, then +commit" and "Two sessions, not one" below; `%catalog`/`%ingest` +(`src/eea_datalakehouse/notebook/magics.py`) implement "Two magics, two +sessions" and "Loading the magics without typing a magic to do it" over them; +relative-path context (a leading `.`, auto-inferred from usage, plus a Comm +channel for an integration to push it invisibly) implements "Pre-filling +catalog context" — see `docs/notebooks/catalog_session_example.ipynb` and +`docs/notebooks/ingest_session_example.ipynb` for worked examples. Still just +a prototype behind an opt-in `notebook` extra, not reviewed or adopted — the +rest of this file is the design reasoning behind it, kept as written. + +## The facade's surface, end to end + +A consolidated view of what all the sections below amount to — every other +section is the reasoning behind one piece of this; this one is just the +result, in one place. + +**What it includes:** + +- `CatalogSession`/`IngestSession` — the engine: queue-then-commit, all-or- + nothing rollback on the catalog side (ingestion has none — a documented + gap, not an oversight), auto-derived idempotency keys, context inference. +- `%catalog`/`%ingest` — the surface a custodian actually touches: two + IPython magics, each a thin dispatcher onto one of the sessions above. +- Auto-registration — `import eea_datalakehouse.notebook` is the only setup + step; no `%load_ext`, no magic syntax to learn just to turn it on. +- A Comm channel — the receiving half of "click a catalog leaf, get a + working path," for whenever the JupyterLab extension's UI side exists. + +**What it exposes** — a small, curated verb set, not the full developer API: + +| catalog | ingestion | +| --- | --- | +| `copy`, `move` | `ingest` | +| `tag`, `untag` | | +| `set_wiki`, `delete_wiki`, `set_meta` | | +| `create_folder`, `delete_folder` | | +| `commit(retry=True)` | `commit(retry=True)` | + +Every call takes a path and the arguments a custodian actually thinks in +(`overwrite=True`, a list of tag strings, a folder path) — never an +`idempotency_key`, never a raw `SqlExecutor`/`CatalogRestClient`, never a +`Catalog`/`FolderIngest` object to manage. A path can be relative +(`.water_temperature`) once context exists, which it usually already does by +the second call in a session. + +**How a custodian uses it:** + +``` +import eea_datalakehouse.notebook + +%ingest ingest(folder="./bw_2026", target_catalog_path="bwd.reference", data_format="parquet") +%ingest commit(retry=True) + +%catalog tag(".water_temperature", ["reviewed"]) +%catalog set_meta(".", tags=[{"tag_name": "owner", "tag_value": "bw-team", "tag_title": "Owner"}]) +%catalog commit(retry=True) +``` + +Ingest first and commit it fully; only then touch the catalog side, in a +separate `%catalog` batch — that ordering is enforced by convention (two +sessions), not by code. Nothing reaches Dremio until a `commit`; a failure +prints a short message and (catalog side) undoes whatever it safely can, +rather than a traceback. + +## The problem + +`eea_datalakehouse.catalog` is built the way a developer library should be: +typed exceptions, explicit `idempotency_key`s, retry state, `overwrite`/`cascade` +flags with precise semantics. That is the right shape for the people writing +this package. It is the wrong shape for a data scientist in the EEA Lakehouse +JupyterLab (`eeadata/EEALakeHouse`) who wants to copy a table or tag a folder +once, in a notebook cell, without first learning what an idempotency key is or +why an operation can raise `CatalogOperationError`. + +This is the same gap `dev-notes.md`'s "The API is shaped for a developer, not +for a custodian writing a one-off script" entry already names for +`dds_ingestion` — worth reading together with this note, since a fix likely +wants to cover both packages the same way rather than twice. + +## Options considered + +1. **Thin notebook facade over the existing `Catalog` client** — a layer that + auto-generates `idempotency_key`s, catches the typed exceptions and prints a + short human-readable message instead of a traceback, and exposes a small set + of high-level verbs as plain functions or IPython `%magic` commands. + `Catalog`/`operations.py` stay exactly as they are underneath — the facade is + additive, not a fork. +2. **A full widget-based UI** (ipywidgets file-browser style) embedded in + JupyterLab — more discoverable for someone who has never touched the API at + all, but a much bigger build and an ongoing maintenance surface (widget + state, layout, JupyterLab version compatibility). + +**Leaning towards (1).** It is the smaller build, keeps one source of truth for +behaviour (the facade never reimplements retry/copy logic, only hides its +ceremony), and can ship incrementally — one friendly wrapper at a time — rather +than as a big-bang UI project. The tradeoff is real, though: the friendly layer +necessarily gives up some fine-grained control (custom idempotency keys, raw +exception handling) in exchange for fewer things to learn, so it has to stay a +layer *in front of* the developer API, never a replacement for it. + +## Session context lives in the Python process, not in Dremio + +The library is used from JupyterLab, so it already has a natural session +boundary: the kernel. The facade should keep its state there — in the Python +process, scoped to that one kernel's lifetime — rather than trying to model or +fetch any notion of "session" from Dremio itself, which has no such concept for +what we'd need (Dremio's REST API is stateless per call). Concretely, a +module-level or singleton facade object, created once per kernel, could hold: + +- a per-session id to seed auto-derived `idempotency_key`s from, so retries + within one notebook session are naturally scoped and stable without the + caller inventing anything; +- a "current" catalog path/context (e.g. the last folder touched) so repeated + calls in the same session can take a relative path instead of the full one + each time; +- cached best-effort lookups (`is_folder`/`is_table_or_view`) for the session's + lifetime, since re-checking Dremio on every call is pure overhead for a + script that touches the same handful of paths repeatedly. + +This resets cleanly when the kernel restarts — which matches how a data +scientist already thinks about a notebook session — and needs nothing new on +the Dremio side. + +`idempotency_key` in particular must be **fully encapsulated** — a custodian +should never see the concept, let alone pass one in. It's session-derived +plumbing (previous bullet), generated and threaded through entirely inside the +facade; it has no business appearing in a signature a data scientist calls. + +## Two sessions, not one — ingestion and catalog are sequential, not peers + +Catalog operations can't run before their target exists — a `datacopy`, tag or +wiki update on a table that hasn't been ingested yet has nothing to act on. So +ingestion and catalog aren't two halves of one atomic batch; they're two +phases, strictly ordered: an ingest fully lands and commits, *then* — and only +then — catalog steps referencing what it produced can be queued and committed. +That argues for **two independent sessions with two independent commits**, +each scoped to its own package, rather than one shared queue trying to unify +things that were never real peers: + +```python +ingest = lakehouse.ingest_session() +ingest.ingest(folder="./bw_2026", target_catalog_path="…/bwd/reference", ...) +ingest.commit(retry=True) # must fully succeed before anything below runs + +catalog = lakehouse.catalog_session() +catalog.copy("draft.raw_2026", "bwd.reference.water_temperature", overwrite=True) +catalog.tag("bwd.reference.water_temperature", reviewed_by="jdoe") +catalog.commit(retry=True) # all-or-nothing, but only across catalog steps +``` + +Each is still "queue calls, then commit" (the ergonomic point that started this +idea — one call that runs everything, not one call per REST request narrated +by hand), and each commit is still **all-or-nothing within its own package**: +every queued step succeeds, or `commit()` rolls back everything that package's +session already did, using that package's own compensating actions. What this +split removes is the harder problem from the earlier draft: a *cross-package* +saga, where an ingest step's rollback had to be composable with a catalog +step's rollback in one undo sequence. That's no longer needed — a catalog +session's rollback only ever touches catalog steps. + +**`catalog_session().commit()`'s own rollback** still needs the table below — +this part is unchanged from before, just scoped to catalog alone now: + +| step | compensating action | +| --- | --- | +| `datacopy`/`createfolder` | delete what was just created (`deletefolder`/a + table-and-data delete) | +| `settagsto`/`setwikito`/`setmeta2wiki` | restore the previous tags/wiki text + (captured before the step ran), or clear them if there was none | + +**`ingest_session().commit()`'s own rollback** — needed only if a batch queues +more than one ingest before committing — still runs into the same sharp edge +as before: **a delete-that-also-removes-the-backing-data operation for a +read-only ingest doesn't exist yet**, the gap recorded in +`docs/read-only-ingest-client-plan.md` ("Noted for later: deleting a read-only +table"), which already says it needs a DDS-side endpoint since this package +holds no S3 credentials to do it directly. Until that lands, an ingest +session's rollback guarantee is either limited to a single queued ingest at a +time, or stated plainly as unavailable for a multi-step ingest batch — but +critically, this no longer blocks the *catalog* session's rollback the way it +did when the two were one saga. + +Two more things this design has to answer, not just note as open: + +- **Compensation can itself fail** (the delete-the-copy call errors while + rolling back). `commit()` can't then pretend the batch is cleanly rolled + back — it needs a distinct failure mode ("rolled back" vs. "**rollback + incomplete**, these steps are unresolved") so a custodian isn't told + everything's fine when it isn't. +- **Retry and rollback are two different recoveries** for the same failure, and + the API has to make the caller choose: `commit(retry=True)` re-attempts the + failed step in place (existing `retry_state` machinery, previous section); + rolling back undoes everything instead. Defaulting to one or the other is a + product decision, not a technical one — worth settling explicitly rather than + picking implicitly by whichever gets built first. + +## Two magics, two sessions + +If we do offer `%magic` commands (previous section's "possibly also reachable +as..."), splitting them along the existing package boundary — `%catalog` for +the `catalog` verbs, `%ingest` for `dds_ingestion` — now maps directly onto the +two-session split above, rather than needing a shared queue underneath: each +magic owns its own session and its own `commit()`. That mirrors how the code is +already organised, gives a custodian tab-completion scoped to the right +vocabulary instead of one long mixed list, and matches the real dependency — +`%ingest` has to be run, and committed, before `%catalog` has anything to act +on. + +## Loading the magics without typing a magic to do it + +**Implemented.** `import eea_datalakehouse.notebook` registers `%catalog`/ +`%ingest` as a side effect (`notebook/__init__.py`'s `_autoregister`, guarded +against double-registering — and against `%load_ext` afterwards clobbering +it — by checking `magics_manager.registry` first) — so the cost is one +ordinary import line, not a magic syntax to memorize. `%load_ext +eea_datalakehouse.notebook.magics` still works, in either order. + +Getting to *zero* typing (no import either) needs something outside this +package's own code, at one of two levels: + +- **Environment provisioning, not a labextension** — an IPython startup file + (`~/.ipython/profile_default/startup/*.py`) or a kernel-spec argument that + runs the import/`%load_ext` at every kernel start. This could be shipped by + this package (a small installer helper) or baked into whatever builds the + Lakehouse JupyterLab image — it doesn't require the `eeadata/EEALakeHouse` + labextension (TypeScript/UI code) to change at all. +- **Actually modifying the JupyterLab extension** — only if the platform wants + this injected centrally at the Jupyter *server* level (a `jupyter_server` + extension hooking kernel startup) rather than per-environment config. The + heaviest option, and the only one that's genuinely "modify the extension + instead of the library." + +Neither of those two is built — recorded here as the next step if literally +zero typing turns out to matter, not assumed to be needed yet. + +## Credentials: what "hiding" can and can't mean + +A kernel is trusted user code, not a sandbox — anything the facade can read +(`os.environ["DREMIO_TOKEN"]`, a session's own private attributes) a +custodian's own cell can read too, deliberately or by accident. So "hide the +token" can only mean two different, both worth doing, things: + +- **Never let this package be the reason it leaks.** Already true today: the + token goes straight from `os.environ` into `Catalog(...)`'s constructor + inside `_build_catalog_session()` and never touches `self.shell.user_ns` + (the notebook's own variables), and neither `Catalog` nor + `RestSqlExecutor`/`CatalogRestClient` put it in a `repr()` or an error + message. Worth keeping as an explicit rule for anything added later to the + facade, not just an accident of how it happens to be written now. +- **Shrink what's actually at risk if a custodian *does* print it** (`%env`, a + stray `print(os.environ)`, then saving the notebook — a real and common way + secrets end up committed in `.ipynb` output cells). This package can't + prevent that; the platform can, by minting a short-lived, scoped-down token + per kernel session instead of injecting a long-lived PAT — so what leaks, if + anything does, expires soon and can't do much. That's a decision for + `eeadata/EEALakeHouse` (the Hub/image that populates the kernel's + environment), not this repo. + +## Pre-filling catalog context from a JupyterLab tree click + +A custodian should never have to know or type the top levels of a catalog +path (space, dataflow/domain, ...) — those are fixed for whatever project +they're in, not a per-call decision. This is the same idea "Session context +lives in the Python process" above already raised (`a "current" catalog +path/context ... so repeated calls can take a relative path`), sharpened with +a concrete trigger: a catalog-tree UI in the JupyterLab extension, where +selecting a leaf pre-fills that context automatically, rather than only +updating lazily from whatever path a call last touched. + +This splits into two sides that live in two different repositories. + +**The receiving side, here — implemented and encapsulated on both routes in:** +`CatalogSession.set_context(path)`/`_resolve` (`catalog/session.py`) resolve the +open question below with an explicit marker (a leading `.`, e.g. +`session.tag(".water_temperature", ...)`), and — critically — `set_context` +itself is not something a data custodian is expected to call: + +- **Ordinary use already keeps it current on its own.** Every queueing verb + updates the context from whatever path it just touched (the target's + parent for `copy`/`move`; the touched path itself for a folder-scoped verb + like `create_folder`/`set_meta`), so a script that never once calls + `set_context` still gets working relative paths after its first absolute + one. +- **The Comm channel below is the *other* caller**, not the custodian either. + +**The emitting side, in `eeadata/EEALakeHouse` — still not built:** the +catalog-tree UI's click handler needs to push the selected path through a +Jupyter Comm to the target name `eea_datalakehouse.catalog_context` +(`notebook/magics.py`'s `_register_context_comm` already listens for +`{"path": ""}` and applies it via `EEALakehouseMagics. +_apply_context` — silently, no cell runs, nothing a custodian could see or +type). That target name and payload shape is the interface contract; opening +the Comm and sending on it from the tree view is separate UI work in that +repository, which this session doesn't have open. + +## What the facade would concretely look like + +- Sensible defaults everywhere the developer API demands an explicit choice — + the session-derived `idempotency_key` above is the main one; there should be + no others once the facade design settles. +- Exceptions translated at the boundary into a short printed message (or a + `rich`/notebook-display block) rather than surfaced as a raw + `CatalogOperationError` traceback. +- A handful of high-level verbs matching how a custodian actually thinks about + the task, e.g. `session.copy(...)`, `session.tag(...)`, rather than the full + `datacopy`/`gettagsfrom`/`settagsto` vocabulary — queued and committed as + above, possibly also reachable as one-shot `%magic` commands for a true + single-call use. +- Ties into the same gaps `dev-notes.md` already flags: missing docstrings on + `Catalog`'s delegate methods (so `Shift+Tab` shows nothing useful today), and + no pinned notebook environment to develop/test this against + (`jupyterlab`/`ipykernel` extra, per that file's last finding). + +## Open questions + +- Where should the facade live — a new module in this package + (`eea_datalakehouse.catalog.notebook`?), or a separate thin package installed + alongside it in the Lakehouse JupyterLab image? +- `%magic` commands vs. plain friendly functions/methods — magics are more + discoverable inside a notebook cell, but harder to discover *outside* one + (no `Shift+Tab`, no import to `help()`), and less testable with the normal + pytest tooling this repo already uses. +- Does this depend on, or should it wait for, the docstring and + notebook-environment gaps already recorded in `dev-notes.md`? diff --git a/docs/notebooks/catalog_session_example.ipynb b/docs/notebooks/catalog_session_example.ipynb new file mode 100644 index 0000000..b9f067e --- /dev/null +++ b/docs/notebooks/catalog_session_example.ipynb @@ -0,0 +1,156 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# `%catalog` — queue, commit, rollback\n", + "\n", + "A worked example of the `%catalog` magic from `eea_datalakehouse.notebook.magics` — see\n", + "`docs/notebook-facade-for-data-scientists.md` for the design behind it.\n", + "\n", + "**Before running this for real:** set `DREMIO_BASE_URL`, `DREMIO_TOKEN` (and `DREMIO_USERNAME`\n", + "if you'll queue `copy`/`move`) in the kernel environment — the same variables\n", + "`debugger/debug_run.py` uses. Install the extra this needs once: `pip install \"EEADataLakehouse[notebook]\"`.\n", + "\n", + "Nothing below invents new vocabulary: every call after `%catalog` is a real\n", + "`CatalogSession` method (`src/eea_datalakehouse/catalog/session.py`) — this notebook is a tour of\n", + "that class, not a separate API." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "import eea_datalakehouse.notebook # registers %catalog/%ingest — no %load_ext needed" + }, + { + "cell_type": "markdown", + "id": "f469f7f5", + "source": "`%load_ext eea_datalakehouse.notebook.magics` still works too, and is safe to run either\nbefore or after the import above — whichever runs first registers the magics, the other is\na no-op (see `magics.load_ipython_extension`'s docstring).", + "metadata": {} + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Queue a few steps\n", + "\n", + "Each cell only **queues** an operation — nothing reaches Dremio yet. The same one\n", + "`CatalogSession` is reused across every `%catalog` cell in this kernel (see \"Session\n", + "context lives in the Python process\" in the design doc), so calls in different cells\n", + "accumulate on the same batch." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%catalog copy(\"bwd.draft.raw_2026\", \"bwd.reference.water_temperature\", overwrite=True)" + ] + }, + { + "cell_type": "markdown", + "id": "d1f58114", + "source": "%catalog tag(\".water_temperature\", [\"reviewed\", \"2026\"])", + "metadata": {} + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%catalog tag(\"bwd.reference.water_temperature\", [\"reviewed\", \"2026\"])" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%catalog set_meta(\"bwd.reference\", tags=[\n", + " {\"tag_name\": \"owner\", \"tag_value\": \"bathing-water-team\", \"tag_title\": \"Owner\"},\n", + "], overwrite=False)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Commit — all-or-nothing\n", + "\n", + "`commit()` runs every queued step in order. `retry=True` re-attempts a step that hits a\n", + "Dremio engine still warming up (`EngineStartingError`) a few times before giving up — any\n", + "other error rolls back immediately regardless. Either way the queue is empty afterwards,\n", + "whether this cell succeeds or raises." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%catalog commit(retry=True)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "`commit` with no parentheses also works, as a shorthand:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%catalog commit" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## What a failed commit looks like\n", + "\n", + "If a step in the batch fails, `%catalog` prints a short message instead of a full\n", + "traceback (see the design doc's \"Exceptions translated at the boundary\") — for example,\n", + "queuing a step against a target that already exists without `overwrite=True`:\n", + "\n", + "```\n", + "%catalog copy(\"bwd.draft.raw_2026\", \"bwd.reference.water_temperature\")\n", + "%catalog commit\n", + "# -> catalog error: commit failed at step 1/1 (copy 'bwd.draft.raw_2026' -> ...):\n", + "# target 'bwd.reference.water_temperature' already exists — pass overwrite=True ...\n", + "# Everything before it was rolled back.\n", + "```\n", + "\n", + "A batch mixing a reversible step with an `overwrite=True` step is explicit about what it\n", + "could *not* undo rather than pretending the rollback was clean — see `CatalogCommitError`'s\n", + "docstring, and \"Must be all-or-nothing\" in the design doc, for exactly which verbs that\n", + "applies to (`delete_folder`, and `copy`/`move` when `overwrite=True`)." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.11" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} \ No newline at end of file diff --git a/docs/notebooks/ingest_session_example.ipynb b/docs/notebooks/ingest_session_example.ipynb new file mode 100644 index 0000000..4a4e74e --- /dev/null +++ b/docs/notebooks/ingest_session_example.ipynb @@ -0,0 +1,144 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# `%ingest` — queue folder ingests, commit as one batch\n", + "\n", + "A worked example of the `%ingest` magic from `eea_datalakehouse.notebook.magics` — see\n", + "`docs/notebook-facade-for-data-scientists.md` for the design behind it.\n", + "\n", + "**Before running this for real:** the kernel environment needs whatever\n", + "`eea_datalakehouse.dds_ingestion.credentials.load_creds`/`load_base_url` expect (same as\n", + "importing `FolderIngest` directly today). Install the extra this needs once:\n", + "`pip install \"EEADataLakehouse[notebook]\"`.\n", + "\n", + "**This is a separate session from `%catalog`'s, on purpose.** A catalog operation can't run\n", + "before its target has actually been ingested, so the two were never one atomic batch — see\n", + "\"Two sessions, not one\" in the design doc. Ingest and commit here first; only afterwards, in\n", + "a fresh set of `%catalog` cells (see `catalog_session_example.ipynb`), tag or copy what just\n", + "landed." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "import eea_datalakehouse.notebook # registers %catalog/%ingest — no %load_ext needed" + }, + { + "cell_type": "markdown", + "id": "d9463089", + "source": "`%load_ext eea_datalakehouse.notebook.magics` still works too, and is safe to run either\nbefore or after the import above — whichever runs first registers the magics, the other is\na no-op (see `magics.load_ipython_extension`'s docstring).", + "metadata": {} + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Queue one or more folder ingests\n", + "\n", + "Each call only records the intent — `folder`/`target_catalog_path`/`data_format` plus\n", + "anything else `FolderIngest` accepts (`intent`, `table_name`, `sub_path`, ...). Nothing\n", + "uploads until `commit()`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%ingest ingest(folder=\"./bw_2026\", target_catalog_path=\"bwd.reference\", data_format=\"parquet\", intent=\"read_only\", table_name=\"water_temperature\", sub_path=\"2026\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "A second ingest into a different table, queued on the *same* session — both run when the\n", + "cell below commits:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%ingest ingest(folder=\"./bw_stations_2026\", target_catalog_path=\"bwd.reference\", data_format=\"parquet\", table_name=\"stations\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Commit — runs every queued ingest in order\n", + "\n", + "Unlike `%catalog`'s `commit()`, this one **cannot roll back** an ingest that already\n", + "succeeded — see the design doc and `docs/read-only-ingest-client-plan.md` for why (deleting\n", + "an ingested table's backing data needs a server-side operation this package does not have\n", + "yet). If the second ingest above were to fail, the first one's table stays exactly as\n", + "ingested — `retry=True` only covers resuming a single failed ingest\n", + "(`FolderIngest.retry`'s own resume-the-load-step / re-upload-under-a-new-session logic),\n", + "not undoing an earlier one." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%ingest commit(retry=True)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## What a partial failure looks like\n", + "\n", + "```\n", + "%ingest commit\n", + "# -> ingest error: ingest './bw_stations_2026' -> 'bwd.reference' failed (...); 1 earlier\n", + "# ingest(s) in this batch already committed and CANNOT be undone\n", + "```\n", + "\n", + "That message is deliberately explicit about what already landed — check\n", + "`IngestCommitError.succeeded` (or just the catalog) before deciding what to do about the one\n", + "that failed, rather than assuming the whole batch was undone." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Now the catalog side, in a fresh session\n", + "\n", + "Once the ingest above has actually committed, a *separate* `%catalog` batch can tag or copy\n", + "what just landed — see `catalog_session_example.ipynb`:\n", + "\n", + "```\n", + "%catalog tag(\"bwd.reference.water_temperature\", [\"reviewed\", \"2026\"])\n", + "%catalog commit\n", + "```" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.11" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index c47a407..e5bbbef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,9 @@ dependencies = [ ] [project.optional-dependencies] +notebook = [ + "ipython>=8.18", # eea_datalakehouse.notebook's %catalog/%ingest magics +] dev = [ "pytest>=7", "ruff>=0.15", @@ -37,6 +40,7 @@ dev = [ "respx>=0.23", # mocks httpx for the ingestion test suite "types-tqdm", "ipykernel>=6.29", # run/debug notebooks against this environment + "EEADataLakehouse[notebook]", ] [project.urls] diff --git a/src/eea_datalakehouse/catalog/__init__.py b/src/eea_datalakehouse/catalog/__init__.py index ee28263..4ea61f4 100644 --- a/src/eea_datalakehouse/catalog/__init__.py +++ b/src/eea_datalakehouse/catalog/__init__.py @@ -30,12 +30,12 @@ deleteview, deletewiki, draft2version, + getmetafromwiki, gettableitemsfrom, gettablesfrom, gettagsfrom, getwikifrom, publishversion, - getmetafromwiki, retry_pending, setmeta2wiki, settagsto, @@ -43,6 +43,7 @@ table2view, ) from .rest import CatalogRestClient +from .session import CatalogCommitError, CatalogSession, CatalogSessionError, CommitReport from .sql import ( FLIGHT_LOCATION_ENV_VAR, TRANSPORT_ENV_VAR, @@ -57,8 +58,12 @@ "FLIGHT_LOCATION_ENV_VAR", "TRANSPORT_ENV_VAR", "Catalog", + "CatalogCommitError", "CatalogOperationError", "CatalogRestClient", + "CatalogSession", + "CatalogSessionError", + "CommitReport", "EngineStartingError", "FlightSqlExecutor", "RestSqlExecutor", diff --git a/src/eea_datalakehouse/catalog/session.py b/src/eea_datalakehouse/catalog/session.py new file mode 100644 index 0000000..b90802b --- /dev/null +++ b/src/eea_datalakehouse/catalog/session.py @@ -0,0 +1,524 @@ +"""`CatalogSession` — queue catalog verbs, then commit them as one batch. + +Prototype of the design in `docs/notebook-facade-for-data-scientists.md` +("Queue calls, then commit"). Wraps an existing `Catalog`, so nothing here +reimplements catalog behaviour — it only hides the ceremony (idempotency +keys) and adds queue/commit/rollback on top:: + + session = CatalogSession(catalog) + session.copy("draft.raw_2026", "bwd.reference.water_temperature") + session.tag("bwd.reference.water_temperature", ["reviewed"]) + session.commit(retry=True) + +`commit()` is all-or-nothing: every queued step runs in order, and the +moment one fails, everything already done in *this* commit is undone before +the failure is reported — see each verb's own docstring below for exactly +what its compensating action is, and where one isn't possible (`delete_folder`, +or `copy`/`move` with `overwrite=True`) that limitation is raised as part of +the failure, not hidden. + +This is NOT the same "session" `dds_ingestion.folder.FolderIngest` already +has (a server-side transfer identified by `session_id`, resumed via +`attach`/`retry`) — `CatalogSession` is a client-side batch queue with no +server-side counterpart at all; the two just happen to share an overloaded +English word. See `eea_datalakehouse.dds_ingestion.session.IngestSession` +for the ingestion-side equivalent of *this* kind of session — deliberately a +separate object with its own `commit()`, not a shared queue, because a +catalog operation can't run before its target has actually been ingested +(see the design doc's "Two sessions, not one"). +""" + +from __future__ import annotations + +import time +import uuid +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Literal + +from . import operations +from .client import Catalog +from .errors import CatalogOperationError, EngineStartingError + +_Undo = Callable[[], None] + + +class CatalogSessionError(RuntimeError): + """Base for `CatalogSession`-specific errors.""" + + +class CatalogCommitError(CatalogSessionError): + """`commit()` failed partway through the batch. + + `rolled_back` is `True` only if every already-succeeded step in this + commit was cleanly undone. When it's `False`, `unresolved` names which + steps are NOT undone — either because that verb has no compensating + action at all (`delete_folder`; `copy`/`move` with `overwrite=True`), or + because undoing it was attempted and itself failed. Either way, the + catalog is left in a state this object cannot fully explain away, and a + human needs to look. + """ + + def __init__( + self, + message: str, + *, + failed_step: str, + original_error: Exception, + rolled_back: bool, + unresolved: list[str], + ) -> None: + super().__init__(message) + self.failed_step = failed_step + self.original_error = original_error + self.rolled_back = rolled_back + self.unresolved = unresolved + + +@dataclass +class CommitReport: + """What a successful `commit()` actually did, in order.""" + + succeeded: list[str] = field(default_factory=list) + + +class _Irreversible(Exception): # noqa: N818 — internal control-flow signal, not a user-facing error + """Raised by a step's `run` to say: this succeeded, but cannot be undone.""" + + +@dataclass +class _Step: + description: str + run: Callable[[Catalog, str], _Undo | None] + + +def _read_wiki_or_none(catalog: Catalog, path: str, *, idempotency_key: str) -> str | None: + """The wiki text at `path`, or `None` if it doesn't have one (or doesn't + exist) — `getwikifrom` can't tell those two apart, and neither needs to + here: either way there's nothing to restore on rollback.""" + try: + return catalog.getwikifrom(path, idempotency_key=idempotency_key) + except CatalogOperationError: + return None + + +def _drop_entry(catalog: Catalog, path: str, *, idempotency_key: str) -> None: + """Drop whatever entity (table or view) now sits at `path`. + + Used to undo a fresh `copy`. Reaches into `operations`' own private + helpers (`_entry_kind`, `_quote_path`) rather than a public op, because + there isn't one: `deleteview` only drops VIEWs, and a plain `datacopy` + target is always a TABLE (`CREATE TABLE ... AS SELECT`). Acceptable + here since this module lives in the same package as `operations.py`. + """ + executor = catalog._flight_executor # noqa: SLF001 — same-package internal, see docstring + kind = operations._entry_kind(executor, path, idempotency_key=idempotency_key) # noqa: SLF001 + executor.execute( + f"DROP {kind} IF EXISTS {operations._quote_path(path)}", # noqa: SLF001 + idempotency_key=idempotency_key, + ) + + +class CatalogSession: + """Queue catalog verbs against `catalog`, then `commit()` them as one batch. + + Every queueing method (`copy`, `move`, `tag`, `untag`, `set_wiki`, + `delete_wiki`, `set_meta`, `create_folder`, `delete_folder`) only records + the intent — nothing reaches Dremio until `commit()`. Each returns + `self`, so calls chain:: + + session.copy(...).tag(...).commit() + + `idempotency_key`s are generated internally (`session--`, one per + step) — never pass or think about one; that's exactly the ceremony this + class exists to hide (see the design doc's "must be fully encapsulated"). + """ + + def __init__(self, catalog: Catalog) -> None: + self._catalog = catalog + self._id = uuid.uuid4().hex[:8] + self._steps: list[_Step] = [] + self._next_step = 1 + self._context: str | None = None + + def __repr__(self) -> str: + return f"CatalogSession(pending={len(self._steps)})" + + def _key(self, suffix: str = "") -> str: + key = f"session-{self._id}-{self._next_step}{suffix}" + return key + + def set_context(self, path: str | None) -> CatalogSession: + """Set the "current" catalog path — a later relative path (a leading + `.`, e.g. `.water_temperature`) resolves against this. `None` clears it. + + Not something a data custodian should normally call: ordinary use + already keeps this up to date on its own (see `_resolve` — every + queued step updates it from whatever path it just touched), and an + integration (the JupyterLab catalog-tree extension pushing a + selected leaf through a Comm — see + `docs/notebook-facade-for-data-scientists.md`, "Pre-filling catalog + context") is the other caller, seeding it before any path has been + typed yet. + """ + self._context = path + return self + + def _resolve_path(self, path: str) -> str: + """Just the relative -> absolute resolution (a leading `.` resolves + against the current context) — does NOT update the context; see + `_resolve` for the verbs where the touched path also becomes the + new context.""" + if not path.startswith("."): + return path + if self._context is None: + raise CatalogSessionError( + f"{path!r} is relative (starts with '.') but no context is set yet — " + "use an absolute path first, or call set_context()" + ) + return f"{self._context}{path}" + + def _resolve(self, path: str) -> str: + """Resolve `path` (see `_resolve_path`), then update the context to + its own parent — so the next short name keeps resolving against + wherever this one just landed, without anyone calling + `set_context` again. Verbs with two paths (`copy`/`move`) resolve + both against the *same* starting context via `_resolve_path` + directly instead — only the target should become the new context, + not whatever `source_path`'s own folder happens to be.""" + resolved = self._resolve_path(path) + self._context = operations._parent_path(resolved) or self._context # noqa: SLF001 + return resolved + + # -- queueing verbs --------------------------------------------------- + + def copy( + self, + source_path: str, + target_path: str, + *, + overwrite: bool = False, + create_target_folder: bool = False, + ) -> CatalogSession: + """Queue a `datacopy`. Reversible only when `overwrite=False` (there + was nothing at `target_path` to lose) — undo drops the table this + step created. With `overwrite=True`, whatever used to be at + `target_path` is gone the moment this step runs; there is nothing to + restore, so this step cannot be undone (see `CatalogCommitError`). + + Either path may be relative (a leading `.`) to the session's current + context — see `set_context`; `target_path` becomes the new context + afterwards.""" + source_path = self._resolve_path(source_path) + target_path = self._resolve_path(target_path) + self._context = operations._parent_path(target_path) or self._context # noqa: SLF001 + + def run(catalog: Catalog, key: str) -> _Undo | None: + catalog.datacopy( + source_path, + target_path, + overwrite=overwrite, + create_target_folder=create_target_folder, + idempotency_key=key, + ) + if overwrite: + raise _Irreversible("overwrote an existing target; its previous content is gone") + + def undo() -> None: + _drop_entry(catalog, target_path, idempotency_key=f"{key}-undo") + + return undo + + self._steps.append(_Step(f"copy {source_path!r} -> {target_path!r}", run)) + return self + + def move( + self, + source_path: str, + target_path: str, + *, + entry_type: Literal["TABLE", "VIEW"] | None = None, + overwrite: bool = False, + create_target_folder: bool = False, + ) -> CatalogSession: + """Queue a `datamove`. Reversible only when `overwrite=False` — undo + is a `datamove` back from `target_path` to `source_path` (the + target's kind is re-detected at undo time, not assumed). Same + `overwrite=True` limitation as `copy`. + + Either path may be relative (a leading `.`) to the session's current + context — see `set_context`; `target_path` becomes the new context + afterwards.""" + source_path = self._resolve_path(source_path) + target_path = self._resolve_path(target_path) + self._context = operations._parent_path(target_path) or self._context # noqa: SLF001 + + def run(catalog: Catalog, key: str) -> _Undo | None: + catalog.datamove( + source_path, + target_path, + entry_type=entry_type, + overwrite=overwrite, + create_target_folder=create_target_folder, + idempotency_key=key, + ) + if overwrite: + raise _Irreversible("overwrote an existing target; its previous content is gone") + + def undo() -> None: + undo_key = f"{key}-undo" + kind = operations._entry_kind( # noqa: SLF001 — see _drop_entry + catalog._flight_executor, # noqa: SLF001 + target_path, + idempotency_key=f"{undo_key}-detect", + ) + catalog.datamove( + target_path, + source_path, + entry_type=kind, + overwrite=False, + idempotency_key=undo_key, + ) + + return undo + + self._steps.append(_Step(f"move {source_path!r} -> {target_path!r}", run)) + return self + + def tag(self, path: str, tags: list[str]) -> CatalogSession: + """Queue `settagsto` (replaces the tag set). Undo restores whatever + tags were on `path` immediately before this step ran. + + `path` may be relative (a leading `.`) to the session's current + context — see `set_context`; it becomes the new context afterwards.""" + path = self._resolve(path) + + def run(catalog: Catalog, key: str) -> _Undo: + existing = catalog.gettagsfrom(path, idempotency_key=f"{key}-read") + catalog.settagsto(path, tags, idempotency_key=key) + + def undo() -> None: + catalog.settagsto(path, existing, idempotency_key=f"{key}-undo") + + return undo + + self._steps.append(_Step(f"tag {path!r} with {tags!r}", run)) + return self + + def untag(self, path: str, tags: list[str]) -> CatalogSession: + """Queue `deletetags`. Undo restores the full tag set `path` had + immediately before this step ran. + + `path` may be relative (a leading `.`) to the session's current + context — see `set_context`; it becomes the new context afterwards.""" + path = self._resolve(path) + + def run(catalog: Catalog, key: str) -> _Undo: + existing = catalog.gettagsfrom(path, idempotency_key=f"{key}-read") + catalog.deletetags(path, tags, idempotency_key=key) + + def undo() -> None: + catalog.settagsto(path, existing, idempotency_key=f"{key}-undo") + + return undo + + self._steps.append(_Step(f"untag {tags!r} from {path!r}", run)) + return self + + def set_wiki( + self, path: str, text: str, *, tags: list[dict[str, str]] | None = None + ) -> CatalogSession: + """Queue `setwikito`. Undo restores the previous wiki text verbatim, + or deletes it if `path` had none. + + `path` may be relative (a leading `.`) to the session's current + context — see `set_context`; it becomes the new context afterwards.""" + path = self._resolve(path) + + def run(catalog: Catalog, key: str) -> _Undo: + existing = _read_wiki_or_none(catalog, path, idempotency_key=f"{key}-read") + catalog.setwikito(path, text, tags=tags, idempotency_key=key) + + def undo() -> None: + undo_key = f"{key}-undo" + if existing is None: + catalog.deletewiki(path, idempotency_key=undo_key) + else: + catalog.setwikito(path, existing, idempotency_key=undo_key) + + return undo + + self._steps.append(_Step(f"set wiki on {path!r}", run)) + return self + + def delete_wiki(self, path: str) -> CatalogSession: + """Queue `deletewiki`. Undo restores the previous wiki text, if there + was one — a no-op if `path` had no wiki to begin with. + + `path` may be relative (a leading `.`) to the session's current + context — see `set_context`; it becomes the new context afterwards.""" + path = self._resolve(path) + + def run(catalog: Catalog, key: str) -> _Undo | None: + existing = _read_wiki_or_none(catalog, path, idempotency_key=f"{key}-read") + catalog.deletewiki(path, idempotency_key=key) + if existing is None: + return None + + def undo() -> None: + catalog.setwikito(path, existing, idempotency_key=f"{key}-undo") + + return undo + + self._steps.append(_Step(f"delete wiki on {path!r}", run)) + return self + + def set_meta( + self, path: str, tags: list[dict[str, str]] | None = None, *, overwrite: bool = True + ) -> CatalogSession: + """Queue `setmeta2wiki`. Undo restores the whole previous wiki text + (not just its Meta Data section) verbatim, or deletes it if `path` + had none. + + `path` may be relative (a leading `.`) to the session's current + context — see `set_context`. Unlike a leaf-level verb (`tag`, + `set_wiki`, ...), `path` here is a folder (see `_require_folder`), + so it becomes the new context *itself*, not its parent — the next + short name is expected to name something *inside* it.""" + path = self._resolve_path(path) + self._context = path + + def run(catalog: Catalog, key: str) -> _Undo: + existing = _read_wiki_or_none(catalog, path, idempotency_key=f"{key}-read") + catalog.setmeta2wiki(path, tags=tags, overwrite=overwrite, idempotency_key=key) + + def undo() -> None: + undo_key = f"{key}-undo" + if existing is None: + catalog.deletewiki(path, idempotency_key=undo_key) + else: + catalog.setwikito(path, existing, idempotency_key=undo_key) + + return undo + + self._steps.append(_Step(f"set metadata on {path!r}", run)) + return self + + def create_folder(self, path: str, *, create_parents: bool = False) -> CatalogSession: + """Queue `createfolder`. Undo deletes it — but only if this step + actually created it; `createfolder` is idempotent, so a `path` + that already existed is left alone on rollback too. + + `path` may be relative (a leading `.`) to the session's current + context — see `set_context` — and becomes the new context *itself* + afterwards (a folder's contents, not its parent, is where a + following short name most likely points).""" + path = self._resolve_path(path) + self._context = path + + def run(catalog: Catalog, key: str) -> _Undo | None: + already_there = catalog._catalog_rest.exists(path) # noqa: SLF001 — see module docstring + catalog.createfolder(path, create_parents=create_parents, idempotency_key=key) + if already_there: + return None + + def undo() -> None: + catalog.deletefolder(path, cascade=False, idempotency_key=f"{key}-undo") + + return undo + + self._steps.append(_Step(f"create folder {path!r}", run)) + return self + + def delete_folder(self, path: str, *, cascade: bool = False) -> CatalogSession: + """Queue `deletefolder`. **Never reversible** — a deleted folder's + contents cannot be recreated, so this always leaves the commit + unable to claim a clean rollback if a later step fails. + + `path` may be relative (a leading `.`) to the session's current + context — see `set_context`. Does not change the context itself — + there's nothing meaningful to navigate into once it's deleted.""" + path = self._resolve_path(path) + + def run(catalog: Catalog, key: str) -> _Undo | None: + catalog.deletefolder(path, cascade=cascade, idempotency_key=key) + raise _Irreversible("deleted folder contents cannot be recreated") + + self._steps.append(_Step(f"delete folder {path!r}", run)) + return self + + # -- commit ------------------------------------------------------------- + + def commit( + self, *, retry: bool = False, max_retries: int = 3, retry_delay: float = 5.0 + ) -> CommitReport: + """Run every queued step, in order. All-or-nothing: the moment one + fails, everything this commit already did is undone (reverse order) + before `CatalogCommitError` is raised — see that class and each + verb's docstring for what "undone" can and can't cover. + + `retry=True` re-attempts a step that raised `EngineStartingError` + (a Dremio engine still warming up) up to `max_retries` times, + `retry_delay` seconds apart, before giving up and rolling back — + every operation here is documented safe to retry from the start + with the same idempotency key, so this just re-runs the same step. + Any other exception rolls back immediately regardless of `retry`; + retrying a real error (bad path, permission denied, ...) would not + help. + + The queue is cleared either way — a failed `commit()` does not + leave the failing step (or the ones after it) still queued. + """ + executed: list[tuple[_Step, _Undo | None]] = [] + irreversible: list[str] = [] + steps, self._steps = self._steps, [] + current: _Step | None = None + try: + for current in steps: + key = self._key() + self._next_step += 1 + attempt = 0 + while True: + try: + try: + undo = current.run(self._catalog, key) + except _Irreversible as marker: + undo = None + irreversible.append(f"{current.description} ({marker})") + executed.append((current, undo)) + break + except EngineStartingError: + attempt += 1 + if retry and attempt <= max_retries: + time.sleep(retry_delay) + continue + raise + except Exception as exc: # noqa: BLE001 — reported via CatalogCommitError below + rollback_problems = self._rollback(executed) + unresolved = irreversible + rollback_problems + assert current is not None + raise CatalogCommitError( + f"commit failed at step {len(executed) + 1}/{len(steps)} " + f"({current.description}): {exc}. " + + ( + "Everything before it was rolled back." + if not unresolved + else f"ROLLBACK INCOMPLETE — see .unresolved: {unresolved}" + ), + failed_step=current.description, + original_error=exc, + rolled_back=not unresolved, + unresolved=unresolved, + ) from exc + return CommitReport(succeeded=[step.description for step, _ in executed]) + + def _rollback(self, executed: list[tuple[_Step, _Undo | None]]) -> list[str]: + problems: list[str] = [] + for step, undo in reversed(executed): + if undo is None: + continue + try: + undo() + except Exception as exc: # noqa: BLE001 — best-effort: collect and keep going + problems.append(f"{step.description}: rollback failed ({exc})") + return problems diff --git a/src/eea_datalakehouse/dds_ingestion/__init__.py b/src/eea_datalakehouse/dds_ingestion/__init__.py index af9f631..ffa3ed6 100644 --- a/src/eea_datalakehouse/dds_ingestion/__init__.py +++ b/src/eea_datalakehouse/dds_ingestion/__init__.py @@ -61,6 +61,7 @@ UploadPart, UploadTarget, ) +from .session import IngestCommitError, IngestCommitReport, IngestSession, IngestSessionError __all__ = [ "DEFAULT_PARALLELISM", @@ -73,7 +74,11 @@ "FolderIngest", "IngestApiError", "IngestClient", + "IngestCommitError", + "IngestCommitReport", "IngestOutcome", + "IngestSession", + "IngestSessionError", "IngestStateError", "Intent", "MissingCredentialsError", diff --git a/src/eea_datalakehouse/dds_ingestion/session.py b/src/eea_datalakehouse/dds_ingestion/session.py new file mode 100644 index 0000000..0e8556f --- /dev/null +++ b/src/eea_datalakehouse/dds_ingestion/session.py @@ -0,0 +1,206 @@ +"""`IngestSession` — queue folder ingests, then run them as one batch. + +Prototype of the design in `docs/notebook-facade-for-data-scientists.md` +("Queue calls, then commit" / "Two sessions, not one"):: + + session = IngestSession() + session.ingest(folder="./bw_2026", target_catalog_path="bwd.reference", + data_format="parquet", table_name="water_temperature") + session.commit(retry=True) + +Deliberately a **separate** object from `eea_datalakehouse.catalog.session. +CatalogSession`, not a shared queue: a catalog operation can't run before its +target has actually been ingested, so the two were never one atomic batch — +run an `IngestSession.commit()` to completion first, then build a +`CatalogSession` for whatever comes after. + +**No rollback.** Unlike `CatalogSession`, a `commit()` here cannot undo an +ingest that already succeeded — once a folder's files are uploaded and +loaded into a table, removing them needs a delete-that-also-removes-the- +backing-data operation this package does not have yet (see +`docs/read-only-ingest-client-plan.md`, "Noted for later: deleting a +read-only table"). So a partial failure is *reported*, not undone: +`IngestCommitError.succeeded` lists every ingest that already landed and +stays landed — check it before deciding what to do next. + +This "session" is unrelated to `FolderIngest`'s own transfer session +(`session_id`, `attach`, `retry`) — that's a server-side handle for one +ingest; this is a client-side queue of possibly many. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .client import IngestClient +from .credentials import DremioCreds +from .folder import DEFAULT_PARALLELISM, FolderIngest, IngestOutcome, IngestStateError +from .models import DataFormat, Intent + + +class IngestSessionError(RuntimeError): + """Base for `IngestSession`-specific errors.""" + + +class IngestCommitError(IngestSessionError): + """`commit()` failed partway through the batch. + + `succeeded` holds the `IngestOutcome` of every ingest that already + completed before the failure — those cannot be undone (see the module + docstring) and stay exactly as ingested. + """ + + def __init__( + self, + message: str, + *, + failed_index: int, + original_error: Exception, + succeeded: list[IngestOutcome], + ) -> None: + super().__init__(message) + self.failed_index = failed_index + self.original_error = original_error + self.succeeded = succeeded + + +@dataclass +class IngestCommitReport: + """What a successful `commit()` actually did, in order.""" + + outcomes: list[IngestOutcome] = field(default_factory=list) + + +@dataclass +class _QueuedIngest: + folder: str | Path + target_catalog_path: str + data_format: DataFormat + kwargs: dict[str, Any] + + def describe(self) -> str: + return f"ingest {self.folder!r} -> {self.target_catalog_path!r}" + + +class IngestSession: + """Queue `FolderIngest` runs, then `commit()` them as one batch. + + `ingest(...)` takes exactly `FolderIngest`'s own constructor arguments + (`folder`, `target_catalog_path`, `data_format`, plus anything else it + accepts — `intent`, `table_name`, `sub_path`, ...) and only records the + intent; nothing runs until `commit()`. `client`/`base_url`/`creds`, + if given here, are shared across every queued ingest exactly like + passing them to `FolderIngest` directly (kernel-environment defaults + otherwise — see `credentials.load_base_url`/`load_creds`). + """ + + def __init__( + self, + *, + client: IngestClient | None = None, + base_url: str | None = None, + creds: DremioCreds | None = None, + ) -> None: + self._client = client + self._base_url = base_url + self._creds = creds + self._pending: list[_QueuedIngest] = [] + + def __repr__(self) -> str: + return f"IngestSession(pending={len(self._pending)})" + + def ingest( + self, + folder: str | Path, + target_catalog_path: str, + *, + data_format: DataFormat, + intent: Intent = "read_only", + conflict_mode: str = "fail", + table_name: str | None = None, + sub_path: str | None = None, + parallelism: int = DEFAULT_PARALLELISM, + multipart: bool | None = None, + show_progress: bool = True, + ) -> IngestSession: + """Queue one folder ingest — see `FolderIngest` for what each argument + means. No `idempotency_key` here: unlike the catalog side, a queued + ingest's own key would only ever be used once per `commit()`, so + there is nothing to encapsulate — `FolderIngest` already defaults + it to `None` (a fresh session every run).""" + self._pending.append( + _QueuedIngest( + folder=folder, + target_catalog_path=target_catalog_path, + data_format=data_format, + kwargs={ + "intent": intent, + "conflict_mode": conflict_mode, + "table_name": table_name, + "sub_path": sub_path, + "parallelism": parallelism, + "multipart": multipart, + "show_progress": show_progress, + }, + ) + ) + return self + + def commit(self, *, retry: bool = False, max_retries: int = 3) -> IngestCommitReport: + """Run every queued ingest, in order. Stops at the first failure — + see the module docstring: nothing already succeeded can be rolled + back, so `IngestCommitError.succeeded` is the record of what to deal + with by hand. + + `retry=True` calls `FolderIngest.retry()` (resume-the-load-step or + re-upload-under-a-new-session, whichever the server says applies — + see that method's own docstring) up to `max_retries` times before + giving up on a failed ingest. + + The queue is cleared either way. + """ + succeeded: list[IngestOutcome] = [] + pending, self._pending = self._pending, [] + for index, item in enumerate(pending): + job = FolderIngest( + item.folder, + item.target_catalog_path, + data_format=item.data_format, + client=self._client, + base_url=self._base_url, + creds=self._creds, + **item.kwargs, + ) + try: + outcome = self._run_with_retry(job, retry=retry, max_retries=max_retries) + except Exception as exc: + raise IngestCommitError( + f"{item.describe()} failed ({exc}); {len(succeeded)} earlier ingest(s) in " + "this batch already committed and CANNOT be undone", + failed_index=index, + original_error=exc, + succeeded=succeeded, + ) from exc + finally: + job.close() + succeeded.append(outcome) + return IngestCommitReport(outcomes=succeeded) + + @staticmethod + def _run_with_retry(job: FolderIngest, *, retry: bool, max_retries: int) -> IngestOutcome: + try: + return job.run() + except Exception as exc: + if not retry: + raise + last_error: Exception = exc + for _ in range(max_retries): + try: + return job.retry() + except IngestStateError: + raise # retry() itself says this isn't resumable — retrying again won't help + except Exception as retry_exc: # noqa: BLE001 — keep trying up to max_retries + last_error = retry_exc + raise last_error from exc diff --git a/src/eea_datalakehouse/notebook/__init__.py b/src/eea_datalakehouse/notebook/__init__.py new file mode 100644 index 0000000..13578ac --- /dev/null +++ b/src/eea_datalakehouse/notebook/__init__.py @@ -0,0 +1,36 @@ +"""The notebook-facing surface — `%catalog`/`%ingest` magics over +`CatalogSession`/`IngestSession` (see `docs/notebook-facade-for-data- +scientists.md`). Requires IPython, which the rest of this package does not +— install the `notebook` extra (`pip install "EEADataLakehouse[notebook]"`) +to get it. + +Importing this package inside a running IPython shell registers both magics +as a side effect — a custodian only ever needs:: + + import eea_datalakehouse.notebook + +not a separate `%load_ext eea_datalakehouse.notebook.magics` line (that still +works too, e.g. from an IPython startup file that imports the module +directly rather than running a magic — see `magics.load_ipython_extension`). +Outside IPython (no shell running yet, or none at all) importing this +package is simply a no-op registration-wise. +""" + +from __future__ import annotations + +from IPython import get_ipython + + +def _autoregister() -> None: + shell = get_ipython() + if shell is None: + return # not running inside IPython (yet) — nothing to register with + if "EEALakehouseMagics" in shell.magics_manager.registry: + return # already registered — see magics.load_ipython_extension's matching guard + + from .magics import EEALakehouseMagics + + shell.register_magics(EEALakehouseMagics) + + +_autoregister() diff --git a/src/eea_datalakehouse/notebook/magics.py b/src/eea_datalakehouse/notebook/magics.py new file mode 100644 index 0000000..f139418 --- /dev/null +++ b/src/eea_datalakehouse/notebook/magics.py @@ -0,0 +1,196 @@ +"""`%catalog` and `%ingest` — thin syntactic sugar over `CatalogSession`/ +`IngestSession` (see `docs/notebook-facade-for-data-scientists.md`, "Two +magics, two sessions"). + +Available the moment this package is imported inside IPython — `import +eea_datalakehouse.notebook` (see that package's `__init__.py`) registers both +magics as a side effect, so nothing needs loading explicitly. `%load_ext +eea_datalakehouse.notebook.magics` still works too (and is the only option +outside IPython's auto-import path, e.g. a config that imports this module +directly) — `load_ipython_extension` below is a no-op if the auto-import +already registered these, so using both never double-registers or drops a +session's queued-but-not-committed state. + +In any cell, once loaded either way:: + + %catalog copy("draft.raw_2026", "bwd.reference.water_temperature") + %catalog tag("bwd.reference.water_temperature", ["reviewed"]) + %catalog commit(retry=True) + + %ingest ingest(folder="./bw_2026", target_catalog_path="bwd.reference", + data_format="parquet", table_name="water_temperature") + %ingest commit(retry=True) + +Each magic evaluates the rest of the line as a method call on ONE session +object, created lazily the first time that magic is used in this kernel and +kept for the kernel's lifetime (see the design doc's "Session context lives +in the Python process") — so calls in different cells accumulate on the same +queue, and `commit()` flushes it. `%catalog`/`%ingest` invent no vocabulary +of their own: every name after them is a real `CatalogSession`/`IngestSession` +method (see those modules for the full list) — this file only builds the +session and turns exceptions into a short printed message instead of a +traceback, per the design doc's facade plan. `commit` with no parentheses is +accepted as a convenience for `commit()`. + +`eval()` below runs exactly the Python the user typed after the magic name, +in their own notebook namespace — no new capability over what the same user +could already type directly into the next cell; that's what makes a `%magic` +different from evaluating untrusted input. + +Credentials/connection details come from the kernel environment, the same +`DREMIO_...` variables `debugger/debug_run.py` and `FolderIngest`'s default +construction already use — this module never asks for or stores a token +itself. + +`CatalogSession`'s "current path" context (a leading `.` on a path resolves +against it — see that class' `set_context`/`_resolve`) is already kept +up to date automatically just from ordinary `%catalog` use, so no data +custodian ever needs to call `set_context` themselves. On top of that, this +module also registers a Jupyter Comm target (see `_register_context_comm`) +so an integrated frontend — the JupyterLab catalog-tree extension, in the +separate `eeadata/EEALakeHouse` repo — can push a selected leaf's path into +this kernel invisibly: no cell runs, nothing a custodian could see or type, +either. See `docs/notebook-facade-for-data-scientists.md`, "Pre-filling +catalog context from a JupyterLab tree click". +""" + +from __future__ import annotations + +import os +from typing import Any + +from IPython.core.magic import Magics, line_magic, magics_class + +from ..catalog import Catalog +from ..catalog.session import CatalogSession, CatalogSessionError +from ..dds_ingestion.session import IngestSession, IngestSessionError + +_USAGE = { + "catalog": ( + '%catalog copy("a.b", "c.d") | %catalog tag("a.b", ["reviewed"]) | %catalog commit' + ), + "ingest": ( + '%ingest ingest(folder="./data", target_catalog_path="a.b", data_format="parquet")' + " | %ingest commit" + ), +} + + +def _build_catalog_session() -> CatalogSession: + base_url = os.environ.get("DREMIO_BASE_URL") + token = os.environ.get("DREMIO_TOKEN") + username = os.environ.get("DREMIO_USERNAME") + if not base_url or not token: + raise RuntimeError( + "%catalog needs DREMIO_BASE_URL and DREMIO_TOKEN set in the kernel environment" + ) + return CatalogSession(Catalog(base_url, token, username=username)) + + +def _dispatch( + session: Any, line: str, user_ns: dict[str, Any], label: str, error_type: type[Exception] +) -> Any: + line = line.strip() + if not line: + print(f"usage: {_USAGE[label]}") + return None + if line.isidentifier(): # bare "commit" (no parens) as a convenience + line = f"{line}()" + namespace = {**user_ns, "__session__": session} + try: + return eval(f"__session__.{line}", namespace) # see module docstring re: eval + except error_type as exc: + print(f"{label} error: {exc}") + return None + + +_CONTEXT_COMM_TARGET = "eea_datalakehouse.catalog_context" + + +def _register_context_comm(shell: Any, magics: EEALakehouseMagics) -> None: + """Wire up a Jupyter Comm target so an integrated frontend (the + JupyterLab catalog-tree extension, in the separate `eeadata/EEALakeHouse` + repo) can push a selected leaf's path into this kernel's CatalogSession + invisibly — no cell runs, nothing a data custodian could see or type. + + A no-op if there's no live kernel Comm manager to register against — a + plain IPython shell, or this package's own test shell, neither of which + is a real Jupyter kernel. + """ + comm_manager = getattr(getattr(shell, "kernel", None), "comm_manager", None) + if comm_manager is None: + return + + def _on_comm_open(comm: Any, open_msg: dict[str, Any]) -> None: + @comm.on_msg + def _on_msg(msg: dict[str, Any]) -> None: + path = msg.get("content", {}).get("data", {}).get("path") + if isinstance(path, str): + magics._apply_context(path) + + comm_manager.register_target(_CONTEXT_COMM_TARGET, _on_comm_open) + + +@magics_class +class EEALakehouseMagics(Magics): + """Registers `%catalog` and `%ingest` — see this module's docstring.""" + + def __init__(self, shell: Any) -> None: + super().__init__(shell) + self._catalog_session: CatalogSession | None = None + self._ingest_session: IngestSession | None = None + self._pending_context: str | None = None + _register_context_comm(shell, self) + + def _apply_context(self, path: str) -> None: + """Set `path` as the CatalogSession's context. Called only by the + Comm handler in `_register_context_comm` — never by a data + custodian directly, and never dispatched through `%catalog`. + + Builds the session now if one doesn't exist yet, so a context + pushed before any `%catalog` cell has run still takes effect. If + credentials aren't available yet either, remembers `path` instead + of failing — applied the moment `%catalog` next builds a session + for real, rather than printing an error outside of any cell a + custodian actually ran. + """ + if self._catalog_session is None: + try: + self._catalog_session = _build_catalog_session() + except RuntimeError: + self._pending_context = path + return + self._catalog_session.set_context(path) + + @line_magic + def catalog(self, line: str) -> Any: + if self._catalog_session is None: + try: + self._catalog_session = _build_catalog_session() + except RuntimeError as exc: + print(f"catalog error: {exc}") + return None + if self._pending_context is not None: + self._catalog_session.set_context(self._pending_context) + self._pending_context = None + return _dispatch( + self._catalog_session, line, self.shell.user_ns, "catalog", CatalogSessionError + ) + + @line_magic + def ingest(self, line: str) -> Any: + if self._ingest_session is None: + self._ingest_session = IngestSession() + return _dispatch( + self._ingest_session, line, self.shell.user_ns, "ingest", IngestSessionError + ) + + +def load_ipython_extension(ipython: Any) -> None: + # A no-op if `eea_datalakehouse.notebook`'s own auto-import already + # registered this (see the module docstring) — registering again would + # replace the live instance with a fresh one, silently dropping any + # already-queued-but-not-committed CatalogSession/IngestSession state. + if "EEALakehouseMagics" in ipython.magics_manager.registry: + return + ipython.register_magics(EEALakehouseMagics) diff --git a/tests/catalog/test_session.py b/tests/catalog/test_session.py new file mode 100644 index 0000000..56aeb7b --- /dev/null +++ b/tests/catalog/test_session.py @@ -0,0 +1,141 @@ +"""CatalogSession: queue catalog verbs, commit as one all-or-nothing batch. + +See src/eea_datalakehouse/catalog/session.py and +docs/notebook-facade-for-data-scientists.md ("Queue calls, then commit"). +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from eea_datalakehouse.catalog import retry_state +from eea_datalakehouse.catalog.client import Catalog +from eea_datalakehouse.catalog.errors import EngineStartingError +from eea_datalakehouse.catalog.session import CatalogCommitError, CatalogSession + +from .conftest import FakeCatalogRest, FakeExecutor + +BASE_URL = "https://dremio.example.test" + + +@pytest.fixture(autouse=True) +def _isolated_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(retry_state, "DEFAULT_STATE_PATH", tmp_path / "state.json") + + +def _catalog(rest: FakeCatalogRest, executor: FakeExecutor | None = None) -> Catalog: + executor = executor or FakeExecutor() + # datacopy/datamove always go over flight_executor — share one FakeExecutor + # so a test can reason about one single call log regardless of which verb + # made the call. + return Catalog(BASE_URL, "pat", executor=executor, flight_executor=executor, catalog_rest=rest) + + +def test_nothing_runs_until_commit() -> None: + rest = FakeCatalogRest(existing={"a", "bwd"}) + session = CatalogSession(_catalog(rest)) + + session.create_folder("bwd.new") + + assert rest.created == [] + assert repr(session) == "CatalogSession(pending=1)" + + +def test_commit_runs_queued_steps_in_order_and_clears_the_queue() -> None: + rest = FakeCatalogRest(existing={"a", "bwd", "bwd.table1"}) + session = CatalogSession(_catalog(rest)) + + session.create_folder("bwd.newfolder").tag("bwd.table1", ["reviewed"]) + report = session.commit() + + assert report.succeeded == [ + "create folder 'bwd.newfolder'", + "tag 'bwd.table1' with ['reviewed']", + ] + assert rest.created == ["bwd.newfolder"] + assert rest.get_tags("bwd.table1") == ["reviewed"] + assert repr(session) == "CatalogSession(pending=0)" # queue cleared + + +def test_commit_rolls_back_a_cleanly_reversible_batch_on_failure() -> None: + rest = FakeCatalogRest(existing={"a", "bwd", "bwd.table1"}) + executor = FakeExecutor(rows=[{"TABLE_NAME": "table1", "TABLE_TYPE": "TABLE"}]) + session = CatalogSession(_catalog(rest, executor)) + + # step 1 succeeds and is reversible; step 2 fails because the target + # already exists and overwrite defaults to False. + session.create_folder("bwd.newfolder") + session.copy("bwd.table1", "bwd.table1") + + with pytest.raises(CatalogCommitError) as exc_info: + session.commit() + + error = exc_info.value + assert error.rolled_back is True + assert error.unresolved == [] + assert error.failed_step == "copy 'bwd.table1' -> 'bwd.table1'" + assert rest.created == ["bwd.newfolder"] + assert rest.deleted == ["bwd.newfolder"] # step 1's undo ran + assert repr(session) == "CatalogSession(pending=0)" # a failed commit still clears the queue + + +def test_commit_reports_what_it_could_not_undo() -> None: + rest = FakeCatalogRest(existing={"a", "bwd", "bwd.table1"}) + executor = FakeExecutor(rows=[{"TABLE_NAME": "table1", "TABLE_TYPE": "TABLE"}]) + session = CatalogSession(_catalog(rest, executor)) + + session.create_folder("bwd.newfolder") # reversible + session.copy("bwd.table1", "bwd.table1", overwrite=True) # succeeds, but NOT reversible + session.tag("bwd.missing", ["x"]) # fails: path does not exist + + with pytest.raises(CatalogCommitError) as exc_info: + session.commit() + + error = exc_info.value + assert error.rolled_back is False + assert len(error.unresolved) == 1 + assert "overwrote an existing target" in error.unresolved[0] + # the reversible step before the irreversible one is still cleaned up + assert rest.deleted == ["bwd.newfolder"] + + +def test_retry_reruns_the_same_step_after_engine_starting_error() -> None: + class _FlakyRest(FakeCatalogRest): + def __init__(self, *args: object, **kwargs: object) -> None: + super().__init__(*args, **kwargs) # type: ignore[arg-type] + self._create_folder_attempts = 0 + + def create_folder(self, path: str) -> bool: + self._create_folder_attempts += 1 + if self._create_folder_attempts == 1: + raise EngineStartingError("engine starting", idempotency_key="k") + return super().create_folder(path) + + rest = _FlakyRest(existing={"a", "bwd"}) + session = CatalogSession(_catalog(rest)) + + session.create_folder("bwd.new") + report = session.commit(retry=True, retry_delay=0) + + assert report.succeeded == ["create folder 'bwd.new'"] + assert rest.created == ["bwd.new"] + assert rest._create_folder_attempts == 2 + + +def test_engine_starting_error_without_retry_rolls_back_immediately() -> None: + class _AlwaysStartingRest(FakeCatalogRest): + def create_folder(self, path: str) -> bool: + raise EngineStartingError("engine starting", idempotency_key="k") + + rest = _AlwaysStartingRest(existing={"a", "bwd"}) + session = CatalogSession(_catalog(rest)) + + session.create_folder("bwd.new") + + with pytest.raises(CatalogCommitError) as exc_info: + session.commit() # retry defaults to False + + assert isinstance(exc_info.value.original_error, EngineStartingError) + assert exc_info.value.rolled_back is True diff --git a/tests/catalog/test_session_context.py b/tests/catalog/test_session_context.py new file mode 100644 index 0000000..154d652 --- /dev/null +++ b/tests/catalog/test_session_context.py @@ -0,0 +1,109 @@ +"""CatalogSession's "current path" context: relative paths (a leading `.`), +auto-inferred from usage, `set_context()` as an explicit seed. + +See src/eea_datalakehouse/catalog/session.py's `_resolve`/`_resolve_path` +and docs/notebook-facade-for-data-scientists.md ("Pre-filling catalog +context"). +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from eea_datalakehouse.catalog import retry_state +from eea_datalakehouse.catalog.client import Catalog +from eea_datalakehouse.catalog.session import CatalogSession, CatalogSessionError + +from .conftest import FakeCatalogRest, FakeExecutor + +BASE_URL = "https://dremio.example.test" + + +@pytest.fixture(autouse=True) +def _isolated_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(retry_state, "DEFAULT_STATE_PATH", tmp_path / "state.json") + + +def _catalog(rest: FakeCatalogRest, executor: FakeExecutor | None = None) -> Catalog: + executor = executor or FakeExecutor() + return Catalog(BASE_URL, "pat", executor=executor, flight_executor=executor, catalog_rest=rest) + + +def test_relative_path_without_any_context_raises() -> None: + rest = FakeCatalogRest(existing={"a", "bwd", "bwd.table1"}) + session = CatalogSession(_catalog(rest)) + + with pytest.raises(CatalogSessionError, match="no context is set"): + session.tag(".table1", ["reviewed"]) + + +def test_touching_a_leaf_infers_context_for_a_later_relative_call() -> None: + rest = FakeCatalogRest( + existing={"a", "bwd", "bwd.reference.water_temperature", "bwd.reference.stations"} + ) + session = CatalogSession(_catalog(rest)) + + session.tag("bwd.reference.water_temperature", ["reviewed"]) # sets context to bwd.reference + session.tag(".stations", ["reviewed"]) # resolves to bwd.reference.stations + + report = session.commit() + + assert report.succeeded == [ + "tag 'bwd.reference.water_temperature' with ['reviewed']", + "tag 'bwd.reference.stations' with ['reviewed']", + ] + assert rest.get_tags("bwd.reference.stations") == ["reviewed"] + + +def test_set_context_seeds_it_explicitly_before_any_path_is_used() -> None: + rest = FakeCatalogRest(existing={"a", "bwd", "bwd.reference.water_temperature"}) + session = CatalogSession(_catalog(rest)) + + session.set_context("bwd.reference") + session.tag(".water_temperature", ["reviewed"]) + + report = session.commit() + + assert report.succeeded == ["tag 'bwd.reference.water_temperature' with ['reviewed']"] + + +def test_create_folder_context_becomes_the_folder_itself_not_its_parent() -> None: + rest = FakeCatalogRest(existing={"a", "bwd", "bwd.reference.water_temperature"}) + session = CatalogSession(_catalog(rest)) + + session.create_folder("bwd.reference.2027", create_parents=True) # context -> that folder + session.set_context("bwd.reference") # re-seed for the rest of this test + session.tag(".water_temperature", ["reviewed"]) + + report = session.commit() + + assert report.succeeded == [ + "create folder 'bwd.reference.2027'", + "tag 'bwd.reference.water_temperature' with ['reviewed']", + ] + + +def test_copy_resolves_both_paths_against_the_same_starting_context() -> None: + rest = FakeCatalogRest( + existing={"a", "bwd", "bwd.draft.raw_2026", "bwd.reference", "bwd.reference.stations"} + ) + executor = FakeExecutor( + rows_sequence=[ + [{"TABLE_NAME": "raw_2026"}], # source exists + [], # target does not exist yet + ] + ) + session = CatalogSession(_catalog(rest, executor)) + + session.set_context("bwd.draft") + session.copy(".raw_2026", "bwd.reference.water_temperature") # target does NOT use bwd.draft + session.tag(".stations", ["reviewed"]) # resolves against bwd.reference (the target's parent) + + report = session.commit() + + assert report.succeeded == [ + "copy 'bwd.draft.raw_2026' -> 'bwd.reference.water_temperature'", + "tag 'bwd.reference.stations' with ['reviewed']", + ] diff --git a/tests/dds_ingestion/test_ingest_session.py b/tests/dds_ingestion/test_ingest_session.py new file mode 100644 index 0000000..1995248 --- /dev/null +++ b/tests/dds_ingestion/test_ingest_session.py @@ -0,0 +1,168 @@ +"""IngestSession: queue folder ingests, run them as one batch, no rollback. + +See src/eea_datalakehouse/dds_ingestion/session.py and +docs/notebook-facade-for-data-scientists.md ("Two sessions, not one"). +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from eea_datalakehouse.dds_ingestion.models import ( + BeginResult, + CommitResult, + S3Plan, + StatusResult, + UploadTarget, +) +from eea_datalakehouse.dds_ingestion.session import IngestCommitError, IngestSession + + +def _make_folder(base: Path, name: str) -> Path: + folder = base / name + folder.mkdir() + (folder / "a.parquet").write_bytes(b"PAR1-a") + return folder + + +class _StubClient: + """Fake IngestClient shared across every queued ingest in a test.""" + + def __init__(self) -> None: + self.begin_calls = 0 + self.commit_calls: list[str] = [] + self.retry_calls: list[str] = [] + self.closed = False + self._next_session = 0 + + def begin(self, *, files: list[Any], **kwargs: Any) -> BeginResult: + self.begin_calls += 1 + 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 upload_file(self, target: UploadTarget, data: bytes) -> str | None: + 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=1) + + def close(self) -> None: + self.closed = True + + +def test_nothing_runs_until_commit(tmp_path: Path) -> None: + client = _StubClient() + session = IngestSession(client=client) # type: ignore[arg-type] + + session.ingest( + _make_folder(tmp_path, "a"), "bio.uploads", data_format="parquet", show_progress=False + ) + + assert client.begin_calls == 0 + assert repr(session) == "IngestSession(pending=1)" + + +def test_commit_runs_every_queued_ingest_and_clears_the_queue(tmp_path: Path) -> None: + client = _StubClient() + session = IngestSession(client=client) # type: ignore[arg-type] + session.ingest( + _make_folder(tmp_path, "a"), "bio.uploads.a", data_format="parquet", show_progress=False + ) + session.ingest( + _make_folder(tmp_path, "b"), "bio.uploads.b", data_format="parquet", show_progress=False + ) + + report = session.commit() + + assert len(report.outcomes) == 2 + assert client.begin_calls == 2 + assert client.commit_calls == ["sess-1", "sess-2"] + assert repr(session) == "IngestSession(pending=0)" + + +def test_commit_stops_at_the_first_failure_and_reports_what_already_landed(tmp_path: Path) -> None: + class _FailsOnSecondBegin(_StubClient): + def begin(self, *, files: list[Any], **kwargs: Any) -> BeginResult: + if self.begin_calls == 1: + raise RuntimeError("target catalog folder does not exist") + return super().begin(files=files, **kwargs) + + client = _FailsOnSecondBegin() + session = IngestSession(client=client) # type: ignore[arg-type] + session.ingest( + _make_folder(tmp_path, "a"), "bio.uploads.a", data_format="parquet", show_progress=False + ) + session.ingest( + _make_folder(tmp_path, "b"), "bio.uploads.b", data_format="parquet", show_progress=False + ) + + with pytest.raises(IngestCommitError) as exc_info: + session.commit() + + error = exc_info.value + assert error.failed_index == 1 + assert len(error.succeeded) == 1 + assert error.succeeded[0].commit.table_path == "t" + assert "1 earlier ingest(s)" in str(error) + assert repr(session) == "IngestSession(pending=0)" # queue cleared even on failure + + +def test_retry_resumes_a_failed_commit_before_giving_up(tmp_path: Path) -> None: + class _FlakyClient(_StubClient): + def __init__(self) -> None: + super().__init__() + self._commit_attempts = 0 + + def commit(self, *, session_id: str, **kwargs: Any) -> CommitResult: + self._commit_attempts += 1 + if self._commit_attempts == 1: + raise RuntimeError("Dremio load failed: engine was restarting") + return super().commit(session_id=session_id, **kwargs) + + def get_status(self, session_id: str) -> StatusResult: + return StatusResult.from_json( + { + "session_id": session_id, + "status": "failed", + "failed_stage": "load", + "resumable": True, + } + ) + + 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": 3, + } + ) + + client = _FlakyClient() + session = IngestSession(client=client) # type: ignore[arg-type] + session.ingest( + _make_folder(tmp_path, "a"), "bio.uploads", data_format="parquet", show_progress=False + ) + + report = session.commit(retry=True) + + assert len(report.outcomes) == 1 + assert report.outcomes[0].resumed is True + assert client.retry_calls == ["sess-1"] diff --git a/tests/notebook/__init__.py b/tests/notebook/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/notebook/conftest.py b/tests/notebook/conftest.py new file mode 100644 index 0000000..c814d9d --- /dev/null +++ b/tests/notebook/conftest.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from typing import Any + +import pytest +from IPython.testing import globalipapp + + +@pytest.fixture(scope="session") +def shell() -> Any: + """The one process-wide IPython test shell. + + `globalipapp.start_ipython()` only actually starts (and returns) a shell + on its very first call anywhere in this process — every later call + returns `None` (see its own "should only ever run once" guard) — so every + test in this package shares this one fixture instead of each starting + their own and getting `None`. + """ + return globalipapp.start_ipython() diff --git a/tests/notebook/test_autoregister.py b/tests/notebook/test_autoregister.py new file mode 100644 index 0000000..4599f22 --- /dev/null +++ b/tests/notebook/test_autoregister.py @@ -0,0 +1,64 @@ +"""Registering `%catalog`/`%ingest` as an import side effect. + +See `eea_datalakehouse/notebook/__init__.py`'s `_autoregister` and the "no +`%load_ext` line needed" behaviour documented in `magics.py`'s module +docstring. + +A literal fresh `import eea_datalakehouse.notebook` can't be re-exercised +inside one test process — Python caches the module after its first import, +which happens at test-collection time, before any IPython shell exists here +— so these tests call `_autoregister()` directly instead. That function +*is* the logic a fresh import runs; calling it again is just re-invoking it +on demand, not a different code path. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +import eea_datalakehouse.notebook as notebook_pkg +from eea_datalakehouse.notebook import magics as magics_module +from eea_datalakehouse.notebook.magics import EEALakehouseMagics + + +def test_autoregister_is_a_noop_outside_ipython(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(notebook_pkg, "get_ipython", lambda: None) + + notebook_pkg._autoregister() # must not raise with no shell to register against + + +def test_autoregister_registers_against_a_running_shell(shell: Any) -> None: + shell.magics_manager.registry.pop("EEALakehouseMagics", None) + + notebook_pkg._autoregister() + + assert isinstance(shell.magics_manager.registry["EEALakehouseMagics"], EEALakehouseMagics) + + +def test_autoregister_does_not_replace_an_already_registered_instance(shell: Any) -> None: + shell.magics_manager.registry.pop("EEALakehouseMagics", None) + notebook_pkg._autoregister() + first = shell.magics_manager.registry["EEALakehouseMagics"] + first._catalog_session = "sentinel" # stands in for real queued session state + + notebook_pkg._autoregister() # calling it again must not clobber the above + + assert shell.magics_manager.registry["EEALakehouseMagics"] is first + assert first._catalog_session == "sentinel" + + +def test_load_ext_after_autoregister_does_not_replace_it_either(shell: Any) -> None: + # The two entry points (auto-import, explicit `%load_ext`) have to agree + # with each other too, not just with themselves — using both in either + # order must still never drop a session's already-queued state. + shell.magics_manager.registry.pop("EEALakehouseMagics", None) + notebook_pkg._autoregister() + first = shell.magics_manager.registry["EEALakehouseMagics"] + first._ingest_session = "sentinel" + + magics_module.load_ipython_extension(shell) + + assert shell.magics_manager.registry["EEALakehouseMagics"] is first + assert first._ingest_session == "sentinel" diff --git a/tests/notebook/test_context_comm.py b/tests/notebook/test_context_comm.py new file mode 100644 index 0000000..45bb2aa --- /dev/null +++ b/tests/notebook/test_context_comm.py @@ -0,0 +1,161 @@ +"""Pushing catalog context into a kernel invisibly, via a Jupyter Comm. + +See `eea_datalakehouse/notebook/magics.py`'s `_register_context_comm`/ +`_apply_context` and `docs/notebook-facade-for-data-scientists.md`, +"Pre-filling catalog context from a JupyterLab tree click". +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from eea_datalakehouse.notebook import magics as magics_module +from eea_datalakehouse.notebook.magics import EEALakehouseMagics, _register_context_comm + + +class _FakeCommManager: + """Records `register_target` calls instead of touching a real kernel.""" + + def __init__(self) -> None: + self.targets: dict[str, Any] = {} + + def register_target(self, name: str, handler: Any) -> None: + self.targets[name] = handler + + +class _FakeKernel: + def __init__(self, comm_manager: _FakeCommManager) -> None: + self.comm_manager = comm_manager + + +class _FakeComm: + """Records the `on_msg` handler a comm_open callback installs.""" + + def __init__(self) -> None: + self._on_msg: Any = None + + def on_msg(self, handler: Any) -> Any: + self._on_msg = handler + return handler + + def deliver(self, path: object) -> None: + assert self._on_msg is not None, "comm_open never called comm.on_msg" + self._on_msg({"content": {"data": {"path": path}}}) + + +def _magics_without_shell_init(shell: Any) -> EEALakehouseMagics: + # EEALakehouseMagics.__init__ already calls _register_context_comm — build + # one directly rather than via shell.register_magics so these tests can + # supply a fake kernel without needing a real Jupyter one. + return EEALakehouseMagics(shell=shell) + + +def test_register_context_comm_is_a_noop_without_a_real_kernel() -> None: + class _ShellWithNoKernel: + pass + + # Must not raise even though there's nothing to register a Comm target + # with — a plain IPython shell (or this package's own test shell) has no + # `.kernel` at all. + _register_context_comm(_ShellWithNoKernel(), magics=object()) # type: ignore[arg-type] + + +def test_comm_message_reaches_apply_context(shell: Any, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[str] = [] + monkeypatch.setattr(EEALakehouseMagics, "_apply_context", lambda self, path: calls.append(path)) + comm_manager = _FakeCommManager() + shell.kernel = _FakeKernel(comm_manager) + try: + _magics_without_shell_init(shell) + comm_open = comm_manager.targets["eea_datalakehouse.catalog_context"] + comm = _FakeComm() + comm_open(comm, {"content": {"data": {}}}) + + comm.deliver("bwd.reference") + + assert calls == ["bwd.reference"] + finally: + del shell.kernel + + +def test_comm_message_with_a_non_string_path_is_ignored( + shell: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + calls: list[str] = [] + monkeypatch.setattr(EEALakehouseMagics, "_apply_context", lambda self, path: calls.append(path)) + comm_manager = _FakeCommManager() + shell.kernel = _FakeKernel(comm_manager) + try: + _magics_without_shell_init(shell) + comm_open = comm_manager.targets["eea_datalakehouse.catalog_context"] + comm = _FakeComm() + comm_open(comm, {"content": {"data": {}}}) + + comm.deliver(None) + + assert calls == [] + finally: + del shell.kernel + + +class _FakeCatalogSession: + def __init__(self) -> None: + self.context: str | None = None + + def set_context(self, path: str) -> None: + self.context = path + + def commit(self, **kwargs: Any) -> str: + return "committed" + + +def test_apply_context_sets_it_on_an_existing_session() -> None: + magics = EEALakehouseMagics.__new__(EEALakehouseMagics) + fake_session = _FakeCatalogSession() + magics._catalog_session = fake_session # type: ignore[assignment] + magics._pending_context = None + + magics._apply_context("bwd.reference") + + assert fake_session.context == "bwd.reference" + + +def test_apply_context_defers_when_no_session_and_no_credentials( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + magics_module, + "_build_catalog_session", + lambda: (_ for _ in ()).throw(RuntimeError("no creds")), + ) + magics = EEALakehouseMagics.__new__(EEALakehouseMagics) + magics._catalog_session = None + magics._pending_context = None + + magics._apply_context("bwd.reference") + + assert magics._catalog_session is None + assert magics._pending_context == "bwd.reference" + + +def test_catalog_magic_applies_a_pending_context_once_it_builds_a_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_session = _FakeCatalogSession() + monkeypatch.setattr(magics_module, "_build_catalog_session", lambda: fake_session) + magics = EEALakehouseMagics.__new__(EEALakehouseMagics) + magics._catalog_session = None + magics._ingest_session = None + magics._pending_context = "bwd.reference" + + class _FakeShell: + user_ns: dict[str, Any] = {} + + magics.shell = _FakeShell() # type: ignore[attr-defined] + + magics.catalog("commit") + + assert fake_session.context == "bwd.reference" + assert magics._pending_context is None diff --git a/tests/notebook/test_magics.py b/tests/notebook/test_magics.py new file mode 100644 index 0000000..8ec94c4 --- /dev/null +++ b/tests/notebook/test_magics.py @@ -0,0 +1,136 @@ +"""`%catalog`/`%ingest` — dispatch onto a per-kernel session, friendly errors. + +Uses IPython's own test shell (`IPython.testing.globalipapp`) rather than a +real kernel — the magics only need `self.shell.user_ns`, which a test shell +provides just as well. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from eea_datalakehouse.catalog.session import CatalogCommitError +from eea_datalakehouse.dds_ingestion.session import IngestSession +from eea_datalakehouse.notebook import magics as magics_module +from eea_datalakehouse.notebook.magics import EEALakehouseMagics + + +class _FakeCatalogSession: + """Stands in for a real `CatalogSession` — records calls, never touches + a real Catalog.""" + + def __init__(self) -> None: + self.calls: list[str] = [] + self.committed = False + + def copy(self, *args: Any, **kwargs: Any) -> _FakeCatalogSession: + self.calls.append(f"copy{args!r}{kwargs!r}") + return self + + def commit(self, **kwargs: Any) -> str: + self.committed = True + return "committed" + + def raise_commit_error(self) -> None: + raise CatalogCommitError( + "boom", + failed_step="copy", + original_error=RuntimeError("x"), + rolled_back=True, + unresolved=[], + ) + + +@pytest.fixture +def ip(shell: Any) -> Any: + # Fresh session state per test, so one test's queued/committed session + # never leaks into the next — same shell and magics instance throughout + # (registering again is a no-op if it's already there, see + # magics.load_ipython_extension's guard). + shell.register_magics(EEALakehouseMagics) + instance = _magics_instance(shell) + instance._catalog_session = None + instance._ingest_session = None + return shell + + +def _magics_instance(ip: Any) -> EEALakehouseMagics: + return ip.magics_manager.registry["EEALakehouseMagics"] # type: ignore[no-any-return] + + +def test_catalog_magic_builds_the_session_once_and_reuses_it( + ip: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + fake = _FakeCatalogSession() + build_calls = [] + monkeypatch.setattr( + magics_module, "_build_catalog_session", lambda: (build_calls.append(1), fake)[1] + ) + + ip.run_line_magic("catalog", 'copy("a.b", "c.d", overwrite=True)') + ip.run_line_magic("catalog", "commit") + + assert build_calls == [1] # only built once, reused across calls + assert fake.calls == ["copy('a.b', 'c.d'){'overwrite': True}"] + assert fake.committed is True + + +def test_bare_commit_without_parens_is_accepted(ip: Any, monkeypatch: pytest.MonkeyPatch) -> None: + fake = _FakeCatalogSession() + monkeypatch.setattr(magics_module, "_build_catalog_session", lambda: fake) + + ip.run_line_magic("catalog", "commit") + + assert fake.committed is True + + +def test_catalog_session_error_is_printed_not_raised( + ip: Any, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + fake = _FakeCatalogSession() + monkeypatch.setattr(magics_module, "_build_catalog_session", lambda: fake) + + ip.run_line_magic("catalog", "raise_commit_error()") # must not raise out of the magic + + out = capsys.readouterr().out + assert "catalog error:" in out + assert "boom" in out + + +def test_missing_credentials_prints_a_friendly_message( + ip: Any, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.delenv("DREMIO_BASE_URL", raising=False) + monkeypatch.delenv("DREMIO_TOKEN", raising=False) + # Force a fresh build attempt even if an earlier test in this module left + # a session cached on a *different* Magics instance — this test's `ip` + # fixture registers its own. + + ip.run_line_magic("catalog", 'copy("a.b", "c.d")') + + out = capsys.readouterr().out + assert "catalog error:" in out + assert "DREMIO_BASE_URL" in out + + +def test_ingest_magic_queues_onto_one_shared_session(ip: Any) -> None: + ip.run_line_magic( + "ingest", + 'ingest(folder="/tmp/does-not-matter", target_catalog_path="a.b", ' + 'data_format="parquet", show_progress=False)', + ) + + session = _magics_instance(ip)._ingest_session + assert isinstance(session, IngestSession) + assert repr(session) == "IngestSession(pending=1)" + + +def test_empty_line_prints_usage(ip: Any, capsys: pytest.CaptureFixture[str]) -> None: + # No credentials needed to reach the usage message — it's printed before + # a session is ever built. + ip.run_line_magic("ingest", "") + + out = capsys.readouterr().out + assert "usage:" in out