From 28483da6b676f18f3ebf245aaf0105f82970e914 Mon Sep 17 00:00:00 2001 From: Petr Date: Wed, 22 Jul 2026 00:18:59 +0200 Subject: [PATCH 1/6] feat(storage): table snapshots -- create/list/detail/delete + table-from-snapshot (0.75.0, #512) Issue #512: kbagent could not create a table from an existing snapshot. Adds the full snapshot lifecycle as first-class storage commands: - storage snapshot-create -- point-in-time backup (data + columns + PK); async tableSnapshotCreate job, receipt carries snapshot_id (write) - storage snapshots -- list a table's snapshots (read) - storage snapshot-detail -- global snapshot ID -> embeds source table (read) - storage snapshot-delete -- batch-tolerant, forecloses restores (destructive) - storage table-from-snapshot -- create a NEW table from a snapshot (write) Endpoint contract verified against the reference PHP client (keboola/storage-api-php-client) and live against connection.us-east4.gcp.keboola.com (project 5946, 2026-07-22): - Restore goes through the classic tables-async endpoint with a snapshotId source; tables-definition (used by create-table) does not accept snapshots -- hence a dedicated command, not a create-table flag. - --name is REQUIRED: the live API rejects an omitted/empty name; the PHP client's 'defaults to snapshot name' docblock is stale. - No overwrite semantics: restoring onto an existing table name fails; restore under a new name, verify, then swap-tables/delete-table. New modules (storage.py/storage_service.py are past their file-size ceilings): services/snapshot_service.py, commands/_storage_snapshots.py (mounted flat onto storage_app so permission keys stay storage.*). Tests: 30 unit tests across three layers (HTTP shape via pytest_httpx, service logic, CliRunner) + E2E step 11.2 snapshot roundtrip in tests/test_e2e.py (create -> list -> detail -> dry-run -> restore -> verify rows/columns/PK -> cleanup). Docs: context.py, CLAUDE.md, SKILL.md (regenerated), commands-reference.md, gotchas.md (since v0.75.0), keboola-expert.md matrix rows (58.7KB, under the 62KB gate). --- .claude-plugin/marketplace.json | 2 +- CLAUDE.md | 8 + plugins/kbagent/.claude-plugin/plugin.json | 2 +- plugins/kbagent/agents/keboola-expert.md | 2 + plugins/kbagent/skills/kbagent/SKILL.md | 5 + .../kbagent/references/commands-reference.md | 5 + .../skills/kbagent/references/gotchas.md | 23 + pyproject.toml | 2 +- src/keboola_agent_cli/changelog.py | 16 + src/keboola_agent_cli/cli.py | 3 + src/keboola_agent_cli/client.py | 120 +++ .../commands/_storage_snapshots.py | 299 ++++++++ src/keboola_agent_cli/commands/context.py | 25 + src/keboola_agent_cli/commands/storage.py | 8 + src/keboola_agent_cli/permissions.py | 7 + .../services/snapshot_service.py | 247 ++++++ tests/test_e2e.py | 163 ++++ tests/test_storage_snapshots.py | 708 ++++++++++++++++++ uv.lock | 2 +- 19 files changed, 1643 insertions(+), 4 deletions(-) create mode 100644 src/keboola_agent_cli/commands/_storage_snapshots.py create mode 100644 src/keboola_agent_cli/services/snapshot_service.py create mode 100644 tests/test_storage_snapshots.py diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 747e2deb..aa1c97b4 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -10,7 +10,7 @@ "plugins": [ { "name": "kbagent", - "version": "0.74.0", + "version": "0.75.0", "source": "./plugins/kbagent", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "category": "development" diff --git a/CLAUDE.md b/CLAUDE.md index 59d1ef65..6f238000 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -344,6 +344,14 @@ kbagent storage delete-column --project NAME --table-id ID --column COL [--colum kbagent storage delete-bucket --project NAME --bucket-id ID [--bucket-id ...] [--force] [--dry-run] [--yes] [--branch ID] kbagent storage swap-tables --project NAME --table-id ID --target-table-id ID --branch ID [--dry-run] [--yes] kbagent storage clone-table --project NAME --table-id ID --branch ID [--dry-run] +kbagent storage snapshot-create --project NAME --table-id ID [--description D] [--branch ID] +kbagent storage snapshots --project NAME --table-id ID [--limit N] [--branch ID] +kbagent storage snapshot-detail --project NAME --snapshot-id ID +kbagent storage snapshot-delete --project NAME --snapshot-id ID [--snapshot-id ...] [--dry-run] [--yes] +kbagent storage table-from-snapshot --project NAME --snapshot-id ID --bucket-id ID --name NAME [--branch ID] [--dry-run] +# Snapshots (0.75.0+, #512): point-in-time table backup (data+columns+PK) and restore as a NEW table. +# table-from-snapshot goes through the classic tables-async endpoint (NOT tables-definition), --name is +# REQUIRED (API rejects empty), no overwrite -- restore under a new name, then swap/delete yourself. kbagent storage describe-bucket --project NAME --bucket-id ID [--text STR | --file PATH | --stdin] [--branch ID] kbagent storage describe-table --project NAME --table-id ID [--text STR | --file PATH | --stdin] [--branch ID] kbagent storage describe-column --project NAME --table-id ID --column NAME=DESC [--column ...] [--branch ID] diff --git a/plugins/kbagent/.claude-plugin/plugin.json b/plugins/kbagent/.claude-plugin/plugin.json index 471b5329..98fd502c 100644 --- a/plugins/kbagent/.claude-plugin/plugin.json +++ b/plugins/kbagent/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "kbagent", - "version": "0.74.0", + "version": "0.75.0", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "author": { "name": "Keboola", diff --git a/plugins/kbagent/agents/keboola-expert.md b/plugins/kbagent/agents/keboola-expert.md index 1d95a5e0..79960f80 100644 --- a/plugins/kbagent/agents/keboola-expert.md +++ b/plugins/kbagent/agents/keboola-expert.md @@ -104,6 +104,8 @@ a critical failure. | Promote typed rebuild back into the original name | `kbagent storage swap-tables --project P --table-id in.c-foo.data --target-table-id in.c-foo.data_change_log --branch --yes` (0.28.0+) -- async storage job (`tableSwap`); client polls to completion. Service refuses without a branch; any branch incl. prod | -- | renaming or deleting + re-uploading (loses history; downstream configs need to be rewritten) | | Repartition / recluster a populated BigQuery table | `kbagent storage create-table --project P --bucket-id in.c-main --name events_repart --source-table-id in.c-main.events --time-partitioning-type DAY --time-partitioning-field created_at --clustering-field tenant_id --primary-key id` (0.66.0+, BigQuery only) to copy the data into the new layout, then `kbagent storage swap-tables --table-id in.c-main.events --target-table-id in.c-main.events_repart --branch --yes` to flip it into place. `--source-table-id` derives the schema from the source so `--column` is forbidden (mutually exclusive); a non-BigQuery project fails fast (pre-flight backend check, exit 2). Range partitioning instead: `--range-partitioning-field/-start/-end/-interval` (all four; bounds are strings; mutually exclusive with time partitioning) | -- | raw `POST /v2/storage/buckets/.../tables-definition` with a `source` object, then manual swap; or a `CREATE TABLE ... AS SELECT` in a workspace (drops NOT NULL + primary key) | | Re-seed a table without losing its schema / PK / dependents | `kbagent storage truncate-table --project P --table-id in.c-foo.data [--branch ID] [--dry-run] [--yes]` (0.32.0+) -- DELETE `/tables/{id}/rows?allowTruncate=1`; endpoint is uniformly async on every branch (returns a queued `tableRowsDelete` job; client polls via `_wait_for_storage_job`). Do NOT pass `async=true` -- the API rejects it. Batch via repeated `--table-id`. Returns `{truncated[], failed[], dry_run, project_alias}` with `truncated[]` entries carrying `{table_id, rows_before, rows_after, branch_id}`. Permission class: `destructive` | `tool call delete_table_rows` if the upstream MCP exposes it | drop + recreate the table (loses descriptions, PK, sharing edges, and breaks every downstream config reference); deleting rows via raw SQL in a workspace (bypasses the Storage API audit trail) | +| Point-in-time backup of a table before a risky change | `kbagent storage snapshot-create --project P --table-id in.c-foo.data [--description D]` (0.75.0+, #512) -- async `tableSnapshotCreate` job; receipt carries `snapshot_id`. List with `storage snapshots --table-id ...`, inspect with `snapshot-detail --snapshot-id ID` (embeds the source table object) | -- | exporting to CSV as a "backup" (loses column types + PK); relying on the destructive command's `--dry-run` alone | +| Restore a snapshot as a NEW table | `kbagent storage table-from-snapshot --project P --snapshot-id ID --bucket-id in.c-foo --name restored [--dry-run]` (0.75.0+, #512) -- classic `tables-async` endpoint; restores data + columns + PK. `--name` REQUIRED (API rejects empty; PHP-client "defaults to snapshot name" docblock is stale). No overwrite: restore under a new name, verify, then `swap-tables`/`delete-table` to promote | -- | trying `create-table --snapshot-id` (not a thing -- `tables-definition` does not accept snapshots); restoring straight onto the production name (duplicate-name error) | | Debug a failed job | `kbagent job detail --project P --job-id J --json` + `kbagent job run ... --log-tail-lines 200` | `kbagent workspace from-transformation` for SQL repro | "I think the issue is..." without reading logs | | Ad-hoc SQL / row-count / type audit | `kbagent workspace create` + `kbagent workspace load` + `kbagent workspace query --sql "..."` (0.59.0+: results come back inline+fast but **capped at `--limit`, default 500** -- check `statements[].truncated`/`total_rows`, use `COUNT(*)` for counts, `--full` for the complete set) | `kbagent workspace from-transformation` for existing transform debugging; `workspace list --qs-compatible` (0.42.0+, #304) for data-app reuse | trusting a default `SELECT *` as the full result (it is truncated at 500); querying Storage via raw Snowflake credentials outside the workspace abstraction | | Export a FILTERED or INCREMENTAL slice of a table (no workspace) | `kbagent storage download-table --project P --table-id in.c-foo.data --where-column status --where-value active [--where-operator eq\|neq] [--changed-since "-2 days"] [--changed-until WHEN]` (0.62.0+) -- server-side row filter + import-time window on the credential-only export path | `kbagent workspace query` with a `WHERE` clause when you need real SQL (needs a workspace) | downloading the whole table then filtering locally | diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index d388a336..566aa2ca 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -168,6 +168,11 @@ When working inside a git repository or project directory, run `kbagent init` (o | Delete one or more Storage Files | `kbagent storage file-delete --project PROJECT --file-id FILE-ID` | | Load a Storage File into a table | `kbagent storage load-file --project PROJECT --file-id FILE-ID --table-id TABLE-ID` | | Export a table to a Storage File | `kbagent storage unload-table --project PROJECT --table-id TABLE-ID` | +| List snapshots of a table | `kbagent storage snapshots --project PROJECT --table-id TABLE-ID` | +| Create a snapshot of a table (data + columns + primary key) | `kbagent storage snapshot-create --project PROJECT --table-id TABLE-ID` | +| Show one snapshot's detail (source table, creation time, description) | `kbagent storage snapshot-detail --project PROJECT --snapshot-id SNAPSHOT-ID` | +| Delete one or more table snapshots (the source tables are untouched) | `kbagent storage snapshot-delete --project PROJECT --snapshot-id SNAPSHOT-ID` | +| Create a NEW table from an existing snapshot (snapshot restore) | `kbagent storage table-from-snapshot --project PROJECT --snapshot-id SNAPSHOT-ID --bucket-id BUCKET-ID --name NAME` | | List Data Streams sources in a project | `kbagent stream list --project PROJECT` | | Create an OTLP (or HTTP) source and return its endpoint | `kbagent stream create-source --project PROJECT --name NAME` | | Show a source's endpoints, protocol, and destination tables | `kbagent stream detail [SOURCE-ID] --project PROJECT` | diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index e83cf22c..3c8fc538 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -115,6 +115,11 @@ Requires a **super-admin** Manage API token (same kind as `org setup`). Same def - `storage delete-bucket --project NAME --bucket-id ID [--bucket-id ...] [--force] [--dry-run] [--yes] [--branch ID]` -- delete buckets (branch-aware) - `storage swap-tables --project NAME --table-id ID --target-table-id ID --branch ID [--dry-run] [--yes]` (since v0.28.0) -- swap two storage tables in any branch, including the default/production branch (POST `/tables/{id}/swap`). Both tables exchange physical positions; aliases are NOT transferred (they keep pointing at the same physical position and therefore expose the OTHER table's data after the swap). Service refuses without a branch (active branch via `branch use` works too). Use to flip a typed rebuild ("data_change_log") into the original name ("data") without touching downstream config references - `storage clone-table --project NAME --table-id ID --branch ID [--dry-run]` (since v0.52.0) -- pull (clone) a production table into a dev branch (POST `/tables/{id}/pull`, operationName `devBranchTablePull`). On `storage-branches` projects a dev branch reads prod tables transparently until the first write, so an in-branch schema mutation (`swap-tables`, dropping a column) fails with a misleading "bucket not found" until the table is materialized branch-local; `clone-table` does that. One-way (default -> branch). Service refuses without a branch (active branch via `branch use` works too). Permission class `write` +- `storage snapshot-create --project NAME --table-id ID [--description D] [--branch ID]` (since v0.75.0) -- create a point-in-time snapshot of a table (data + columns + primary key; async `tableSnapshotCreate` job, client polls to completion). The receipt carries `snapshot_id` -- restores are addressed by it. Permission class `write` +- `storage snapshots --project NAME --table-id ID [--limit N] [--branch ID]` (since v0.75.0) -- list a table's snapshots (`id`, `createdTime`, `description`, `creatorToken`). Read command: production endpoint by default +- `storage snapshot-detail --project NAME --snapshot-id ID` (since v0.75.0) -- one snapshot's detail. Snapshot IDs are global (not table-scoped); the detail embeds the source `table` object (id, columns, primaryKey), so it traces a bare snapshot ID back to its origin table +- `storage table-from-snapshot --project NAME --snapshot-id ID --bucket-id ID --name NAME [--branch ID] [--dry-run]` (since v0.75.0) -- create a NEW table from an existing snapshot (issue #512; restore). Restores data, columns, and primary key into `--bucket-id` under `--name` via the classic `tables-async` endpoint (NOT `tables-definition` -- that is why it is not a `create-table` flag). `--name` is REQUIRED (API rejects empty). No overwrite: restoring onto an existing table name fails -- restore under a new name, verify, then `swap-tables` or `delete-table` yourself. Permission class `write` +- `storage snapshot-delete --project NAME --snapshot-id ID [--snapshot-id ...] [--dry-run] [--yes]` (since v0.75.0) -- delete snapshots (destructive: forecloses restores; source tables untouched). Batch-tolerant, exit 1 when any ID failed - `storage describe-bucket --project NAME --bucket-id ID [--text STR | --file PATH | --stdin] [--branch ID]` -- set a bucket description (stored as `KBC.description` in bucket metadata, upsert). Provide exactly one of `--text`, `--file`, `--stdin`. Read back via `storage bucket-detail` - `storage describe-table --project NAME --table-id ID [--text STR | --file PATH | --stdin] [--branch ID]` -- set a table description (stored as `KBC.description` in table metadata, upsert). Provide exactly one of `--text`, `--file`, `--stdin`. Read back via `storage table-detail` - `storage describe-column --project NAME --table-id ID --column NAME=DESCRIPTION [--column ...] [--branch ID]` -- set one or more column descriptions. Stored as `KBC.column.{name}.description` keys in the table's metadata (Keboola has no user-writable column-metadata endpoint). Read back in `storage table-detail` under `column_details[].description` diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index ace9091c..e72e3601 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -3026,3 +3026,26 @@ Four related sync-engine behaviors landed together (issues #466 / #467 / #472 / detected ... Local check only -- run 'kbagent sync diff'": `status` never contacts the API, so it cannot see remote drift; treating it as a local-vs-production audit was the #466 trap. + +## Table snapshots: restore is `tables-async`, `--name` is required, no overwrite (since v0.75.0) + +`kbagent storage table-from-snapshot` (issue #512) creates a NEW table from an +existing snapshot. Three traps, all verified live (us-east4.gcp, 2026-07-22): + +- **`--name` is REQUIRED.** The API rejects an omitted or empty name with + `Table create option "name" is required and cannot be empty` -- the reference + PHP client's "defaults to the snapshotted table's name" docblock is stale. +- **No overwrite semantics.** Restoring onto an existing table name fails with + a duplicate-name error. The safe pattern: restore under a new name, verify + the data, then `storage swap-tables` (or `delete-table` + re-restore) to + promote it. The destination bucket must already exist. +- **It is not a `create-table` flag.** Restore goes through the classic + `tables-async` import endpoint; `create-table` uses `tables-definition`, + which does not accept `snapshotId` -- hence the dedicated command. + +Lifecycle: `snapshot-create` (write; receipt carries `snapshot_id`), +`snapshots` (read, per table), `snapshot-detail` (read; global snapshot ID -> +embeds the source `table` object), `snapshot-delete` (destructive: forecloses +restores, source tables untouched; batch-tolerant, exit 1 on any failure). +Snapshot create and restore are async storage jobs -- the CLI polls to +completion, so the receipt's `table.rowsCount` is authoritative. diff --git a/pyproject.toml b/pyproject.toml index 522d5506..6a445cff 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "keboola-cli" -version = "0.74.0" +version = "0.75.0" description = "AI-friendly CLI for managing Keboola projects" readme = "README.md" requires-python = ">=3.12" diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index 551776e5..3da9f3f7 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -24,6 +24,22 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { + "0.75.0": [ + "New (#512): table snapshots -- `kbagent storage table-from-snapshot` creates a NEW " + "table from an existing snapshot (restore), plus the full lifecycle: " + "`snapshot-create`, `snapshots` (list), `snapshot-detail`, `snapshot-delete`. " + "A snapshot captures a table's data, columns, and primary key at a point in time; " + "restore rebuilds them into a fresh table in any existing bucket.", + "Note (#512): `table-from-snapshot --name` is REQUIRED -- the live API rejects an " + "omitted/empty name (the reference PHP client's 'defaults to snapshot name' docblock " + "is stale). Restore goes through the classic `tables-async` endpoint, not " + "`tables-definition`, which is why it is a dedicated command instead of a " + "`create-table` flag. No overwrite semantics: restore under a new name, verify, then " + "swap or delete the old table yourself.", + "Permissions: `storage.snapshots`/`snapshot-detail` classify as read, " + "`snapshot-create`/`table-from-snapshot` as write, `snapshot-delete` as destructive " + "(it forecloses restores; source tables are untouched).", + ], "0.74.0": [ "MCP passthrough deprecation (#478 phase 2, epic #390): `tool call` / `tool list` " "/ `agent --type mcp_tool` are now formally deprecated in favor of native commands. " diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index 7cd7120f..f5164f10 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -72,6 +72,7 @@ from .services.search_service import SearchService from .services.semantic_layer_service import SemanticLayerService from .services.sharing_service import SharingService +from .services.snapshot_service import SnapshotService from .services.storage_service import StorageService from .services.stream_service import StreamService from .services.sync_service import SyncService @@ -326,6 +327,7 @@ def main( branch_service = BranchService(config_store=config_store) sharing_service = SharingService(config_store=config_store) search_service = SearchService(config_store=config_store) + snapshot_service = SnapshotService(config_store=config_store) storage_service = StorageService(config_store=config_store) stream_service = StreamService(config_store=config_store) token_service = TokenService(config_store=config_store) @@ -383,6 +385,7 @@ def main( ctx.obj["branch_service"] = branch_service ctx.obj["sharing_service"] = sharing_service ctx.obj["search_service"] = search_service + ctx.obj["snapshot_service"] = snapshot_service ctx.obj["storage_service"] = storage_service ctx.obj["stream_service"] = stream_service ctx.obj["token_service"] = token_service diff --git a/src/keboola_agent_cli/client.py b/src/keboola_agent_cli/client.py index e6cc602d..24ebfc1e 100644 --- a/src/keboola_agent_cli/client.py +++ b/src/keboola_agent_cli/client.py @@ -2246,6 +2246,126 @@ def pull_table(self, table_id: str, branch_id: int) -> dict[str, Any]: response = self._request("POST", f"{prefix}/tables/{safe_id}/pull") return self._wait_for_storage_job(response.json()) + def create_table_snapshot( + self, + table_id: str, + description: str | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Create a snapshot of a table (async, waits for completion). + + Snapshots capture the table's data, columns, primary key, and + attributes at a point in time. A new table can later be created from + the snapshot with :meth:`create_table_from_snapshot`. + + Args: + table_id: Full ID of the table to snapshot (e.g. "in.c-bucket.table"). + description: Optional human-readable snapshot description. + branch_id: If set, snapshot the table in a specific dev branch. + + Returns: + Completed storage job results dict; contains the new snapshot "id". + """ + prefix = f"/v2/storage/branch/{branch_id}" if branch_id else "/v2/storage" + safe_id = quote(table_id, safe="") + body: dict[str, Any] = {} + if description is not None: + body["description"] = description + response = self._request("POST", f"{prefix}/tables/{safe_id}/snapshots", json=body) + job = self._wait_for_storage_job(response.json()) + return job.get("results", {}) + + def list_table_snapshots( + self, + table_id: str, + limit: int | None = None, + branch_id: int | None = None, + ) -> list[dict[str, Any]]: + """List snapshots of a table (sync). + + Args: + table_id: Full ID of the table (e.g. "in.c-bucket.table"). + limit: Optional maximum number of snapshots to return. + branch_id: If set, list snapshots in a specific dev branch. + + Returns: + List of snapshot dicts (id, createdTime, description, creatorToken, ...). + """ + prefix = f"/v2/storage/branch/{branch_id}" if branch_id else "/v2/storage" + safe_id = quote(table_id, safe="") + params: dict[str, str] = {} + if limit is not None: + params["limit"] = str(limit) + response = self._request( + "GET", f"{prefix}/tables/{safe_id}/snapshots", params=params or None + ) + return response.json() + + def get_snapshot(self, snapshot_id: str) -> dict[str, Any]: + """Get a single snapshot's detail by its ID (sync). + + Snapshot IDs are global (not table-scoped): the detail includes the + source ``table`` object, so this is also how a bare snapshot ID is + traced back to its origin table. + + Args: + snapshot_id: Numeric snapshot ID. + + Returns: + Snapshot dict (id, table, createdTime, description, ...). + """ + safe_id = quote(str(snapshot_id), safe="") + response = self._request("GET", f"/v2/storage/snapshots/{safe_id}") + return response.json() + + def delete_snapshot(self, snapshot_id: str) -> None: + """Delete a snapshot by its ID. + + The endpoint normally responds synchronously (204); a 202 job + response is polled to completion for forward compatibility. + + Args: + snapshot_id: Numeric snapshot ID. + """ + safe_id = quote(str(snapshot_id), safe="") + response = self._request("DELETE", f"/v2/storage/snapshots/{safe_id}") + if response.status_code == 202: + self._wait_for_storage_job(response.json()) + + def create_table_from_snapshot( + self, + bucket_id: str, + snapshot_id: str, + name: str, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Create a NEW table from an existing snapshot (async, waits). + + Hits the classic ``tables-async`` import endpoint with a + ``snapshotId`` source (the ``tables-definition`` endpoint used by + :meth:`create_table` does not accept snapshots). The new table + restores the snapshot's data, columns, and primary key. ``name`` is + required: the API rejects an omitted/empty name ("Table create option + \"name\" is required and cannot be empty", verified live 2026-07-22 -- + the reference PHP client's "fetched from snapshot" docblock is stale). + + Args: + bucket_id: Destination bucket ID (e.g. "in.c-my-bucket"). + snapshot_id: Numeric ID of the source snapshot. + name: Name for the new table (required by the API). + branch_id: If set, create the table in a specific dev branch. + + Returns: + Completed storage job results dict -- the created table (its "id" + is the new full table ID). + """ + prefix = f"/v2/storage/branch/{branch_id}" if branch_id else "/v2/storage" + safe_id = quote(bucket_id, safe="") + body: dict[str, Any] = {"snapshotId": snapshot_id, "name": name} + response = self._request("POST", f"{prefix}/buckets/{safe_id}/tables-async", json=body) + job = self._wait_for_storage_job(response.json()) + return job.get("results", {}) + def list_tables_with_metadata(self) -> list[dict[str, Any]]: """List all storage tables with columns and metadata. diff --git a/src/keboola_agent_cli/commands/_storage_snapshots.py b/src/keboola_agent_cli/commands/_storage_snapshots.py new file mode 100644 index 00000000..02222c5f --- /dev/null +++ b/src/keboola_agent_cli/commands/_storage_snapshots.py @@ -0,0 +1,299 @@ +"""Table snapshot commands for the ``kbagent storage`` group (issue #512). + +Thin CLI layer over :class:`services.snapshot_service.SnapshotService`: +``snapshots`` (list), ``snapshot-create``, ``snapshot-detail``, +``snapshot-delete``, and ``table-from-snapshot`` (create a NEW table from an +existing snapshot -- the core ask of #512). + +Lives in a private module because ``commands/storage.py`` is already past the +1,200-LOC commands-file ceiling (CONTRIBUTING.md). The commands are mounted +flat onto ``storage_app`` via :func:`register`, so permission keys stay in the +``storage.*`` namespace and ``kbagent storage --help`` lists them alongside +the other table commands. +""" + +from __future__ import annotations + +from typing import Any + +import typer +from rich.markup import escape + +from ..config_store import ConfigStore +from ..errors import ConfigError, ErrorCode, KeboolaApiError +from ._helpers import ( + get_formatter, + get_service, + map_error_to_exit_code, + resolve_branch, +) + +_SNAPSHOTS = "Snapshots" + + +def _handle_errors(formatter: Any, exc: Exception) -> None: + """Map ConfigError / KeboolaApiError to a structured error + Exit.""" + if isinstance(exc, ConfigError): + formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR) + raise typer.Exit(code=5) from None + if isinstance(exc, KeboolaApiError): + formatter.error(message=exc.message, error_code=exc.error_code, retryable=exc.retryable) + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + raise exc + + +def _snapshot_created(snapshot: dict[str, Any]) -> str: + """Best-effort creation timestamp from a raw API snapshot dict.""" + return str(snapshot.get("created") or snapshot.get("createdTime") or "") + + +def register(app: typer.Typer) -> None: + """Mount the snapshot commands onto ``app`` (the ``storage`` Typer group).""" + + @app.command("snapshots", rich_help_panel=_SNAPSHOTS) + def storage_snapshots( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + table_id: str = typer.Option( + ..., "--table-id", help="Table ID (e.g. 'in.c-my-bucket.my-table')" + ), + limit: int | None = typer.Option(None, "--limit", help="Max snapshots to return"), + branch: int | None = typer.Option( + None, "--branch", help="Dev branch ID (production endpoint by default)" + ), + ) -> None: + """List snapshots of a table. + + Each snapshot captures the table's data, columns, and primary key at + a point in time. Use the snapshot ID with `storage table-from-snapshot` + to restore it as a new table. + """ + formatter = get_formatter(ctx) + service = get_service(ctx, "snapshot_service") + config_store: ConfigStore = ctx.obj["config_store"] + # Read command: ignore implicit active dev branch (empty listing trap). + _, effective_branch = resolve_branch( + config_store, formatter, project, branch, ignore_active_branch=True + ) + + try: + result = service.list_snapshots( + alias=project, + table_id=table_id, + limit=limit, + branch_id=effective_branch, + ) + except (ConfigError, KeboolaApiError) as exc: + _handle_errors(formatter, exc) + + if formatter.json_mode: + formatter.output(result) + return + + from rich.table import Table + + rich_table = Table(title=f"Snapshots - {result['table_id']} ({result['count']})") + rich_table.add_column("ID", style="cyan") + rich_table.add_column("Created") + rich_table.add_column("Description") + for snap in result["snapshots"]: + rich_table.add_row( + str(snap.get("id", "")), + _snapshot_created(snap), + escape(str(snap.get("description") or "")), + ) + formatter.console.print(rich_table) + if result["count"]: + formatter.console.print( + "[dim]Restore one as a new table with:[/dim] " + f"kbagent storage table-from-snapshot --project {result['project_alias']} " + "--snapshot-id --bucket-id --name " + ) + + @app.command("snapshot-create", rich_help_panel=_SNAPSHOTS) + def storage_snapshot_create( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + table_id: str = typer.Option( + ..., "--table-id", help="Table ID to snapshot (e.g. 'in.c-my-bucket.my-table')" + ), + description: str | None = typer.Option( + None, "--description", help="Human-readable snapshot description" + ), + branch: int | None = typer.Option( + None, + "--branch", + help="Dev branch ID (defaults to active branch if set via 'branch use')", + ), + ) -> None: + """Create a snapshot of a table (data + columns + primary key). + + The snapshot is a point-in-time backup stored by the platform. Restore + it later as a NEW table with `storage table-from-snapshot`. + """ + formatter = get_formatter(ctx) + service = get_service(ctx, "snapshot_service") + config_store: ConfigStore = ctx.obj["config_store"] + _, effective_branch = resolve_branch(config_store, formatter, project, branch) + + try: + result = service.create_snapshot( + alias=project, + table_id=table_id, + description=description, + branch_id=effective_branch, + ) + except (ConfigError, KeboolaApiError) as exc: + _handle_errors(formatter, exc) + + if formatter.json_mode: + formatter.output(result) + return + formatter.console.print( + f"[bold green]Snapshot created:[/bold green] ID {result['snapshot_id']} " + f"(table {result['table_id']})" + ) + formatter.console.print( + "[dim]Restore as a new table with:[/dim] " + f"kbagent storage table-from-snapshot --project {result['project_alias']} " + f"--snapshot-id {result['snapshot_id']} --bucket-id --name " + ) + + @app.command("snapshot-detail", rich_help_panel=_SNAPSHOTS) + def storage_snapshot_detail( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + snapshot_id: str = typer.Option(..., "--snapshot-id", help="Snapshot ID"), + ) -> None: + """Show one snapshot's detail (source table, creation time, description).""" + formatter = get_formatter(ctx) + service = get_service(ctx, "snapshot_service") + + try: + result = service.get_snapshot(alias=project, snapshot_id=snapshot_id) + except (ConfigError, KeboolaApiError) as exc: + _handle_errors(formatter, exc) + + if formatter.json_mode: + formatter.output(result) + return + snap = result["snapshot"] + source_table = snap.get("table") or {} + formatter.console.print(f"[bold]Snapshot:[/bold] {snap.get('id', '')}") + formatter.console.print(f" Created: {_snapshot_created(snap)}") + formatter.console.print(f" Description: {escape(str(snap.get('description') or ''))}") + if source_table: + formatter.console.print(f" Source table: {source_table.get('id', '')}") + + @app.command("snapshot-delete", rich_help_panel=_SNAPSHOTS) + def storage_snapshot_delete( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + snapshot_id: list[str] = typer.Option( + ..., "--snapshot-id", help="Snapshot ID to delete (repeat for multiple)" + ), + dry_run: bool = typer.Option( + False, "--dry-run", help="Show what would be deleted without executing" + ), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt"), + ) -> None: + """Delete one or more table snapshots (the source tables are untouched).""" + formatter = get_formatter(ctx) + service = get_service(ctx, "snapshot_service") + + if ( + not dry_run + and not yes + and not formatter.json_mode + and not typer.confirm( + f"Delete {len(snapshot_id)} snapshot(s) in project '{project}'? " + "Restoring from them will no longer be possible." + ) + ): + formatter.console.print("Aborted.") + raise typer.Exit(code=0) + + try: + result = service.delete_snapshots( + alias=project, + snapshot_ids=snapshot_id, + dry_run=dry_run, + ) + except (ConfigError, KeboolaApiError) as exc: + _handle_errors(formatter, exc) + + if formatter.json_mode: + formatter.output(result) + else: + if dry_run: + for sid in result.get("would_delete", []): + formatter.console.print(f"[bold blue]Would delete:[/bold blue] snapshot {sid}") + else: + for sid in result["deleted"]: + formatter.console.print(f"[bold green]Deleted:[/bold green] snapshot {sid}") + for s_err in result["failed"]: + formatter.console.print( + f"[bold red]Failed:[/bold red] snapshot {s_err['id']}: {s_err['error']}" + ) + + if result["failed"]: + raise typer.Exit(code=1) + + @app.command("table-from-snapshot", rich_help_panel=_SNAPSHOTS) + def storage_table_from_snapshot( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + snapshot_id: str = typer.Option(..., "--snapshot-id", help="Source snapshot ID"), + bucket_id: str = typer.Option( + ..., "--bucket-id", help="Destination bucket ID (e.g. 'in.c-my-bucket')" + ), + name: str = typer.Option( + ..., "--name", help="Name for the new table (required by the API)" + ), + branch: int | None = typer.Option( + None, + "--branch", + help="Dev branch ID (defaults to active branch if set via 'branch use')", + ), + dry_run: bool = typer.Option( + False, "--dry-run", help="Show what would be created without executing" + ), + ) -> None: + """Create a NEW table from an existing snapshot (snapshot restore). + + Restores the snapshot's data, columns, and primary key into a fresh + table in --bucket-id. The destination bucket must exist; a table with + the target name must not (the API rejects overwrites). Find snapshot + IDs with `storage snapshots --table-id ...`. + """ + formatter = get_formatter(ctx) + service = get_service(ctx, "snapshot_service") + config_store: ConfigStore = ctx.obj["config_store"] + _, effective_branch = resolve_branch(config_store, formatter, project, branch) + + try: + result = service.create_table_from_snapshot( + alias=project, + bucket_id=bucket_id, + snapshot_id=snapshot_id, + name=name, + branch_id=effective_branch, + dry_run=dry_run, + ) + except (ConfigError, KeboolaApiError) as exc: + _handle_errors(formatter, exc) + + if formatter.json_mode: + formatter.output(result) + return + if result["dry_run"]: + formatter.console.print( + f"[bold blue]Would create:[/bold blue] table " + f"'{result['name']}' in bucket " + f"{result['bucket_id']} from snapshot {result['snapshot_id']}" + ) + return + formatter.console.print( + f"[bold green]Table created:[/bold green] {result['table_id']} " + f"(from snapshot {result['snapshot_id']})" + ) diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index 628eeea6..57a3bab6 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -453,6 +453,31 @@ (swap-tables, dropping columns) first needs a branch-local copy. This materializes that copy (one-way: default -> branch). Branch is mandatory; service guards before any HTTP call when no branch is set. +### Table Snapshots (point-in-time backup + restore-as-new-table) + + kbagent storage snapshot-create --project NAME --table-id ID [--description D] [--branch ID] + Create a snapshot of a table: data + columns + primary key at a point in time (async job; polls to + completion). The receipt carries the new snapshot_id -- keep it, restores are addressed by it. + + kbagent storage snapshots --project NAME --table-id ID [--limit N] [--branch ID] + List snapshots of a table (id, createdTime, description, creator). Production endpoint by default. + + kbagent storage snapshot-detail --project NAME --snapshot-id ID + One snapshot's detail. Snapshot IDs are global (not table-scoped); the detail includes the source + table object (id, columns, primaryKey), so this is how a bare snapshot ID is traced back to its table. + + kbagent storage table-from-snapshot --project NAME --snapshot-id ID --bucket-id ID --name NAME [--branch ID] [--dry-run] + Create a NEW table from an existing snapshot (snapshot restore; async job). Restores the snapshot's + data, columns, and primary key into --bucket-id under --name. --name is REQUIRED (the API rejects an + omitted/empty name). The destination bucket must exist; a table with the same name must not (no + overwrite semantics -- restore under a new name, verify, then swap or delete the old table yourself). + Goes through the classic tables-async endpoint, NOT tables-definition -- so it is a separate command, + not a flag on create-table. + + kbagent storage snapshot-delete --project NAME --snapshot-id ID [--snapshot-id ...] [--dry-run] [--yes] + Delete one or more snapshots (destructive: forecloses restores; source tables untouched). + Batch-tolerant: one failure does not abort the rest; exit 1 if any ID failed. + ### Storage Descriptions kbagent storage describe-bucket --project NAME --bucket-id ID [--text STR | --file PATH | --stdin] [--branch ID] diff --git a/src/keboola_agent_cli/commands/storage.py b/src/keboola_agent_cli/commands/storage.py index 47fe9ee0..0ec0ea6c 100644 --- a/src/keboola_agent_cli/commands/storage.py +++ b/src/keboola_agent_cli/commands/storage.py @@ -2726,3 +2726,11 @@ def storage_unload_table( slice_count = result.get("slice_count") suffix = f", {slice_count} slices" if slice_count else "" formatter.console.print(f" Downloaded to: {result['output_path']} ({dl_size}{suffix})") + + +# Snapshot commands (issue #512) live in a private module -- storage.py is +# past the commands-file size ceiling. Mounted here so they share the +# `storage.*` permission namespace and appear in `kbagent storage --help`. +from ._storage_snapshots import register as _register_snapshot_commands # noqa: E402 + +_register_snapshot_commands(storage_app) diff --git a/src/keboola_agent_cli/permissions.py b/src/keboola_agent_cli/permissions.py index c1f45886..48444295 100644 --- a/src/keboola_agent_cli/permissions.py +++ b/src/keboola_agent_cli/permissions.py @@ -223,6 +223,13 @@ "storage.file-delete": "destructive", "storage.swap-tables": "destructive", "storage.truncate-table": "destructive", + # Storage snapshots (issue #512). snapshot-delete removes only the backup + # (source table untouched) but forecloses restores -- destructive. + "storage.snapshots": "read", + "storage.snapshot-detail": "read", + "storage.snapshot-create": "write", + "storage.table-from-snapshot": "write", + "storage.snapshot-delete": "destructive", # Storage descriptions "storage.describe-bucket": "write", "storage.describe-table": "write", diff --git a/src/keboola_agent_cli/services/snapshot_service.py b/src/keboola_agent_cli/services/snapshot_service.py new file mode 100644 index 00000000..1db3f7b2 --- /dev/null +++ b/src/keboola_agent_cli/services/snapshot_service.py @@ -0,0 +1,247 @@ +"""Table snapshot service -- create, list, inspect, delete, restore-as-new-table. + +Business logic for the ``kbagent storage snapshot-*`` / ``table-from-snapshot`` +commands (issue #512). Wraps the Storage API table-snapshot endpoints behind +kbagent-alias resolution and an injectable :class:`KeboolaClient` factory: + +- ``POST /v2/storage/tables/{id}/snapshots`` -- create (async storage job) +- ``GET /v2/storage/tables/{id}/snapshots`` -- list (sync) +- ``GET /v2/storage/snapshots/{id}`` -- detail (sync) +- ``DELETE /v2/storage/snapshots/{id}`` -- delete (sync) +- ``POST /v2/storage/buckets/{id}/tables-async`` with ``snapshotId`` -- + create a NEW table from a snapshot (async storage job) + +Endpoint contract verified against the reference PHP client +(``keboola/storage-api-php-client``: ``createTableSnapshot``, +``listTableSnapshots``, ``getSnapshot``, ``deleteSnapshot``, +``createTableFromSnapshot``). Restore goes through the classic +``tables-async`` import endpoint -- NOT ``tables-definition`` -- which is why +this is a dedicated command instead of a flag on ``storage create-table``. + +Lives in its own module because ``storage_service.py`` is already past its +file-size hard ceiling (CONTRIBUTING.md); mirrors the ``TokenService`` layout. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from typing import Any + +from ..client import KeboolaClient +from ..config_store import ConfigStore +from ..errors import ConfigError, KeboolaApiError + +logger = logging.getLogger(__name__) + +KeboolaClientFactory = Callable[[str, str], KeboolaClient] + + +def default_snapshot_client_factory(stack_url: str, token: str) -> KeboolaClient: + """Construct a :class:`KeboolaClient` bound to ``stack_url`` + ``token``.""" + return KeboolaClient(stack_url=stack_url, token=token) + + +class SnapshotService: + """Business logic for table snapshot create / list / detail / delete / restore.""" + + def __init__( + self, + config_store: ConfigStore, + client_factory: KeboolaClientFactory | None = None, + ) -> None: + self._config_store = config_store + self._client_factory = client_factory or default_snapshot_client_factory + + def create_snapshot( + self, + *, + alias: str, + table_id: str, + description: str | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Snapshot ``table_id`` and return the new snapshot's ID. + + Returns: + Dict with 'project_alias', 'table_id', 'branch_id', 'snapshot_id'. + """ + stack_url, token = self._resolve_project(alias) + client = self._client_factory(stack_url, token) + try: + results = client.create_table_snapshot( + table_id=table_id, + description=description, + branch_id=branch_id, + ) + finally: + client.close() + return { + "project_alias": alias, + "table_id": table_id, + "branch_id": branch_id, + "snapshot_id": results.get("id"), + "description": description, + } + + def list_snapshots( + self, + *, + alias: str, + table_id: str, + limit: int | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """List snapshots of ``table_id`` (newest first, as returned by the API). + + Returns: + Dict with 'project_alias', 'table_id', 'branch_id', 'count', + 'snapshots' (raw API snapshot dicts). + """ + stack_url, token = self._resolve_project(alias) + client = self._client_factory(stack_url, token) + try: + snapshots = client.list_table_snapshots( + table_id=table_id, + limit=limit, + branch_id=branch_id, + ) + finally: + client.close() + return { + "project_alias": alias, + "table_id": table_id, + "branch_id": branch_id, + "count": len(snapshots), + "snapshots": snapshots, + } + + def get_snapshot(self, *, alias: str, snapshot_id: str) -> dict[str, Any]: + """Fetch one snapshot's detail (includes the source table object). + + Returns: + Dict with 'project_alias', 'snapshot' (raw API snapshot dict). + """ + stack_url, token = self._resolve_project(alias) + client = self._client_factory(stack_url, token) + try: + snapshot = client.get_snapshot(snapshot_id) + finally: + client.close() + return {"project_alias": alias, "snapshot": snapshot} + + def delete_snapshots( + self, + *, + alias: str, + snapshot_ids: list[str], + dry_run: bool = False, + ) -> dict[str, Any]: + """Delete one or more snapshots. Batch-tolerant: accumulates errors per ID. + + Returns: + Dict with 'project_alias', 'deleted', 'failed', 'dry_run' (and + 'would_delete' when dry_run). + """ + stack_url, token = self._resolve_project(alias) + + deleted: list[str] = [] + failed: list[dict[str, Any]] = [] + would_delete: list[str] = [] + + client = self._client_factory(stack_url, token) + try: + for sid in snapshot_ids: + if dry_run: + would_delete.append(sid) + continue + try: + client.delete_snapshot(sid) + deleted.append(sid) + except KeboolaApiError as exc: + failed.append({"id": sid, "error": exc.message}) + finally: + client.close() + + result: dict[str, Any] = { + "project_alias": alias, + "deleted": deleted, + "failed": failed, + "dry_run": dry_run, + } + if dry_run: + result["would_delete"] = would_delete + return result + + def create_table_from_snapshot( + self, + *, + alias: str, + bucket_id: str, + snapshot_id: str, + name: str, + branch_id: int | None = None, + dry_run: bool = False, + ) -> dict[str, Any]: + """Create a NEW table in ``bucket_id`` from snapshot ``snapshot_id``. + + The snapshot's data, columns, and primary key are restored into a + fresh table named ``name`` (required -- the API rejects an empty + name). The destination bucket must already exist; restoring over an + existing table name fails with the API's duplicate-table error -- + this service adds no overwrite semantics on top. + + Returns: + Dict with 'project_alias', 'bucket_id', 'snapshot_id', 'name', + 'branch_id', 'dry_run', and (when applied) 'table' (raw job + results incl. the created table 'id') + 'table_id'. + """ + if not bucket_id.strip(): + raise ConfigError("table-from-snapshot requires a non-empty --bucket-id.") + if not str(snapshot_id).strip(): + raise ConfigError("table-from-snapshot requires a non-empty --snapshot-id.") + if not name.strip(): + raise ConfigError("table-from-snapshot requires a non-empty --name.") + + stack_url, token = self._resolve_project(alias) + + if dry_run: + return { + "project_alias": alias, + "bucket_id": bucket_id, + "snapshot_id": snapshot_id, + "name": name, + "branch_id": branch_id, + "dry_run": True, + } + + client = self._client_factory(stack_url, token) + try: + table = client.create_table_from_snapshot( + bucket_id=bucket_id, + snapshot_id=snapshot_id, + name=name, + branch_id=branch_id, + ) + finally: + client.close() + + return { + "project_alias": alias, + "bucket_id": bucket_id, + "snapshot_id": snapshot_id, + "name": name, + "branch_id": branch_id, + "dry_run": False, + "table": table, + "table_id": table.get("id"), + } + + def _resolve_project(self, alias: str) -> tuple[str, str]: + """Resolve ``alias`` to its ``(stack_url, token)``.""" + project = self._config_store.get_project(alias) + if project is None: + raise ConfigError( + f"Project alias '{alias}' is not registered. Run `kbagent project list`." + ) + return project.stack_url, project.token diff --git a/tests/test_e2e.py b/tests/test_e2e.py index b71bfe39..24983b2c 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -455,6 +455,13 @@ def test_full_cli_e2e(self) -> None: ) self._test_truncate_table_roundtrip(table_id) + _step( + 11.2, + "storage snapshot-* + table-from-snapshot", + "snapshot lifecycle + restore as new table (issue #512)", + ) + self._test_snapshot_roundtrip(bucket_id, table_id) + _step(12, "storage tables + table-detail") self._test_table_listing(bucket_id, table_id) @@ -1096,6 +1103,162 @@ def _test_truncate_table_roundtrip(self, table_id: str) -> None: f"restore failed: expected {before['rows_count']} rows, got {restored['rows_count']}" ) + def _test_snapshot_roundtrip(self, bucket_id: str, table_id: str) -> None: + """Snapshot lifecycle + restore-as-new-table (issue #512). + + create -> list -> detail -> table-from-snapshot (dry-run, then apply) + -> verify the restored table matches the source (rows, columns, + primary key) -> delete the restored table and the snapshot -> verify + the snapshot is gone. Leaves the source table untouched for the + downstream hops. + """ + before = self._run_ok( + "storage", + "table-detail", + "--project", + self.alias, + "--table-id", + table_id, + )["data"] + assert before["rows_count"] > 0, "snapshot roundtrip needs a non-empty table" + before_columns = sorted(c["name"] for c in before["column_details"]) + before_pk = list(before.get("primary_key") or []) + + # Create a snapshot; the receipt carries the new snapshot ID. + created = self._run_ok( + "storage", + "snapshot-create", + "--project", + self.alias, + "--table-id", + table_id, + "--description", + "kbagent E2E snapshot roundtrip", + )["data"] + snapshot_id = str(created["snapshot_id"]) + assert snapshot_id, f"snapshot-create returned no snapshot_id: {created}" + + # List: the new snapshot must appear for the source table. + listed = self._run_ok( + "storage", + "snapshots", + "--project", + self.alias, + "--table-id", + table_id, + )["data"] + assert listed["count"] >= 1 + assert snapshot_id in [str(s["id"]) for s in listed["snapshots"]] + + # Detail: the snapshot must point back at the source table. + detail = self._run_ok( + "storage", + "snapshot-detail", + "--project", + self.alias, + "--snapshot-id", + snapshot_id, + )["data"] + assert str(detail["snapshot"]["id"]) == snapshot_id + assert detail["snapshot"]["table"]["id"] == table_id + + # Restore dry-run: receipt only, no table created. + restored_name = "e2e_snapshot_restore" + restored_table_id = f"{bucket_id}.{restored_name}" + dry = self._run_ok( + "storage", + "table-from-snapshot", + "--project", + self.alias, + "--snapshot-id", + snapshot_id, + "--bucket-id", + bucket_id, + "--name", + restored_name, + "--dry-run", + )["data"] + assert dry["dry_run"] is True + tables = self._run_ok( + "storage", "tables", "--project", self.alias, "--bucket-id", bucket_id + )["data"]["tables"] + assert restored_table_id not in [t["id"] for t in tables], ( + "dry-run must not create the table" + ) + + # Restore for real: a NEW table with the snapshot's data appears. + applied = self._run_ok( + "storage", + "table-from-snapshot", + "--project", + self.alias, + "--snapshot-id", + snapshot_id, + "--bucket-id", + bucket_id, + "--name", + restored_name, + )["data"] + assert applied["dry_run"] is False + assert applied["table_id"] == restored_table_id + + # The restored table must match the source: rows, columns, primary key. + restored = self._run_ok( + "storage", + "table-detail", + "--project", + self.alias, + "--table-id", + restored_table_id, + )["data"] + assert restored["rows_count"] == before["rows_count"], ( + f"restored rows {restored['rows_count']} != source rows {before['rows_count']}" + ) + assert sorted(c["name"] for c in restored["column_details"]) == before_columns + assert list(restored.get("primary_key") or []) == before_pk + + # Drop the restored table so downstream bucket listings are unchanged. + self._run_ok( + "storage", + "delete-table", + "--project", + self.alias, + "--table-id", + restored_table_id, + "--yes", + ) + + # Delete the snapshot (dry-run first) and verify it is gone. + del_dry = self._run_ok( + "storage", + "snapshot-delete", + "--project", + self.alias, + "--snapshot-id", + snapshot_id, + "--dry-run", + )["data"] + assert del_dry["would_delete"] == [snapshot_id] + deleted = self._run_ok( + "storage", + "snapshot-delete", + "--project", + self.alias, + "--snapshot-id", + snapshot_id, + )["data"] + assert deleted["deleted"] == [snapshot_id] + assert deleted["failed"] == [] + after_delete = self._run_ok( + "storage", + "snapshots", + "--project", + self.alias, + "--table-id", + table_id, + )["data"] + assert snapshot_id not in [str(s["id"]) for s in after_delete["snapshots"]] + def _test_table_listing(self, bucket_id: str, table_id: str) -> None: """Verify table appears in listings and detail is correct.""" # tables diff --git a/tests/test_storage_snapshots.py b/tests/test_storage_snapshots.py new file mode 100644 index 00000000..5b77766e --- /dev/null +++ b/tests/test_storage_snapshots.py @@ -0,0 +1,708 @@ +"""Tests for storage table snapshots (issue #512): client, service, and CLI. + +Covers the five commands -- snapshots, snapshot-create, snapshot-detail, +snapshot-delete, table-from-snapshot -- across the three layers. Endpoint +shapes mirror the live API (verified on connection.us-east4.gcp.keboola.com, +2026-07-22): snapshot create and table-from-snapshot return queued storage +jobs; list/detail/delete are synchronous. +""" + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from keboola_agent_cli.cli import app +from keboola_agent_cli.client import KeboolaClient +from keboola_agent_cli.config_store import ConfigStore +from keboola_agent_cli.errors import ConfigError, KeboolaApiError +from keboola_agent_cli.models import AppConfig, ProjectConfig +from keboola_agent_cli.services.snapshot_service import SnapshotService + +runner = CliRunner() + +TEST_TOKEN = "901-55555-fakeTestTokenDoNotUseXXXXXXXX" + +SNAPSHOT = { + "id": "954", + "description": "before migration", + "createdTime": "2026-07-22T00:05:24+0200", + "type": "table", + "creatorToken": {"id": 1, "description": "petr@example.com"}, +} + + +def _make_store(tmp_path: Path) -> ConfigStore: + config_dir = tmp_path / "config" + config_dir.mkdir(exist_ok=True) + store = ConfigStore(config_dir=config_dir) + config = AppConfig( + projects={ + "test": ProjectConfig( + stack_url="https://connection.keboola.com", + token=TEST_TOKEN, + ) + }, + ) + store.save(config) + return store + + +def _make_service(store: ConfigStore, mock_client: MagicMock) -> SnapshotService: + return SnapshotService( + config_store=store, + client_factory=lambda url, token: mock_client, + ) + + +# --------------------------------------------------------------------------- +# Client layer +# --------------------------------------------------------------------------- + + +class TestSnapshotClient: + """HTTP-shape tests for the five KeboolaClient snapshot methods.""" + + def _client(self) -> KeboolaClient: + return KeboolaClient( + stack_url="https://connection.keboola.com", + token=TEST_TOKEN, + ) + + def test_create_snapshot_posts_and_polls_job(self, httpx_mock) -> None: + """POST tables/{id}/snapshots with description; 202 job polled to results.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/tables/in.c-foo.data/snapshots", + method="POST", + json={"id": 555, "status": "waiting", "operationName": "tableSnapshotCreate"}, + status_code=202, + ) + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/jobs/555", + method="GET", + json={"id": 555, "status": "success", "results": {"id": "954"}}, + status_code=200, + ) + + client = self._client() + results = client.create_table_snapshot( + table_id="in.c-foo.data", description="before migration" + ) + client.close() + + assert results == {"id": "954"} + request = httpx_mock.get_requests()[0] + assert json.loads(request.content) == {"description": "before migration"} + + def test_create_snapshot_branch_prefix(self, httpx_mock) -> None: + """branch_id=42 routes through /v2/storage/branch/42/tables/.../snapshots.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/branch/42/tables/in.c-foo.data/snapshots", + method="POST", + json={"id": 1, "status": "success", "results": {"id": "7"}}, + status_code=202, + ) + + client = self._client() + results = client.create_table_snapshot(table_id="in.c-foo.data", branch_id=42) + client.close() + + assert results == {"id": "7"} + # No description passed -> empty JSON body (the API treats it as optional). + assert json.loads(httpx_mock.get_requests()[0].content) == {} + + def test_list_snapshots_plain_and_with_limit(self, httpx_mock) -> None: + """GET tables/{id}/snapshots returns the raw list; limit becomes a query param.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/tables/in.c-foo.data/snapshots", + method="GET", + json=[SNAPSHOT], + status_code=200, + ) + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/tables/in.c-foo.data/snapshots?limit=5", + method="GET", + json=[SNAPSHOT], + status_code=200, + ) + + client = self._client() + assert client.list_table_snapshots(table_id="in.c-foo.data") == [SNAPSHOT] + assert client.list_table_snapshots(table_id="in.c-foo.data", limit=5) == [SNAPSHOT] + client.close() + + def test_get_snapshot(self, httpx_mock) -> None: + """GET /v2/storage/snapshots/{id} returns the snapshot dict.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/snapshots/954", + method="GET", + json={**SNAPSHOT, "table": {"id": "in.c-foo.data"}}, + status_code=200, + ) + + client = self._client() + snapshot = client.get_snapshot("954") + client.close() + + assert snapshot["id"] == "954" + assert snapshot["table"]["id"] == "in.c-foo.data" + + def test_delete_snapshot_sync_204(self, httpx_mock) -> None: + """DELETE /v2/storage/snapshots/{id} with a 204 response does not poll.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/snapshots/954", + method="DELETE", + status_code=204, + ) + + client = self._client() + client.delete_snapshot("954") + client.close() + + assert len(httpx_mock.get_requests()) == 1 + + def test_delete_snapshot_202_job_polled(self, httpx_mock) -> None: + """Forward-compat: a 202 job response is polled to completion.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/snapshots/954", + method="DELETE", + json={"id": 888, "status": "waiting"}, + status_code=202, + ) + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/jobs/888", + method="GET", + json={"id": 888, "status": "success"}, + status_code=200, + ) + + client = self._client() + client.delete_snapshot("954") + client.close() + + assert len(httpx_mock.get_requests()) == 2 + + def test_table_from_snapshot_posts_tables_async(self, httpx_mock) -> None: + """POST buckets/{id}/tables-async with snapshotId+name; job polled to results. + + The restore goes through the classic tables-async import endpoint -- + NOT tables-definition -- matching the reference PHP client's + createTableFromSnapshot. + """ + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/buckets/in.c-dest/tables-async", + method="POST", + json={"id": 999, "status": "waiting", "operationName": "tableCreate"}, + status_code=202, + ) + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/jobs/999", + method="GET", + json={ + "id": 999, + "status": "success", + "results": {"id": "in.c-dest.restored", "rowsCount": 3}, + }, + status_code=200, + ) + + client = self._client() + results = client.create_table_from_snapshot( + bucket_id="in.c-dest", snapshot_id="954", name="restored" + ) + client.close() + + assert results["id"] == "in.c-dest.restored" + request = httpx_mock.get_requests()[0] + assert json.loads(request.content) == {"snapshotId": "954", "name": "restored"} + + def test_table_from_snapshot_branch_prefix(self, httpx_mock) -> None: + """branch_id routes through /v2/storage/branch/{id}/buckets/.../tables-async.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/branch/42/buckets/in.c-dest/tables-async", + method="POST", + json={"id": 1, "status": "success", "results": {"id": "in.c-dest.restored"}}, + status_code=202, + ) + + client = self._client() + results = client.create_table_from_snapshot( + bucket_id="in.c-dest", snapshot_id="954", name="restored", branch_id=42 + ) + client.close() + + assert results["id"] == "in.c-dest.restored" + + def test_api_error_propagates(self, httpx_mock) -> None: + """Storage API 4xx propagates as KeboolaApiError.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/snapshots/999", + method="GET", + json={"error": "Snapshot 999 not found."}, + status_code=404, + ) + + client = self._client() + with pytest.raises(KeboolaApiError): + client.get_snapshot("999") + client.close() + + +# --------------------------------------------------------------------------- +# Service layer +# --------------------------------------------------------------------------- + + +class TestSnapshotService: + """Business-logic tests for SnapshotService.""" + + def test_create_snapshot_returns_id(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + mock_client = MagicMock() + mock_client.create_table_snapshot.return_value = {"id": "954"} + service = _make_service(store, mock_client) + + result = service.create_snapshot( + alias="test", table_id="in.c-foo.data", description="d", branch_id=None + ) + + assert result["snapshot_id"] == "954" + assert result["table_id"] == "in.c-foo.data" + assert result["project_alias"] == "test" + mock_client.create_table_snapshot.assert_called_once_with( + table_id="in.c-foo.data", description="d", branch_id=None + ) + mock_client.close.assert_called_once() + + def test_list_snapshots_counts(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + mock_client = MagicMock() + mock_client.list_table_snapshots.return_value = [SNAPSHOT, {**SNAPSHOT, "id": "955"}] + service = _make_service(store, mock_client) + + result = service.list_snapshots(alias="test", table_id="in.c-foo.data", limit=10) + + assert result["count"] == 2 + assert result["snapshots"][1]["id"] == "955" + mock_client.list_table_snapshots.assert_called_once_with( + table_id="in.c-foo.data", limit=10, branch_id=None + ) + + def test_get_snapshot_passthrough(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + mock_client = MagicMock() + mock_client.get_snapshot.return_value = {**SNAPSHOT, "table": {"id": "in.c-foo.data"}} + service = _make_service(store, mock_client) + + result = service.get_snapshot(alias="test", snapshot_id="954") + + assert result["snapshot"]["table"]["id"] == "in.c-foo.data" + mock_client.get_snapshot.assert_called_once_with("954") + + def test_delete_snapshots_partial_failure(self, tmp_path: Path) -> None: + """One failing delete does not block the rest (error accumulation).""" + store = _make_store(tmp_path) + mock_client = MagicMock() + mock_client.delete_snapshot.side_effect = [ + KeboolaApiError(message="Snapshot 1 not found.", status_code=404), + None, + ] + service = _make_service(store, mock_client) + + result = service.delete_snapshots(alias="test", snapshot_ids=["1", "2"]) + + assert result["deleted"] == ["2"] + assert result["failed"] == [{"id": "1", "error": "Snapshot 1 not found."}] + mock_client.close.assert_called_once() + + def test_delete_snapshots_dry_run_makes_no_calls(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + mock_client = MagicMock() + service = _make_service(store, mock_client) + + result = service.delete_snapshots(alias="test", snapshot_ids=["1", "2"], dry_run=True) + + assert result["would_delete"] == ["1", "2"] + assert result["deleted"] == [] + mock_client.delete_snapshot.assert_not_called() + + def test_table_from_snapshot_happy_path(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + mock_client = MagicMock() + mock_client.create_table_from_snapshot.return_value = { + "id": "in.c-dest.restored", + "rowsCount": 3, + } + service = _make_service(store, mock_client) + + result = service.create_table_from_snapshot( + alias="test", + bucket_id="in.c-dest", + snapshot_id="954", + name="restored", + ) + + assert result["table_id"] == "in.c-dest.restored" + assert result["dry_run"] is False + mock_client.create_table_from_snapshot.assert_called_once_with( + bucket_id="in.c-dest", snapshot_id="954", name="restored", branch_id=None + ) + mock_client.close.assert_called_once() + + def test_table_from_snapshot_dry_run_makes_no_calls(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + mock_client = MagicMock() + service = _make_service(store, mock_client) + + result = service.create_table_from_snapshot( + alias="test", + bucket_id="in.c-dest", + snapshot_id="954", + name="restored", + dry_run=True, + ) + + assert result["dry_run"] is True + mock_client.create_table_from_snapshot.assert_not_called() + + @pytest.mark.parametrize( + ("bucket_id", "snapshot_id", "name"), + [ + ("", "954", "restored"), + ("in.c-dest", " ", "restored"), + ("in.c-dest", "954", ""), + ], + ) + def test_table_from_snapshot_rejects_blank_inputs( + self, tmp_path: Path, bucket_id: str, snapshot_id: str, name: str + ) -> None: + store = _make_store(tmp_path) + service = _make_service(store, MagicMock()) + + with pytest.raises(ConfigError): + service.create_table_from_snapshot( + alias="test", bucket_id=bucket_id, snapshot_id=snapshot_id, name=name + ) + + def test_unknown_project_raises_config_error(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + service = _make_service(store, MagicMock()) + + with pytest.raises(ConfigError): + service.list_snapshots(alias="nope", table_id="in.c-foo.data") + + def test_client_closed_when_create_raises(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + mock_client = MagicMock() + mock_client.create_table_snapshot.side_effect = KeboolaApiError( + message="boom", status_code=500 + ) + service = _make_service(store, mock_client) + + with pytest.raises(KeboolaApiError): + service.create_snapshot(alias="test", table_id="in.c-foo.data") + mock_client.close.assert_called_once() + + +# --------------------------------------------------------------------------- +# CLI layer +# --------------------------------------------------------------------------- + + +class TestSnapshotCLI: + """CLI tests for the `kbagent storage snapshot*` / table-from-snapshot commands.""" + + def _project_with_active_branch(self, store: ConfigStore, branch_id: int) -> None: + config = store.load() + config.projects["test"].active_branch_id = branch_id + store.save(config) + + def test_snapshots_json_happy_path(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService") as MockSvc, + ): + MockStore.return_value = store + svc = MockSvc.return_value + svc.list_snapshots.return_value = { + "project_alias": "test", + "table_id": "in.c-foo.data", + "branch_id": None, + "count": 1, + "snapshots": [SNAPSHOT], + } + result = runner.invoke( + app, + [ + "--json", + "storage", + "snapshots", + "--project", + "test", + "--table-id", + "in.c-foo.data", + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output)["data"] + assert data["count"] == 1 + assert data["snapshots"][0]["id"] == "954" + svc.list_snapshots.assert_called_once_with( + alias="test", table_id="in.c-foo.data", limit=None, branch_id=None + ) + + def test_snapshots_ignores_active_branch(self, tmp_path: Path) -> None: + """Read command: implicit active dev branch is skipped (production endpoint).""" + store = _make_store(tmp_path) + self._project_with_active_branch(store, 42) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService") as MockSvc, + ): + MockStore.return_value = store + svc = MockSvc.return_value + svc.list_snapshots.return_value = { + "project_alias": "test", + "table_id": "in.c-foo.data", + "branch_id": None, + "count": 0, + "snapshots": [], + } + result = runner.invoke( + app, + [ + "--json", + "storage", + "snapshots", + "--project", + "test", + "--table-id", + "in.c-foo.data", + ], + ) + + assert result.exit_code == 0, result.output + assert svc.list_snapshots.call_args.kwargs["branch_id"] is None + + def test_snapshot_create_uses_active_branch(self, tmp_path: Path) -> None: + """Write command: the active dev branch set via `branch use` is honored.""" + store = _make_store(tmp_path) + self._project_with_active_branch(store, 42) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService") as MockSvc, + ): + MockStore.return_value = store + svc = MockSvc.return_value + svc.create_snapshot.return_value = { + "project_alias": "test", + "table_id": "in.c-foo.data", + "branch_id": 42, + "snapshot_id": "954", + "description": None, + } + result = runner.invoke( + app, + [ + "--json", + "storage", + "snapshot-create", + "--project", + "test", + "--table-id", + "in.c-foo.data", + ], + ) + + assert result.exit_code == 0, result.output + assert json.loads(result.output)["data"]["snapshot_id"] == "954" + assert svc.create_snapshot.call_args.kwargs["branch_id"] == 42 + + def test_snapshot_detail_json(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService") as MockSvc, + ): + MockStore.return_value = store + svc = MockSvc.return_value + svc.get_snapshot.return_value = { + "project_alias": "test", + "snapshot": {**SNAPSHOT, "table": {"id": "in.c-foo.data"}}, + } + result = runner.invoke( + app, + [ + "--json", + "storage", + "snapshot-detail", + "--project", + "test", + "--snapshot-id", + "954", + ], + ) + + assert result.exit_code == 0, result.output + assert json.loads(result.output)["data"]["snapshot"]["table"]["id"] == "in.c-foo.data" + + def test_snapshot_delete_failed_exits_1(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService") as MockSvc, + ): + MockStore.return_value = store + svc = MockSvc.return_value + svc.delete_snapshots.return_value = { + "project_alias": "test", + "deleted": [], + "failed": [{"id": "954", "error": "Snapshot 954 not found."}], + "dry_run": False, + } + result = runner.invoke( + app, + [ + "--json", + "storage", + "snapshot-delete", + "--project", + "test", + "--snapshot-id", + "954", + ], + ) + + assert result.exit_code == 1, result.output + + def test_snapshot_delete_confirm_abort(self, tmp_path: Path) -> None: + """Human mode without --yes prompts; answering 'n' aborts with exit 0.""" + store = _make_store(tmp_path) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService") as MockSvc, + ): + MockStore.return_value = store + svc = MockSvc.return_value + result = runner.invoke( + app, + ["storage", "snapshot-delete", "--project", "test", "--snapshot-id", "954"], + input="n\n", + ) + + assert result.exit_code == 0, result.output + assert "Aborted" in result.output + svc.delete_snapshots.assert_not_called() + + def test_table_from_snapshot_json_happy_path(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService") as MockSvc, + ): + MockStore.return_value = store + svc = MockSvc.return_value + svc.create_table_from_snapshot.return_value = { + "project_alias": "test", + "bucket_id": "in.c-dest", + "snapshot_id": "954", + "name": "restored", + "branch_id": None, + "dry_run": False, + "table": {"id": "in.c-dest.restored", "rowsCount": 3}, + "table_id": "in.c-dest.restored", + } + result = runner.invoke( + app, + [ + "--json", + "storage", + "table-from-snapshot", + "--project", + "test", + "--snapshot-id", + "954", + "--bucket-id", + "in.c-dest", + "--name", + "restored", + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output)["data"] + assert data["table_id"] == "in.c-dest.restored" + svc.create_table_from_snapshot.assert_called_once_with( + alias="test", + bucket_id="in.c-dest", + snapshot_id="954", + name="restored", + branch_id=None, + dry_run=False, + ) + + def test_table_from_snapshot_requires_name(self, tmp_path: Path) -> None: + """--name is required (the live API rejects an empty name).""" + store = _make_store(tmp_path) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService"), + ): + MockStore.return_value = store + result = runner.invoke( + app, + [ + "--json", + "storage", + "table-from-snapshot", + "--project", + "test", + "--snapshot-id", + "954", + "--bucket-id", + "in.c-dest", + ], + ) + + assert result.exit_code == 2, result.output + + def test_table_from_snapshot_config_error_exits_5(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.SnapshotService") as MockSvc, + ): + MockStore.return_value = store + svc = MockSvc.return_value + svc.create_table_from_snapshot.side_effect = ConfigError("unknown project") + result = runner.invoke( + app, + [ + "--json", + "storage", + "table-from-snapshot", + "--project", + "nope", + "--snapshot-id", + "954", + "--bucket-id", + "in.c-dest", + "--name", + "restored", + ], + ) + + assert result.exit_code == 5, result.output diff --git a/uv.lock b/uv.lock index 9188a691..b261cb61 100644 --- a/uv.lock +++ b/uv.lock @@ -590,7 +590,7 @@ wheels = [ [[package]] name = "keboola-cli" -version = "0.74.0" +version = "0.75.0" source = { editable = "." } dependencies = [ { name = "croniter" }, From 8fb5c67e894b4c42b81954d56b2c5ff9dbcc0eec Mon Sep 17 00:00:00 2001 From: Petr Date: Wed, 22 Jul 2026 00:23:30 +0200 Subject: [PATCH 2/6] fix(e2e): parity step parses the expected-error envelope without asserting exit 0 _test_mcp_parity_commands (added in #508) deliberately runs 'component sync-action testConnection' with a bad config and asserts exit_code != 0 -- then immediately called _json(result), which asserts exit_code == 0. The step could never pass; every full-E2E run since 0.73.0 died there. Parse the error envelope with json.loads directly. --- tests/test_e2e.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 24983b2c..cb4a3268 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -3165,7 +3165,9 @@ def _test_mcp_parity_commands(self) -> None: '{"parameters": {"db": {"host": "invalid.example.com"}}}', ) assert result.exit_code != 0 - err = _json(result) + # NOT _json(): that helper asserts exit_code == 0, but this command is + # EXPECTED to fail -- parse the error envelope directly (#508 regression). + err = json.loads(result.output) assert err["error"]["code"] in ("API_ERROR", "VALIDATION_ERROR") # flow examples — bundled, offline From 82f0a89640fe90500abb826dc3ebd816e9ef189c Mon Sep 17 00:00:00 2001 From: Petr Date: Wed, 22 Jul 2026 00:25:44 +0200 Subject: [PATCH 3/6] feat(serve): snapshot REST routes + plugin consistency per CONTRIBUTING checklist (#512) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the per-command checklist items missed in the first commit: - serve 1:1 parity: 5 snapshot routes in server/routers/storage.py (POST tables/{p}/{t}/snapshots, GET snapshots/{p}/{t}, GET snapshot-detail/{p}/{id}, DELETE snapshots/{p}, POST table-from-snapshot/{p}) + SnapshotService in ServiceRegistry; 5 router->service kwarg-parity tests in test_server_router_calls.py - new references/snapshot-workflow.md (backup -> find -> restore-as-new -> swap promote -> cleanup; the three verified traps) + SKILL.md bottom-table row + description trigger keywords (table snapshot, snapshot restore, create table from snapshot, table backup) - keboola-expert.md §3 inline gotcha (restore never overwrites; --name required; snapshot-delete forecloses restores only) -- 59.0KB, under the 62KB prompt budget --- plugins/kbagent/agents/keboola-expert.md | 4 + plugins/kbagent/skills/kbagent/SKILL.md | 7 +- .../kbagent/references/snapshot-workflow.md | 104 ++++++++++++++++ src/keboola_agent_cli/server/dependencies.py | 3 + .../server/routers/storage.py | 91 ++++++++++++++ tests/test_server_router_calls.py | 114 ++++++++++++++++++ 6 files changed, 321 insertions(+), 2 deletions(-) create mode 100644 plugins/kbagent/skills/kbagent/references/snapshot-workflow.md diff --git a/plugins/kbagent/agents/keboola-expert.md b/plugins/kbagent/agents/keboola-expert.md index 79960f80..6019fd14 100644 --- a/plugins/kbagent/agents/keboola-expert.md +++ b/plugins/kbagent/agents/keboola-expert.md @@ -254,6 +254,10 @@ read it when a trigger fires. Each `(X.Y.Z+)` tag is the version floor. "bucket not found" until you clone the prod table branch-local first. - **`truncate-table` is row-only** (0.32.0+): schema / PK / dependents survive; uniformly async-via-job on every branch; do NOT pass `async=true`. +- **Snapshot restore never overwrites** (0.75.0+, #512): `table-from-snapshot` + needs a REQUIRED `--name` (API rejects empty) and fails on an existing table + name -- restore under a new name, verify, then `swap-tables`. `snapshot-delete` + only forecloses restores; the source table is untouched. - **`bucket-detail` is dialect-aware** (0.25.3+): read `sql_dialect` + per-table `sql_path` (already correctly quoted) -- don't branch on the backend yourself. - **`set-default-bucket`** (0.26.0+) writes `storage.output.default_bucket` diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index 566aa2ca..a44bf756 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -6,7 +6,8 @@ description: > job runs, cross-project data lineage, Keboola MCP tools, development branches, SQL debugging in temporary workspaces, GitOps sync of configs as local files (pull/push/diff/clone), bucket sharing and linking, encrypting - secrets, Storage tables and files, data apps (create/deploy/logs/secrets), + secrets, Storage tables and files, table snapshots (backup + restore as a + new table), data apps (create/deploy/logs/secrets), conditional flows and schedules, project members and invitations, feature flags, OTLP data streams, scoped Storage tokens, the semantic layer (models, datasets, metrics, constraints, reference data), and the Keboola @@ -14,7 +15,8 @@ description: > keboola lineage, keboola sync, gitops, dev branch, workspace SQL, data app, streamlit deploy, semantic layer, sl, dev-portal, data stream, OTLP, scoped token, bucket sharing, encrypt secrets, feature flag, flow schedule, - invite member, SQL transformation edit, sync action, keboola docs. + invite member, SQL transformation edit, sync action, keboola docs, + table snapshot, snapshot restore, create table from snapshot, table backup. --- # kbagent -- Keboola Agent CLI @@ -382,6 +384,7 @@ For detailed response parsing rules and common pitfalls, see [gotchas](reference | **Agent Tasks via REST** (`kbagent http /agents...` from inside scheduled subprocesses; SSE streaming) | [agent-tasks-rest-workflow](references/agent-tasks-rest-workflow.md) | | **Data apps** (create / deploy / start / stop / password / delete; the §9 redeploy contract) | [data-app-workflow](references/data-app-workflow.md) | | Storage Files (upload, download, tags, load/unload) | [storage-files-workflow](references/storage-files-workflow.md) | +| **Table snapshots** (point-in-time backup; restore as a NEW table; `--name` required, no overwrite) | [snapshot-workflow](references/snapshot-workflow.md) | | **Python library** (`from keboola_agent_cli import Client` -- in-process query + Storage Files, no CLI/daemon/config-dir) | [library-workflow](references/library-workflow.md) | | **Data Streams (OTLP / OpenTelemetry)** (create/inspect OTLP source, masked secret-in-URL, OTEL_EXPORTER_OTLP_ENDPOINT) | [stream-workflow](references/stream-workflow.md) | | **Storage column types** (native types, NOT NULL, DEFAULT, branch materialize) | [storage-types-workflow](references/storage-types-workflow.md) | diff --git a/plugins/kbagent/skills/kbagent/references/snapshot-workflow.md b/plugins/kbagent/skills/kbagent/references/snapshot-workflow.md new file mode 100644 index 00000000..a881c76f --- /dev/null +++ b/plugins/kbagent/skills/kbagent/references/snapshot-workflow.md @@ -0,0 +1,104 @@ +# Table Snapshot Workflow + +Table snapshots (since v0.75.0, issue #512) are point-in-time backups of a +Storage table -- data, columns, and primary key -- stored by the platform. +A snapshot can later be restored as a **NEW** table in any existing bucket. +Typical uses: a safety net before a risky schema change or re-seed, and +"copy this table as it was" duplication. + +## Quick reference + +| Command | Purpose | Permission | +|---------|---------|------------| +| `storage snapshot-create` | Snapshot a table (async job; receipt carries `snapshot_id`) | write | +| `storage snapshots` | List a table's snapshots | read | +| `storage snapshot-detail` | One snapshot's detail (embeds the source table) | read | +| `storage table-from-snapshot` | Restore a snapshot as a NEW table | write | +| `storage snapshot-delete` | Delete snapshots (forecloses restores) | destructive | + +## Backup before a risky change + +```bash +kbagent --json storage snapshot-create \ + --project ALIAS \ + --table-id in.c-main.customers \ + --description "before typify 2026-07-22" +``` + +- Returns `snapshot_id` -- **save it**; restores are addressed by it. +- Async `tableSnapshotCreate` storage job; the CLI polls to completion. +- The snapshot captures data + columns + primary key at this moment. + Later imports/schema changes on the source table do not touch it. + +## Find a snapshot later + +```bash +kbagent --json storage snapshots --project ALIAS --table-id in.c-main.customers +kbagent --json storage snapshot-detail --project ALIAS --snapshot-id 954 +``` + +- List is per source table; entries carry `id`, `createdTime`, `description`, + `creatorToken`. +- Snapshot IDs are **global** (not table-scoped): `snapshot-detail` embeds the + source `table` object (id, columns, primaryKey), so a bare ID can always be + traced back to its origin table. + +## Restore as a new table + +```bash +kbagent --json storage table-from-snapshot \ + --project ALIAS \ + --snapshot-id 954 \ + --bucket-id in.c-main \ + --name customers_restored \ + --dry-run # preview first, then re-run without it +``` + +Three traps (all verified live): + +1. **`--name` is REQUIRED.** The API rejects an omitted/empty name + (`Table create option "name" is required and cannot be empty`). +2. **No overwrite.** Restoring onto an existing table name fails with a + duplicate-name error. Restore under a NEW name, verify the data, then + promote with `storage swap-tables` (or `delete-table` + rename pattern). +3. **The destination bucket must already exist** -- any bucket works, not + just the source one. + +The restore is an async storage job; the CLI polls to completion, so the +receipt's `table.rowsCount` is authoritative. Verify with: + +```bash +kbagent --json storage table-detail --project ALIAS --table-id in.c-main.customers_restored +``` + +## Roll a table back to a snapshot (full pattern) + +There is no in-place restore. The safe sequence: + +1. `storage table-from-snapshot ... --name customers_rollback` (new table) +2. `storage table-detail` on the new table -- verify rows/columns/PK +3. `storage swap-tables --table-id in.c-main.customers --target-table-id in.c-main.customers_rollback --branch --yes` + (any branch incl. production; aliases are NOT transferred -- see + storage-types-workflow) +4. Optionally `storage delete-table` the swapped-out copy once satisfied + +## Cleanup + +```bash +kbagent --json storage snapshot-delete --project ALIAS --snapshot-id 954 [--snapshot-id 955] [--dry-run] [--yes] +``` + +- Destructive: restores from that snapshot become impossible. The source + table is untouched. +- Batch-tolerant: one failing ID does not abort the rest; exit 1 when any + ID failed (`failed[]` in JSON). + +## Anti-patterns + +- Exporting to CSV as a "backup" -- loses column types and primary key; + snapshots keep both. +- Restoring straight onto the production table name -- duplicate-name error; + use the restore-then-swap pattern above. +- Expecting `create-table --snapshot-id` -- not a thing; the restore goes + through the classic `tables-async` endpoint, `create-table` uses + `tables-definition`, which does not accept snapshots. diff --git a/src/keboola_agent_cli/server/dependencies.py b/src/keboola_agent_cli/server/dependencies.py index eb31a582..fc4908ab 100644 --- a/src/keboola_agent_cli/server/dependencies.py +++ b/src/keboola_agent_cli/server/dependencies.py @@ -39,6 +39,7 @@ from ..services.search_service import SearchService from ..services.semantic_layer_service import SemanticLayerService from ..services.sharing_service import SharingService +from ..services.snapshot_service import SnapshotService from ..services.storage_service import StorageService from ..services.stream_service import StreamService from ..services.sync_service import SyncService @@ -67,6 +68,7 @@ class ServiceRegistry: project: ProjectService = field(init=False) config: ConfigService = field(init=False) component: ComponentService = field(init=False) + snapshot: SnapshotService = field(init=False) storage: StorageService = field(init=False) stream: StreamService = field(init=False) job: JobService = field(init=False) @@ -102,6 +104,7 @@ def __post_init__(self) -> None: self.project = ProjectService(config_store=cs) self.config = ConfigService(config_store=cs) self.component = ComponentService(config_store=cs) + self.snapshot = SnapshotService(config_store=cs) self.storage = StorageService(config_store=cs) self.stream = StreamService(config_store=cs) self.job = JobService(config_store=cs) diff --git a/src/keboola_agent_cli/server/routers/storage.py b/src/keboola_agent_cli/server/routers/storage.py index 9c2618e9..63a4bbad 100644 --- a/src/keboola_agent_cli/server/routers/storage.py +++ b/src/keboola_agent_cli/server/routers/storage.py @@ -103,6 +103,18 @@ class CloneTable(BaseModel): branch_id: int +class CreateSnapshot(BaseModel): + description: str | None = None + branch_id: int | None = None + + +class TableFromSnapshot(BaseModel): + snapshot_id: str + bucket_id: str + name: str + branch_id: int | None = None + + @router.get("/buckets", summary="List storage buckets") def list_buckets( project: str | None = None, @@ -434,6 +446,85 @@ def clone_table( ) +@router.post("/tables/{project}/{table_id:path}/snapshots", summary="Create a table snapshot") +def create_snapshot( + project: str, + table_id: str, + body: CreateSnapshot, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """Create a point-in-time snapshot of a table. Mirrors `kbagent storage snapshot-create`.""" + return registry.snapshot.create_snapshot( + alias=project, + table_id=table_id, + description=body.description, + branch_id=body.branch_id, + ) + + +@router.get("/snapshots/{project}/{table_id:path}", summary="List table snapshots") +def list_snapshots( + project: str, + table_id: str, + limit: int | None = None, + branch_id: int | None = None, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """List snapshots of a table. Mirrors `kbagent storage snapshots`.""" + return registry.snapshot.list_snapshots( + alias=project, + table_id=table_id, + limit=limit, + branch_id=branch_id, + ) + + +@router.get("/snapshot-detail/{project}/{snapshot_id}", summary="Get snapshot detail") +def snapshot_detail( + project: str, + snapshot_id: str, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """One snapshot's detail (embeds the source table). Mirrors `kbagent storage snapshot-detail`.""" + return registry.snapshot.get_snapshot(alias=project, snapshot_id=snapshot_id) + + +@router.delete("/snapshots/{project}", summary="Delete snapshots") +def delete_snapshots( + project: str, + snapshot_id: list[str] = Query(...), + dry_run: bool = False, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """Delete one or more snapshots (batch-tolerant). Mirrors `kbagent storage snapshot-delete`.""" + return registry.snapshot.delete_snapshots( + alias=project, + snapshot_ids=snapshot_id, + dry_run=dry_run, + ) + + +@router.post("/table-from-snapshot/{project}", summary="Create a NEW table from a snapshot") +def table_from_snapshot( + project: str, + body: TableFromSnapshot, + dry_run: bool = False, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """Restore a snapshot as a NEW table (data + columns + primary key). + + Mirrors `kbagent storage table-from-snapshot`. + """ + return registry.snapshot.create_table_from_snapshot( + alias=project, + bucket_id=body.bucket_id, + snapshot_id=body.snapshot_id, + name=body.name, + branch_id=body.branch_id, + dry_run=dry_run, + ) + + @router.post("/tables/{project}/{table_id:path}/describe", summary="Set table description") def describe_table( project: str, diff --git a/tests/test_server_router_calls.py b/tests/test_server_router_calls.py index 0f7c778d..380f764a 100644 --- a/tests/test_server_router_calls.py +++ b/tests/test_server_router_calls.py @@ -1564,3 +1564,117 @@ def test_flow_examples_unknown_component_is_400(tmp_path: Path) -> None: assert res.status_code == 400, res.text assert "No bundled flow examples" in res.json()["error"]["message"] + + +# --------------------------------------------------------------------------- +# storage.py snapshot routes (issue #512) +# Service: registry.snapshot (SnapshotService) -- create_snapshot, +# list_snapshots, get_snapshot, delete_snapshots, create_table_from_snapshot. +# --------------------------------------------------------------------------- + + +def test_snapshot_create_passes_kwargs(tmp_path: Path) -> None: + """POST /storage/tables/{p}/{table}/snapshots -> snapshot.create_snapshot.""" + snapshot_svc = MagicMock() + snapshot_svc.create_snapshot.return_value = {"snapshot_id": "954"} + registry = _mock_registry(snapshot=snapshot_svc) + app = _make_app_with_registry(tmp_path, registry) + + with TestClient(app) as client: + res = client.post( + f"/storage/tables/{PROJECT}/{TABLE_ID}/snapshots", + headers=AUTH, + json={"description": "before migration", "branch_id": 42}, + ) + + assert res.status_code == 200, res.text + snapshot_svc.create_snapshot.assert_called_once_with( + alias=PROJECT, + table_id=TABLE_ID, + description="before migration", + branch_id=42, + ) + + +def test_snapshot_list_passes_kwargs(tmp_path: Path) -> None: + """GET /storage/snapshots/{p}/{table} -> snapshot.list_snapshots.""" + snapshot_svc = MagicMock() + snapshot_svc.list_snapshots.return_value = {"count": 0, "snapshots": []} + registry = _mock_registry(snapshot=snapshot_svc) + app = _make_app_with_registry(tmp_path, registry) + + with TestClient(app) as client: + res = client.get( + f"/storage/snapshots/{PROJECT}/{TABLE_ID}", + headers=AUTH, + params={"limit": 5}, + ) + + assert res.status_code == 200, res.text + snapshot_svc.list_snapshots.assert_called_once_with( + alias=PROJECT, + table_id=TABLE_ID, + limit=5, + branch_id=None, + ) + + +def test_snapshot_detail_passes_kwargs(tmp_path: Path) -> None: + """GET /storage/snapshot-detail/{p}/{id} -> snapshot.get_snapshot.""" + snapshot_svc = MagicMock() + snapshot_svc.get_snapshot.return_value = {"snapshot": {"id": "954"}} + registry = _mock_registry(snapshot=snapshot_svc) + app = _make_app_with_registry(tmp_path, registry) + + with TestClient(app) as client: + res = client.get(f"/storage/snapshot-detail/{PROJECT}/954", headers=AUTH) + + assert res.status_code == 200, res.text + snapshot_svc.get_snapshot.assert_called_once_with(alias=PROJECT, snapshot_id="954") + + +def test_snapshot_delete_passes_kwargs(tmp_path: Path) -> None: + """DELETE /storage/snapshots/{p}?snapshot_id=... -> snapshot.delete_snapshots.""" + snapshot_svc = MagicMock() + snapshot_svc.delete_snapshots.return_value = {"deleted": ["954"], "failed": []} + registry = _mock_registry(snapshot=snapshot_svc) + app = _make_app_with_registry(tmp_path, registry) + + with TestClient(app) as client: + res = client.delete( + f"/storage/snapshots/{PROJECT}", + headers=AUTH, + params={"snapshot_id": ["954", "955"], "dry_run": True}, + ) + + assert res.status_code == 200, res.text + snapshot_svc.delete_snapshots.assert_called_once_with( + alias=PROJECT, + snapshot_ids=["954", "955"], + dry_run=True, + ) + + +def test_table_from_snapshot_passes_kwargs(tmp_path: Path) -> None: + """POST /storage/table-from-snapshot/{p} -> snapshot.create_table_from_snapshot.""" + snapshot_svc = MagicMock() + snapshot_svc.create_table_from_snapshot.return_value = {"table_id": "in.c-dest.restored"} + registry = _mock_registry(snapshot=snapshot_svc) + app = _make_app_with_registry(tmp_path, registry) + + with TestClient(app) as client: + res = client.post( + f"/storage/table-from-snapshot/{PROJECT}", + headers=AUTH, + json={"snapshot_id": "954", "bucket_id": "in.c-dest", "name": "restored"}, + ) + + assert res.status_code == 200, res.text + snapshot_svc.create_table_from_snapshot.assert_called_once_with( + alias=PROJECT, + bucket_id="in.c-dest", + snapshot_id="954", + name="restored", + branch_id=None, + dry_run=False, + ) From 4f05a519361ae087d87cae4e3c650dc7885402a2 Mon Sep 17 00:00:00 2001 From: Petr Date: Wed, 22 Jul 2026 00:28:45 +0200 Subject: [PATCH 4/6] chore(skill): keep SKILL.md description under the 1024-char Claude Desktop cap Adding the snapshot trigger keywords pushed the description to 1145 chars (the cap check in test_skill_frontmatter guards issue #447). Compress stale phrasing (dev branches, flows and schedules, Developer Portal, drop 'datasets' from the semantic-layer parenthetical) instead of dropping the new triggers; lands at 1013 chars. --- plugins/kbagent/skills/kbagent/SKILL.md | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index a44bf756..84e64dbe 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -2,21 +2,20 @@ name: kbagent description: > Use when working with Keboola Connection projects via the kbagent CLI. - Covers: exploring and searching component configurations, job history and - job runs, cross-project data lineage, Keboola MCP tools, development - branches, SQL debugging in temporary workspaces, GitOps sync of configs as - local files (pull/push/diff/clone), bucket sharing and linking, encrypting - secrets, Storage tables and files, table snapshots (backup + restore as a - new table), data apps (create/deploy/logs/secrets), - conditional flows and schedules, project members and invitations, feature + Covers: exploring and searching configurations, job history and runs, + cross-project data lineage, Keboola MCP tools, dev branches, SQL debugging in workspaces, GitOps sync of configs + (pull/push/diff/clone), bucket sharing and linking, encrypting + secrets, Storage tables, files, and snapshots (backup + restore as a new + table), data apps (create/deploy/logs/secrets), + flows and schedules, members and invitations, feature flags, OTLP data streams, scoped Storage tokens, the semantic layer - (models, datasets, metrics, constraints, reference data), and the Keboola - Developer Portal. Triggers: kbagent, Keboola, keboola config, keboola job, + (models, metrics, constraints, reference data), and the Developer + Portal. Triggers: kbagent, Keboola, keboola config, keboola job, keboola lineage, keboola sync, gitops, dev branch, workspace SQL, data app, streamlit deploy, semantic layer, sl, dev-portal, data stream, OTLP, scoped token, bucket sharing, encrypt secrets, feature flag, flow schedule, invite member, SQL transformation edit, sync action, keboola docs, - table snapshot, snapshot restore, create table from snapshot, table backup. + table snapshot, snapshot restore. --- # kbagent -- Keboola Agent CLI From 8957708f8ac159dc2cca7ef724072aeef19c3321 Mon Sep 17 00:00:00 2001 From: Petr Date: Wed, 22 Jul 2026 00:49:36 +0200 Subject: [PATCH 5/6] fix(e2e): poll table-detail after add/delete-column (read-after-DDL staleness) Step 40 asserted the new/removed column on the FIRST table-detail read after the DDL receipt. Measured live on us-east4.gcp (2026-07-22): the column listing is read-after-write eventually consistent -- the add-column receipt is authoritative but an immediate table-detail can serve stale metadata for ~5-10s (wider when the table recently had snapshot activity; reproduced 2x in-suite and 1x standalone, resolved by itself in every case). New _poll_columns helper polls up to ~30s before asserting, on both the add and the delete side. --- tests/test_e2e.py | 46 ++++++++++++++++++++++++++++++---------------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index cb4a3268..4657726f 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -3296,6 +3296,32 @@ def _test_config_delete(self, config_id: str) -> None: # Remove from cleanup since we deleted via CLI self._created_config_ids.remove((TEST_COMPONENT_ID, config_id)) + def _poll_columns( + self, table_id: str, *, present: str | None = None, absent: str | None = None + ) -> list[str]: + """Poll table-detail until a column appears/disappears (max ~30s). + + The Storage API's column listing is read-after-write eventually + consistent on some stacks: an add-column/delete-column receipt is + authoritative (the API confirmed the DDL), but an immediately-following + table-detail can serve stale metadata for a few seconds -- measured + live on connection.us-east4.gcp.keboola.com 2026-07-22 (~5-10s lag, + wider when the table recently had snapshot activity). Poll instead of + asserting the first read. + """ + columns: list[str] = [] + for _ in range(15): + data = self._run_ok( + "storage", "table-detail", "--project", self.alias, "--table-id", table_id + ) + columns = data["data"]["columns"] + if (present is None or present in columns) and ( + absent is None or absent not in columns + ): + return columns + time.sleep(2) + return columns + def _test_delete_column(self, table_id: str) -> None: """Delete a column from a table: dry-run, actual delete, verify.""" # Verify the table has 'value' column before we delete it @@ -3324,12 +3350,8 @@ def _test_delete_column(self, table_id: str) -> None: assert data["data"]["column"] == "status" assert data["data"]["definition"]["type"] == "VARCHAR" assert data["data"]["table_id"] == table_id - data = self._run_ok( - "storage", "table-detail", "--project", self.alias, "--table-id", table_id - ) - assert "status" in data["data"]["columns"], ( - f"Expected 'status' column after add-column, got {data['data']['columns']}" - ) + columns = self._poll_columns(table_id, present="status") + assert "status" in columns, f"Expected 'status' column after add-column, got {columns}" # delete-column dry-run data = self._run_ok( @@ -3363,16 +3385,8 @@ def _test_delete_column(self, table_id: str) -> None: assert data["data"]["failed"] == [] assert data["data"]["table_id"] == table_id - # Verify the column is gone - data = self._run_ok( - "storage", - "table-detail", - "--project", - self.alias, - "--table-id", - table_id, - ) - columns_after = data["data"]["columns"] + # Verify the column is gone (poll: same read-after-DDL staleness as add) + columns_after = self._poll_columns(table_id, absent="value") assert "value" not in columns_after, ( f"'value' column should be deleted, got {columns_after}" ) From c44876b1bb206ab3f6d3da3415b9f537eba5a47c Mon Sep 17 00:00:00 2001 From: Petr Date: Wed, 22 Jul 2026 01:06:06 +0200 Subject: [PATCH 6/6] refactor(storage): type _handle_errors as NoReturn (review NIT-1) Matches the identical helper in commands/token.py, feature.py, stream.py. --- src/keboola_agent_cli/commands/_storage_snapshots.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/keboola_agent_cli/commands/_storage_snapshots.py b/src/keboola_agent_cli/commands/_storage_snapshots.py index 02222c5f..54fbf877 100644 --- a/src/keboola_agent_cli/commands/_storage_snapshots.py +++ b/src/keboola_agent_cli/commands/_storage_snapshots.py @@ -14,7 +14,7 @@ from __future__ import annotations -from typing import Any +from typing import Any, NoReturn import typer from rich.markup import escape @@ -31,7 +31,7 @@ _SNAPSHOTS = "Snapshots" -def _handle_errors(formatter: Any, exc: Exception) -> None: +def _handle_errors(formatter: Any, exc: Exception) -> NoReturn: """Map ConfigError / KeboolaApiError to a structured error + Exit.""" if isinstance(exc, ConfigError): formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR)