diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index d2a4da47..acc933ec 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -10,7 +10,7 @@ "plugins": [ { "name": "kbagent", - "version": "0.61.1", + "version": "0.62.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 09621aab..2119a878 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -319,7 +319,8 @@ kbagent storage table-detail --project NAME --table-id ID [--branch ID] kbagent storage create-bucket --project NAME --stage STAGE --name NAME [--description D] [--backend B] [--branch ID] kbagent storage create-table --project NAME --bucket-id ID --name NAME --column COL:TYPE[(length)] [...] [--primary-key COL] [--not-null COL ...] [--default NAME=VALUE ...] [--branch ID] [--if-not-exists] kbagent storage upload-table --project NAME --table-id ID --file PATH [--incremental] [--branch ID] -kbagent storage download-table --project NAME --table-id ID [--output FILE] [--columns COL ...] [--limit N] [--branch ID] +kbagent storage download-table --project NAME --table-id ID [--output FILE] [--columns COL ...] [--limit N] [--where-column COL --where-value VAL ... [--where-operator eq|neq]] [--changed-since WHEN] [--changed-until WHEN] [--branch ID] +kbagent storage add-column --project NAME --table-id ID --column COL:TYPE[(length)] [--not-null] [--default VALUE] [--branch ID] kbagent storage delete-table --project NAME --table-id ID [--table-id ...] [--force] [--dry-run] [--yes] [--branch ID] kbagent storage truncate-table --project NAME --table-id ID [--table-id ...] [--dry-run] [--yes] [--branch ID] kbagent storage delete-column --project NAME --table-id ID --column COL [--column ...] [--force] [--dry-run] [--yes] [--branch ID] diff --git a/plugins/kbagent/.claude-plugin/plugin.json b/plugins/kbagent/.claude-plugin/plugin.json index b5ab10a7..13b90145 100644 --- a/plugins/kbagent/.claude-plugin/plugin.json +++ b/plugins/kbagent/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "kbagent", - "version": "0.61.1", + "version": "0.62.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 427fbccf..5b8d80bf 100644 --- a/plugins/kbagent/agents/keboola-expert.md +++ b/plugins/kbagent/agents/keboola-expert.md @@ -97,10 +97,12 @@ a critical failure. | Cross-project migration | `kbagent sync pull` + edit files locally + `kbagent sync push --dry-run` | -- | repeated `tool call` loops, one per resource | | Retype table columns | fetch types via `workspace query`, draft types YAML, write new transformation that produces typed output table, then `kbagent storage swap-tables` (0.28.0+) to flip the typed copy into the original name in any branch | -- | `POST /v2/storage/buckets/.../tables-definition` (REST) followed by manual config rewrites | | Create typed table with native types | `kbagent storage create-table --column pk:VARCHAR(40) --column amount:NUMBER(18,2) --not-null pk --default amount=0` (0.25.0+) | `tool call create_table` (accepts the same `definition.length` shape via MCP) | re-creating via raw REST to `/v2/storage/...tables-definition` | +| Add one column to an existing table | `kbagent storage add-column --project P --table-id in.c-foo.data --column status:VARCHAR(20) [--not-null] [--default active]` (0.62.0+) -- synchronous Storage endpoint, same `name:TYPE(length)` grammar as `create-table`; the add-side mirror of `delete-column` | -- | re-creating the whole table just to add a field (loses data/PK/dependents); raw `POST /v2/storage/tables/.../columns` | | 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) | | 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) | | 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 | | Run Keboola SQL or read/write Storage Files from INSIDE a Python process you control (Data App, transformation, hosted service) | `from keboola_agent_cli import Client` (0.61.0+) -- stateless `Client(url, token)`; `.query(workspace_id, sql) -> list[dict]`, `.files.upload(path_or_bytes)` / `.files.read_bytes(id) -> bytes` / `.files.list() -> [FileEntry]`; no CLI subprocess, no `serve`, no config-dir | the `kbagent` CLI or `kbagent serve` REST when you are NOT already inside Python | shelling out to the `kbagent` binary from a Python process you control (import the library instead); using it for AI-driven exploration (it is fixed typed ops, not MCP tools) | | Inspect dev branch | `kbagent branch list --project P`, `kbagent branch use --project P --branch ID` | `tool call get_branch` | acting on `main` when a dev branch exists | | Audit project capabilities / features | `kbagent project info --project P` (0.30.0+) -- returns project ID, name, backend, enabled features, quota limits, and metrics | `tool call verify_token` (returns less structured info; no feature list) | inspecting the UI project settings manually | diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index ec7ba3c0..9814b0fd 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -193,6 +193,7 @@ When working inside a git repository or project directory, run `kbagent init` (o | Export a storage table to a local CSV file | `kbagent storage download-table --project PROJECT --table-id TABLE-ID` | | Delete one or more storage tables | `kbagent storage delete-table --project PROJECT --table-id TABLE-ID` | | Truncate (delete all rows from) one or more storage tables | `kbagent storage truncate-table --project PROJECT --table-id TABLE-ID` | +| Add a single column to an existing table (synchronous, typed) | `kbagent storage add-column --project PROJECT --table-id TABLE-ID --column COLUMN` | | Delete one or more columns from a storage table | `kbagent storage delete-column --project PROJECT --table-id TABLE-ID --column COLUMN` | | Swap two storage tables (any branch, including the default/production branch) | `kbagent storage swap-tables --project PROJECT --table-id TABLE-ID --target-table-id TARGET-TABLE-ID` | | Clone (pull) a production table into a development branch | `kbagent storage clone-table --project PROJECT --table-id TABLE-ID` | diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 36995df6..b01012ac 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -105,7 +105,8 @@ Requires a **super-admin** Manage API token (same kind as `org setup`). Same def - `storage create-bucket --project NAME --stage STAGE --name NAME [--description D] [--backend B] [--branch ID]` -- create bucket (branch-aware). With `--branch ID` on a project lacking the `storage-branches` feature (legacy fake-branch), response carries `legacy_branch_storage: true` and human mode prints a warning -- the runner will create a parallel `out.c--*` bucket at job time. See `storage-types-workflow.md` - `storage create-table --project NAME --bucket-id ID --name NAME --column col:TYPE[(length)] [...] [--primary-key COL] [--not-null COL ...] [--default NAME=VALUE ...] [--branch ID] [--if-not-exists]` -- create typed table. Base types `STRING/INTEGER/NUMERIC/FLOAT/BOOLEAN/DATE/TIMESTAMP` plus native backend types with length (`VARCHAR(40)`, `NUMBER(18,2)`, `TIMESTAMP_TZ`, `VARIANT`, etc.) -- type/length validation delegated to the Storage API. `--not-null` marks a column `nullable=false`; `--default NAME=VALUE` sets a DEFAULT expression (booleans must be lowercase `true`/`false`). In a dev branch, the target bucket is auto-materialized if it has not yet been written to there -- response surfaces this via `auto_created_bucket: bool`. On legacy fake-branch projects (no `storage-branches` feature), `legacy_branch_storage: true` flags that the runner will use a separate `out.c--*` bucket at job time. `--if-not-exists` (0.47.0+) turns a duplicate-display-name failure into `action: skipped` when the table really exists at the expected id (safe for parallel workers). Since 0.47.1 the skipped envelope reports the EXISTING table's actual `columns`/`primary_key`/`name`, mirrors the request under `requested_columns`/`requested_primary_key`, and sets `schema_drift: true` when they diverge. See `storage-types-workflow.md` - `storage upload-table --project NAME --table-id ID --file PATH [--incremental] [--branch ID]` -- upload CSV (branch-aware) -- `storage download-table --project NAME --table-id ID [--output FILE] [--columns COL ...] [--limit N] [--branch ID]` -- export table to CSV (branch-aware) +- `storage download-table --project NAME --table-id ID [--output FILE] [--columns COL ...] [--limit N] [--where-column COL --where-value VAL ... [--where-operator eq|neq]] [--changed-since WHEN] [--changed-until WHEN] [--branch ID]` -- export table to CSV (branch-aware). `--where-column` + `--where-value` (repeatable, OR within the set) + `--where-operator eq|neq` filter rows server-side; `--changed-since`/`--changed-until` (unix ts or strtotime like `-2 days`) filter by import time -- the credential-only, no-workspace way to pull a filtered/incremental slice (0.62.0+) +- `storage add-column --project NAME --table-id ID --column COL:TYPE[(length)] [--not-null] [--default VALUE] [--branch ID]` -- add a single column to an existing table (0.62.0+). Same `name:TYPE(length)` grammar as `create-table --column`; a bare `name` adds an untyped STRING column. Synchronous endpoint (no job to wait on). `--not-null` needs an empty table or a `--default`. Mirror of `delete-column` - `storage delete-table --project NAME --table-id ID [--table-id ...] [--force] [--dry-run] [--yes] [--branch ID]` -- delete tables, --force cascade-deletes aliased tables (branch-aware) - `storage truncate-table --project NAME --table-id ID [--table-id ...] [--dry-run] [--yes] [--branch ID]` (since v0.32.0) -- delete all rows while preserving table schema, primary key, descriptions, sharing edges, and downstream dependents. Batch via repeated `--table-id`. Endpoint is uniformly async-via-job on every branch (returns a queued `tableRowsDelete` job; client polls via `_wait_for_storage_job` before returning). Idempotent (truncating an empty table is a no-op). Use when re-seeding a table without losing the schema contract - `storage delete-column --project NAME --table-id ID --column COL [--column ...] [--force] [--dry-run] [--yes] [--branch ID]` -- delete columns from a table (branch-aware) diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index 90280577..56be166e 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -2682,3 +2682,21 @@ is a thin wrapper, not a workspace manager. Three non-obvious behaviors: `"true"` -- with SQL `NULL` as `None`. The facade is transparent and does not coerce, so callers must cast (`int(row["x"])` etc.) for typed values. (Verified live against a Snowflake workspace; BigQuery behavior not yet verified.) + +### `storage download-table` row filters send `whereValues[]` (array notation) (since v0.62.0) + +`--where-column` + `--where-value` + `--where-operator eq|neq` and +`--changed-since` / `--changed-until` filter the export server-side. If you call +the raw Storage API instead of the CLI, the values parameter is `whereValues[]` +(WITH the bracket suffix), not `whereValues`; `whereColumn`/`whereOperator` and +`changedSince`/`changedUntil` are plain. Repeat `--where-value` for an OR set. +This is the credential-only, no-workspace way to pull a filtered/incremental +slice -- `workspace query` with a `WHERE` clause needs a live workspace. + +### `storage add-column --not-null` needs an empty table or `--default` (since v0.62.0) + +`storage add-column --column name:TYPE(length) [--not-null] [--default VALUE]` +hits the SYNCHRONOUS Storage endpoint (no job to poll). `--not-null` on a table +that ALREADY HAS ROWS is rejected by the backend with an API error (not a local +validation error) unless you also pass `--default` -- the existing rows need a +value for the new non-null column. Add `--default` when the table is non-empty. diff --git a/pyproject.toml b/pyproject.toml index 185601a4..83d099fa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "keboola-agent-cli" -version = "0.61.1" +version = "0.62.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 1da88a1d..8ba6615b 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -24,6 +24,23 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { + "0.62.0": [ + "New (#417): `storage download-table` gains server-side row filtering -- " + "`--where-column` + `--where-value` (repeatable, OR within the set) + " + "`--where-operator eq|neq`, plus `--changed-since` / `--changed-until` (unix ts " + "or strtotime like `-2 days`) to export only rows imported in a time window. " + "This is the credential-only, no-workspace way to pull a filtered or incremental " + "slice of a table -- the Query Service path needs a live workspace. The filters " + "thread through both `export_table_async` and `get_table_data_preview` via a " + "shared `_apply_table_filters` helper, so the sync-preview and async-export " + "endpoints honor an identical contract.", + "New (#417): `storage add-column` adds a single column to an existing table, " + "using the same `name:TYPE(length)` grammar as `create-table --column` (with " + "`--not-null` and `--default`). This closes a long-standing asymmetry -- kbagent " + "could drop a column (`delete-column`) but not add one. The Storage add-column " + "endpoint is synchronous (no job to wait on); the operation is classified `write` " + "in the permission registry.", + ], "0.61.1": [ "Note (#416 follow-up): clarified the value-typing contract of the importable " "`Client.query()`. The Query Service `/results` endpoint serializes Snowflake " diff --git a/src/keboola_agent_cli/client.py b/src/keboola_agent_cli/client.py index ebd028ea..cb2ec191 100644 --- a/src/keboola_agent_cli/client.py +++ b/src/keboola_agent_cli/client.py @@ -1921,11 +1921,62 @@ def list_tables_with_metadata(self) -> list[dict[str, Any]]: """ return self.list_tables(include="columns,metadata,buckets") + @staticmethod + def _apply_table_filters( + params: dict[str, Any], + *, + where_column: str | None = None, + where_operator: str = "eq", + where_values: list[str] | None = None, + changed_since: str | None = None, + changed_until: str | None = None, + ) -> None: + """Mutate ``params`` with Storage table export/preview filter clauses. + + Shared by :meth:`get_table_data_preview` and :meth:`export_table_async` + so the ``whereColumn`` / ``whereOperator`` / ``whereValues[]`` and + ``changedSince`` / ``changedUntil`` contract is identical across the + sync-preview and async-export endpoints. + + Args: + where_column: Column to filter on. Must be paired with ``where_values``. + where_operator: ``"eq"`` or ``"neq"`` (only meaningful with a filter). + where_values: Values the column is matched against (OR within the set). + changed_since: Lower bound on import time -- a unix timestamp or a + strtotime string like ``"-2 days"``. + changed_until: Upper bound on import time (same formats). + + Raises: + ValueError: On an invalid ``where_operator`` or a half-specified + where-clause (a column without values, or values without a column). + """ + if (where_column is None) != (where_values is None): + raise ValueError( + "where_column and where_values must be given together " + "(the column to match and the values to match it against)." + ) + if where_column is not None: + if where_operator not in ("eq", "neq"): + raise ValueError(f"where_operator must be 'eq' or 'neq', got {where_operator!r}.") + params["whereColumn"] = where_column + params["whereOperator"] = where_operator + params["whereValues[]"] = where_values + if changed_since is not None: + params["changedSince"] = changed_since + if changed_until is not None: + params["changedUntil"] = changed_until + def get_table_data_preview( self, table_id: str, limit: int = 100, columns: list[str] | None = None, + *, + where_column: str | None = None, + where_operator: str = "eq", + where_values: list[str] | None = None, + changed_since: str | None = None, + changed_until: str | None = None, ) -> str: """Get a CSV preview of table data. @@ -1934,6 +1985,11 @@ def get_table_data_preview( limit: Max number of rows to return. columns: Optional list of column names to export. Storage API limits sync export to 30 columns max. + where_column: Filter to rows where this column matches ``where_values``. + where_operator: ``"eq"`` (default) or ``"neq"``. + where_values: Values for the ``where_column`` filter. + changed_since: Only rows imported since this time (unix ts / strtotime). + changed_until: Only rows imported up to this time. Returns: CSV string with table data preview. @@ -1942,6 +1998,14 @@ def get_table_data_preview( params: dict[str, Any] = {"limit": limit} if columns: params["columns"] = ",".join(columns) + self._apply_table_filters( + params, + where_column=where_column, + where_operator=where_operator, + where_values=where_values, + changed_since=changed_since, + changed_until=changed_until, + ) response = self._request( "GET", f"/v2/storage/tables/{safe_id}/data-preview", @@ -1956,6 +2020,12 @@ def export_table_async( limit: int | None = None, branch_id: int | None = None, file_type: str = "csv", + *, + where_column: str | None = None, + where_operator: str = "eq", + where_values: list[str] | None = None, + changed_since: str | None = None, + changed_until: str | None = None, ) -> dict[str, Any]: """Start an async table export and wait for completion. @@ -1967,6 +2037,11 @@ def export_table_async( file_type: Output format, either "csv" (default) or "parquet". Parquet exports are always sliced and Snappy-compressed inside the parquet format (not gzipped at the slice level). + where_column: Filter to rows where this column matches ``where_values``. + where_operator: ``"eq"`` (default) or ``"neq"``. + where_values: Values for the ``where_column`` filter. + changed_since: Only rows imported since this time (unix ts / strtotime). + changed_until: Only rows imported up to this time. Returns: Completed export job dict (results contain file info). @@ -1980,6 +2055,14 @@ def export_table_async( params["columns"] = ",".join(columns) if limit is not None: params["limit"] = str(limit) + self._apply_table_filters( + params, + where_column=where_column, + where_operator=where_operator, + where_values=where_values, + changed_since=changed_since, + changed_until=changed_until, + ) response = self._request( "POST", f"{prefix}/tables/{safe_id}/export-async", @@ -1987,6 +2070,38 @@ def export_table_async( ) return self._wait_for_storage_job(response.json(), max_wait=EXPORT_JOB_MAX_WAIT) + def add_column( + self, + table_id: str, + name: str, + definition: dict[str, Any] | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Add a single column to an existing table (synchronous). + + Unlike ``delete_column`` (async storage job), the Storage API + ``POST /tables/{id}/columns`` endpoint is synchronous and returns the + updated table resource directly -- there is no job to poll. + + Args: + table_id: Full table ID (e.g. "in.c-bucket.table"). + name: Name of the new column. + definition: Optional typed-column definition for a typed table, e.g. + ``{"type": "NUMBER", "length": "18,2", "nullable": False, + "default": "0"}``. Omit for an untyped column. + branch_id: If set, target a specific dev branch. + + Returns: + The updated table resource dict from the API. + """ + prefix = f"/v2/storage/branch/{branch_id}" if branch_id else "/v2/storage" + safe_id = quote(table_id, safe="") + body: dict[str, Any] = {"name": name} + if definition: + body["definition"] = definition + response = self._request("POST", f"{prefix}/tables/{safe_id}/columns", json=body) + return response.json() + def get_file_info(self, file_id: int, branch_id: int | None = None) -> dict[str, Any]: """Get file metadata including download URL. diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index e2d35657..b4131801 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -377,10 +377,13 @@ Use --no-auto-create to require the table to already exist. Full load by default; --incremental to append rows. Supports files up to 5 GB via async file-first upload flow. Branch-aware. - kbagent storage download-table --project NAME --table-id TABLE_ID [--output FILE] [--columns COL ...] [--limit N] [--branch ID] + kbagent storage download-table --project NAME --table-id TABLE_ID [--output FILE] [--columns COL ...] [--limit N] [--where-column COL --where-value VAL ... [--where-operator eq|neq]] [--changed-since WHEN] [--changed-until WHEN] [--branch ID] Export table data to a local CSV file. Async export with streaming download. + --where-column + --where-value (repeatable) + --where-operator eq|neq filter rows; --changed-since/--changed-until (unix ts or strtotime) filter by import time. Default filename: TABLE_NAME.csv. Use --columns to select columns (see table-detail for names). Use --limit to cap row count. Handles sliced files and gzip decompression transparently. Branch-aware. + kbagent storage add-column --project NAME --table-id ID --column COL:TYPE[(length)] [--not-null] [--default VALUE] [--branch ID] + Add a single column to an existing table (synchronous). Same name:TYPE(length) grammar as create-table --column. kbagent storage delete-table --project NAME --table-id ID [--table-id ...] [--force] [--dry-run] [--yes] [--branch ID] Delete one or more tables. Batch: repeat --table-id. --force to cascade-delete aliased tables. --dry-run to preview. Branch-aware. diff --git a/src/keboola_agent_cli/commands/storage.py b/src/keboola_agent_cli/commands/storage.py index d5ac9906..1a9571a5 100644 --- a/src/keboola_agent_cli/commands/storage.py +++ b/src/keboola_agent_cli/commands/storage.py @@ -811,6 +811,31 @@ def storage_download_table( "polars, Spark. A _columns.csv sidecar holds the column order." ), ), + where_column: str | None = typer.Option( + None, + "--where-column", + help="Export only rows where this column matches --where-value(s).", + ), + where_operator: str = typer.Option( + "eq", + "--where-operator", + help="Filter operator: 'eq' (default) or 'neq'.", + ), + where_value: list[str] | None = typer.Option( + None, + "--where-value", + help="Value(s) for --where-column (repeat for multiple: matched as OR).", + ), + changed_since: str | None = typer.Option( + None, + "--changed-since", + help="Only rows imported since this time (unix ts or strtotime, e.g. '-2 days').", + ), + changed_until: str | None = typer.Option( + None, + "--changed-until", + help="Only rows imported up to this time (unix ts or strtotime).", + ), ) -> None: """Export a storage table to a local CSV file. @@ -844,6 +869,11 @@ def storage_download_table( limit=limit, branch_id=effective_branch, keep_slices=keep_slices, + where_column=where_column, + where_operator=where_operator, + where_values=where_value, + changed_since=changed_since, + changed_until=changed_until, ) except ValueError as exc: formatter.error(message=str(exc), error_code=ErrorCode.INVALID_ARGUMENT) @@ -1080,6 +1110,83 @@ def storage_truncate_table( raise typer.Exit(code=1) +@storage_app.command("add-column", rich_help_panel=_TABLES) +def storage_add_column( + ctx: typer.Context, + project: str = typer.Option( + ..., + "--project", + help="Project alias", + ), + table_id: str = typer.Option( + ..., + "--table-id", + help="Table ID to add the column to (e.g. 'in.c-bucket.table')", + ), + column: str = typer.Option( + ..., + "--column", + help=( + "Column spec: 'name', 'name:TYPE', or 'name:TYPE(length)' " + "(e.g. 'status:VARCHAR(20)', 'amount:NUMBER(18,2)')." + ), + ), + not_null: bool = typer.Option( + False, + "--not-null", + help="Make the new column NOT NULL (needs an empty table or a --default).", + ), + default: str | None = typer.Option( + None, + "--default", + help="Default value for the new column.", + ), + branch: int | None = typer.Option( + None, + "--branch", + help="Dev branch ID (defaults to active branch if set via 'branch use')", + ), +) -> None: + """Add a single column to an existing table (synchronous, typed). + + Mirrors ``create-table --column``: ``name:TYPE(length)`` creates a typed + column; a bare ``name`` adds an untyped STRING column. The Storage + add-column endpoint is synchronous -- there is no job to wait on. + """ + formatter = get_formatter(ctx) + service = get_service(ctx, "storage_service") + config_store: ConfigStore = ctx.obj["config_store"] + _, effective_branch = resolve_branch(config_store, formatter, project, branch) + + try: + result = service.add_column( + alias=project, + table_id=table_id, + column=column, + not_null=not_null, + default=default, + branch_id=effective_branch, + ) + except ValueError as exc: + formatter.error(message=str(exc), error_code=ErrorCode.INVALID_ARGUMENT) + raise typer.Exit(code=2) from None + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR) + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + 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 + + if formatter.json_mode: + formatter.output(result) + else: + col_type = result["definition"].get("type", "STRING") + formatter.console.print( + f"[bold green]Added column:[/bold green] {result['column']} " + f"({col_type}) to {result['table_id']}" + ) + + @storage_app.command("delete-column", rich_help_panel=_TABLES) def storage_delete_column( ctx: typer.Context, diff --git a/src/keboola_agent_cli/permissions.py b/src/keboola_agent_cli/permissions.py index 733de02a..e197fe5a 100644 --- a/src/keboola_agent_cli/permissions.py +++ b/src/keboola_agent_cli/permissions.py @@ -183,6 +183,7 @@ # Storage write "storage.create-bucket": "write", "storage.create-table": "write", + "storage.add-column": "write", "storage.upload-table": "write", # clone-table pulls a prod table into a dev branch (materialization); it # creates a branch-local copy and never deletes -- write, not destructive. diff --git a/src/keboola_agent_cli/server/routers/storage.py b/src/keboola_agent_cli/server/routers/storage.py index ef18de87..f4c5b215 100644 --- a/src/keboola_agent_cli/server/routers/storage.py +++ b/src/keboola_agent_cli/server/routers/storage.py @@ -174,6 +174,11 @@ def preview_table_v2( table_id: str, limit: int = 100, columns: list[str] | None = Query(None), + where_column: str | None = None, + where_operator: str = "eq", + where_value: list[str] | None = Query(None), + changed_since: str | None = None, + changed_until: str | None = None, registry: ServiceRegistry = Depends(get_registry), ) -> dict[str, Any]: """Return up to ``limit`` rows via the synchronous data-preview endpoint. @@ -192,7 +197,16 @@ def preview_table_v2( proj = projects[project] client = registry.storage._client_factory(proj.stack_url, proj.token) try: - text = client.get_table_data_preview(table_id=table_id, limit=limit, columns=columns) + text = client.get_table_data_preview( + table_id=table_id, + limit=limit, + columns=columns, + where_column=where_column, + where_operator=where_operator, + where_values=where_value, + changed_since=changed_since, + changed_until=changed_until, + ) finally: client.close() reader = _csv.reader(_io.StringIO(text)) @@ -209,9 +223,14 @@ def download_table_v2( columns: list[str] | None = Query(None), limit: int | None = None, branch_id: int | None = None, + where_column: str | None = None, + where_operator: str = "eq", + where_value: list[str] | None = Query(None), + changed_since: str | None = None, + changed_until: str | None = None, registry: ServiceRegistry = Depends(get_registry), ) -> FileResponse: - """Download the full table as CSV (uses async export).""" + """Download the table as CSV (uses async export). Optional where/changed filters.""" out_path = Path(tempfile.mkstemp(suffix=".csv", prefix="kbagent-")[1]) registry.storage.download_table( alias=project, @@ -220,6 +239,11 @@ def download_table_v2( columns=columns, limit=limit, branch_id=branch_id, + where_column=where_column, + where_operator=where_operator, + where_values=where_value, + changed_since=changed_since, + changed_until=changed_until, ) return FileResponse( path=str(out_path), @@ -401,6 +425,30 @@ def describe_columns( ) +# Registered AFTER the more specific /columns/.../describe route above: the +# greedy {table_id:path} would otherwise shadow that POST and swallow a +# ".../describe" suffix as part of the table id. +@router.post("/columns/{project}/{table_id:path}", summary="Add a table column") +def add_column( + project: str, + table_id: str, + column: str = Query(...), + not_null: bool = False, + default: str | None = None, + branch_id: int | None = None, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """Add a single column to a table. Mirrors `kbagent storage add-column`.""" + return registry.storage.add_column( + alias=project, + table_id=table_id, + column=column, + not_null=not_null, + default=default, + branch_id=branch_id, + ) + + # ---- Files ---- diff --git a/src/keboola_agent_cli/services/storage_service.py b/src/keboola_agent_cli/services/storage_service.py index cef5c50f..d4ee80d9 100644 --- a/src/keboola_agent_cli/services/storage_service.py +++ b/src/keboola_agent_cli/services/storage_service.py @@ -1004,9 +1004,18 @@ def download_table( limit: int | None = None, branch_id: int | None = None, keep_slices: bool = False, + *, + where_column: str | None = None, + where_operator: str = "eq", + where_values: list[str] | None = None, + changed_since: str | None = None, + changed_until: str | None = None, ) -> dict[str, Any]: """Export a storage table to a local CSV file. + Optional ``where_*`` / ``changed_*`` arguments filter the exported rows + server-side (forwarded verbatim to ``export_table_async``). + Uses the async export flow: export-async -> poll job -> get file info -> download from cloud URL. Handles gzip decompression transparently. @@ -1059,6 +1068,11 @@ def download_table( columns=columns, limit=limit, branch_id=branch_id, + where_column=where_column, + where_operator=where_operator, + where_values=where_values, + changed_since=changed_since, + changed_until=changed_until, ) # Step 2: Get file info from job results @@ -1298,6 +1312,59 @@ def truncate_tables( result["would_truncate"] = would_truncate return result + def add_column( + self, + alias: str, + table_id: str, + column: str, + not_null: bool = False, + default: str | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Add a single column to an existing table (synchronous). + + Parses the ``name:TYPE(length)`` column spec (the same grammar as + ``storage create-table --column``) into a Storage API column definition + and POSTs it to the synchronous add-column endpoint. + + Args: + alias: Project alias. + table_id: Full table ID (e.g. "in.c-bucket.table"). + column: Column spec, e.g. ``status:VARCHAR(20)`` or a bare ``notes`` + (implicit STRING). + not_null: If True, the new column is NOT NULL (the backend rejects + this unless the table is empty or a default is supplied). + default: Optional default value for the new column. + branch_id: If set, target a specific dev branch. + + Returns: + Dict with the added column name, its definition, table_id, and alias. + """ + projects = self.resolve_projects([alias]) + project = projects[alias] + + col_name = column.split(":", 1)[0].strip() + not_null_set = {col_name} if not_null else set() + defaults = {col_name: default} if default is not None else {} + parsed = _parse_column_spec(column, not_null_set, defaults) + + client = self._client_factory(project.stack_url, project.token) + try: + client.add_column( + table_id, + name=parsed["name"], + definition=parsed["definition"], + branch_id=branch_id, + ) + finally: + client.close() + return { + "table_id": table_id, + "column": parsed["name"], + "definition": parsed["definition"], + "project_alias": alias, + } + def delete_columns( self, alias: str, diff --git a/tests/test_e2e.py b/tests/test_e2e.py index caa05c91..3cffdd82 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -2925,6 +2925,27 @@ def _test_delete_column(self, table_id: str) -> None: columns_before = data["data"]["columns"] assert "value" in columns_before, f"Expected 'value' column, got {columns_before}" + # add-column: add a typed column and verify it appears in table-detail + data = self._run_ok( + "storage", + "add-column", + "--project", + self.alias, + "--table-id", + table_id, + "--column", + "status:VARCHAR(20)", + ) + 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']}" + ) + # delete-column dry-run data = self._run_ok( "storage", diff --git a/tests/test_storage_export_filters.py b/tests/test_storage_export_filters.py new file mode 100644 index 00000000..bf7969ba --- /dev/null +++ b/tests/test_storage_export_filters.py @@ -0,0 +1,223 @@ +"""Tests for storage export row-filters (where / changed_since) and add-column (0.62.0).""" + +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.models import AppConfig, ProjectConfig +from keboola_agent_cli.services.storage_service import StorageService + +runner = CliRunner() +TEST_TOKEN = "901-55555-fakeTestTokenDoNotUseXXXXXXXX" +STACK = "https://connection.keboola.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) + store.save(AppConfig(projects={"test": ProjectConfig(stack_url=STACK, token=TEST_TOKEN)})) + return store + + +def _make_service(store: ConfigStore, mock_client: MagicMock) -> StorageService: + return StorageService(config_store=store, client_factory=lambda url, token: mock_client) + + +class TestApplyTableFilters: + """Direct unit tests for the shared filter helper.""" + + def test_where_and_changed(self) -> None: + params: dict = {"fileType": "csv"} + KeboolaClient._apply_table_filters( + params, + where_column="status", + where_operator="neq", + where_values=["active", "pending"], + changed_since="-2 days", + changed_until="now", + ) + assert params["whereColumn"] == "status" + assert params["whereOperator"] == "neq" + assert params["whereValues[]"] == ["active", "pending"] + assert params["changedSince"] == "-2 days" + assert params["changedUntil"] == "now" + + def test_no_filters_leaves_params_untouched(self) -> None: + params: dict = {"limit": 100} + KeboolaClient._apply_table_filters(params) + assert params == {"limit": 100} + + def test_invalid_operator_raises(self) -> None: + with pytest.raises(ValueError, match="where_operator must be 'eq' or 'neq'"): + KeboolaClient._apply_table_filters( + {}, where_column="c", where_operator="like", where_values=["x"] + ) + + def test_half_specified_where_raises(self) -> None: + with pytest.raises(ValueError, match="must be given together"): + KeboolaClient._apply_table_filters({}, where_column="c") + with pytest.raises(ValueError, match="must be given together"): + KeboolaClient._apply_table_filters({}, where_values=["x"]) + + +class TestExportFiltersClient: + def _client(self) -> KeboolaClient: + return KeboolaClient(STACK, TEST_TOKEN) + + def test_export_async_forwards_filters(self) -> None: + client = self._client() + with ( + patch.object(client, "_request") as mock_req, + patch.object(client, "_wait_for_storage_job", return_value={"status": "success"}), + ): + mock_req.return_value.json.return_value = {"id": "job1"} + client.export_table_async( + "in.c-b.t", where_column="x", where_values=["1"], changed_since="-1 day" + ) + data = mock_req.call_args.kwargs["data"] + assert data["whereColumn"] == "x" + assert data["whereValues[]"] == ["1"] + assert data["whereOperator"] == "eq" + assert data["changedSince"] == "-1 day" + + def test_preview_forwards_filters(self) -> None: + client = self._client() + with patch.object(client, "_request") as mock_req: + mock_req.return_value.text = "id\n1\n" + client.get_table_data_preview("in.c-b.t", where_column="x", where_values=["1"]) + params = mock_req.call_args.kwargs["params"] + assert params["whereColumn"] == "x" + assert params["whereValues[]"] == ["1"] + + +class TestAddColumnClient: + def test_posts_to_columns_endpoint(self) -> None: + client = KeboolaClient(STACK, TEST_TOKEN) + with patch.object(client, "_request") as mock_req: + mock_req.return_value.json.return_value = {"id": "in.c-b.t"} + client.add_column( + "in.c-b.t", name="status", definition={"type": "VARCHAR", "length": "20"} + ) + assert mock_req.call_args.args[0] == "POST" + assert mock_req.call_args.args[1].endswith("/tables/in.c-b.t/columns") + assert mock_req.call_args.kwargs["json"] == { + "name": "status", + "definition": {"type": "VARCHAR", "length": "20"}, + } + + def test_untyped_column_omits_definition(self) -> None: + client = KeboolaClient(STACK, TEST_TOKEN) + with patch.object(client, "_request") as mock_req: + mock_req.return_value.json.return_value = {} + client.add_column("in.c-b.t", name="notes", definition=None) + assert mock_req.call_args.kwargs["json"] == {"name": "notes"} + + +class TestAddColumnService: + def test_parses_spec_and_calls_client(self, tmp_path: Path) -> None: + store = _make_store(tmp_path) + mock_client = MagicMock() + mock_client.add_column.return_value = {"id": "in.c-b.t"} + service = _make_service(store, mock_client) + + result = service.add_column( + alias="test", + table_id="in.c-b.t", + column="amount:NUMBER(18,2)", + not_null=True, + default="0", + ) + assert result["column"] == "amount" + assert result["definition"] == { + "type": "NUMBER", + "length": "18,2", + "nullable": False, + "default": "0", + } + mock_client.add_column.assert_called_once_with( + "in.c-b.t", + name="amount", + definition={"type": "NUMBER", "length": "18,2", "nullable": False, "default": "0"}, + branch_id=None, + ) + mock_client.close.assert_called_once() + + +class TestStorageCLI: + def _invoke(self, tmp_path: Path, mock_client: MagicMock, args: list[str]): + store = _make_store(tmp_path) + with ( + patch("keboola_agent_cli.cli.ConfigStore", return_value=store), + patch( + "keboola_agent_cli.cli.StorageService", + return_value=_make_service(store, mock_client), + ), + ): + return runner.invoke(app, args) + + def test_add_column_cli(self, tmp_path: Path) -> None: + mock_client = MagicMock() + mock_client.add_column.return_value = {"id": "in.c-b.t"} + result = self._invoke( + tmp_path, + mock_client, + [ + "--json", + "storage", + "add-column", + "--project", + "test", + "--table-id", + "in.c-b.t", + "--column", + "status:VARCHAR(20)", + ], + ) + assert result.exit_code == 0, result.output + assert mock_client.add_column.call_args.kwargs["name"] == "status" + assert mock_client.add_column.call_args.kwargs["definition"]["type"] == "VARCHAR" + + def test_download_table_forwards_filters(self, tmp_path: Path) -> None: + mock_service = MagicMock() + mock_service.download_table.return_value = { + "table_id": "in.c-b.t", + "file_size_bytes": 10, + "output_path": "t.csv", + } + store = _make_store(tmp_path) + with ( + patch("keboola_agent_cli.cli.ConfigStore", return_value=store), + patch("keboola_agent_cli.cli.StorageService", return_value=mock_service), + ): + result = runner.invoke( + app, + [ + "--json", + "storage", + "download-table", + "--project", + "test", + "--table-id", + "in.c-b.t", + "--where-column", + "status", + "--where-value", + "active", + "--where-operator", + "neq", + "--changed-since", + "-2 days", + ], + ) + assert result.exit_code == 0, result.output + kwargs = mock_service.download_table.call_args.kwargs + assert kwargs["where_column"] == "status" + assert kwargs["where_values"] == ["active"] + assert kwargs["where_operator"] == "neq" + assert kwargs["changed_since"] == "-2 days" diff --git a/tests/test_storage_write.py b/tests/test_storage_write.py index 6a370d6e..106cb8ed 100644 --- a/tests/test_storage_write.py +++ b/tests/test_storage_write.py @@ -2018,6 +2018,11 @@ def _fake_download(url, path): columns=None, limit=None, branch_id=None, + where_column=None, + where_operator="eq", + where_values=None, + changed_since=None, + changed_until=None, ) mock_client.get_file_info.assert_called_once_with(42, branch_id=None) mock_client.close.assert_called_once() @@ -2061,6 +2066,11 @@ def _fake_download(url, path): columns=["id", "name"], limit=100, branch_id=None, + where_column=None, + where_operator="eq", + where_values=None, + changed_since=None, + changed_until=None, ) def test_derives_filename_from_table_id(self, tmp_path: Path) -> None: @@ -2195,6 +2205,11 @@ def _fake_download(url, path): columns=None, limit=None, branch_id=42, + where_column=None, + where_operator="eq", + where_values=None, + changed_since=None, + changed_until=None, ) # Issue #161: get_file_info must also receive branch_id mock_client.get_file_info.assert_called_once_with(7, branch_id=42) @@ -2250,6 +2265,11 @@ def test_download_table_json(self, tmp_path: Path) -> None: limit=None, branch_id=None, keep_slices=False, + where_column=None, + where_operator="eq", + where_values=None, + changed_since=None, + changed_until=None, ) def test_download_table_with_columns_and_limit(self, tmp_path: Path) -> None: @@ -2297,6 +2317,11 @@ def test_download_table_with_columns_and_limit(self, tmp_path: Path) -> None: limit=50, branch_id=None, keep_slices=False, + where_column=None, + where_operator="eq", + where_values=None, + changed_since=None, + changed_until=None, ) def test_download_table_api_error(self, tmp_path: Path) -> None: diff --git a/uv.lock b/uv.lock index 4417273c..267132ba 100644 --- a/uv.lock +++ b/uv.lock @@ -580,7 +580,7 @@ wheels = [ [[package]] name = "keboola-agent-cli" -version = "0.61.1" +version = "0.62.0" source = { editable = "." } dependencies = [ { name = "croniter" },