diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c43d4e..bb3b12d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `load_managed_table(..., mode=...)` selects the load mode (`replace` (default), `append`, `delete`, `update`, `upsert`) instead of always replacing the table. `replace`/`append` apply the upload directly; `delete`/`update`/`upsert` match rows by the table's declared key. Backward compatible — omitting `mode` still replaces. - `create_managed_database(..., keys={table: [cols]})` and `add_managed_table(..., key=[cols])` declare a table's row-identity key, enabling the key-based load modes on it. Requires a `hotdata` client whose managed-table decl models carry `key` (see the dependency floor bump); tables declared without a key stay `replace`/`append`-only. +### Fixed + +- `load_managed_table(..., mode="append")` is no longer retried on transient errors. Every other mode is idempotent, but retrying an `append` whose commit succeeded before the response was received would duplicate the uploaded rows; `append` now runs at most once. `mode` is also now typed as a literal of the accepted values. + ## [0.6.3] - 2026-07-08 ### Added diff --git a/hotdata_framework/client.py b/hotdata_framework/client.py index df5bf12..bbb2822 100644 --- a/hotdata_framework/client.py +++ b/hotdata_framework/client.py @@ -4,7 +4,7 @@ import time from collections.abc import Iterator from dataclasses import asdict, dataclass -from typing import Any +from typing import Any, Literal from hotdata import ApiClient, Configuration from hotdata.api.connections_api import ConnectionsApi @@ -46,6 +46,10 @@ from hotdata_framework.http import default_http_retries from hotdata_framework.result import QueryResult +# Load modes the managed-table endpoint accepts: replace overwrites, append adds +# rows, delete/update/upsert match by the table's declared key. +ManagedLoadMode = Literal["replace", "append", "delete", "update", "upsert"] + _TERMINAL = frozenset({"succeeded", "failed", "cancelled"}) _RESULT_FAILURE = frozenset({"failed", "cancelled"}) @@ -321,7 +325,7 @@ def load_managed_table( schema: str = DEFAULT_SCHEMA, upload_id: str | None = None, file: str | None = None, - mode: str = "replace", + mode: ManagedLoadMode = "replace", ) -> LoadManagedTableResult: if (upload_id is None) == (file is None): raise ValueError("Exactly one of upload_id or file is required") diff --git a/hotdata_framework/managed_client.py b/hotdata_framework/managed_client.py index 376261c..bac18cb 100644 --- a/hotdata_framework/managed_client.py +++ b/hotdata_framework/managed_client.py @@ -21,6 +21,7 @@ from hotdata.models.query_response import QueryResponse from hotdata_framework.client import HotdataClient as RuntimeClient +from hotdata_framework.client import ManagedLoadMode from hotdata_framework.databases import LoadManagedTableResult, ManagedDatabase from hotdata_framework.errors import ( HotdataTransientError, @@ -203,8 +204,11 @@ def load_managed_table( *, schema: str, upload_id: str, - mode: str = "replace", + mode: ManagedLoadMode = "replace", ) -> LoadManagedTableResult: + # append is the only non-idempotent mode: if the server commits the load + # but the response is lost, a retry re-appends the same rows. Run it + # at-most-once; every other mode is safe to retry. return self._request_with_retry( lambda: self._runtime.load_managed_table( database, @@ -212,16 +216,18 @@ def load_managed_table( schema=schema, upload_id=upload_id, mode=mode, - ) + ), + retryable=(mode != "append"), ) - def _request_with_retry(self, operation: Callable[[], T]) -> T: - for attempt in range(1, self._max_retries + 1): + def _request_with_retry(self, operation: Callable[[], T], *, retryable: bool = True) -> T: + max_attempts = self._max_retries if retryable else 1 + for attempt in range(1, max_attempts + 1): try: return operation() except Exception as error: mapped_error = classify_sdk_error(error.__cause__ or error) - if isinstance(mapped_error, HotdataTransientError) and attempt < self._max_retries: + if isinstance(mapped_error, HotdataTransientError) and attempt < max_attempts: backoff = min(self._retry_backoff_seconds * attempt, self._MAX_BACKOFF_SECONDS) time.sleep(backoff) continue diff --git a/tests/test_managed_client.py b/tests/test_managed_client.py index 260d67c..3966c0c 100644 --- a/tests/test_managed_client.py +++ b/tests/test_managed_client.py @@ -156,3 +156,56 @@ def get_result_arrow(self, result_id: str, *, x_database_id: str) -> pa.Table: assert table is not None assert result_scopes == ["db1"] assert arrow_scopes == ["db1"] + + +def _load_recording_runtime(calls: list[str]) -> SimpleNamespace: + """A runtime whose ``load_managed_table`` records each mode and always fails + with a transient error, so retry behaviour is observable via ``calls``.""" + + def load_managed_table( + database: str, table: str, *, schema: str, upload_id: str, mode: str + ) -> SimpleNamespace: + calls.append(mode) + raise TimeoutError("commit succeeded but response was lost") + + runtime = _fake_runtime() + runtime.load_managed_table = load_managed_table + return runtime + + +def _managed_client(max_retries: int) -> Any: + return mc.ManagedDatabaseClient( + api_key="k", + workspace_id="w", + api_base_url="https://example.test", + max_retries=max_retries, + retry_backoff_seconds=0.0, + ) + + +def test_append_load_runs_at_most_once(monkeypatch: pytest.MonkeyPatch) -> None: + """``append`` is not idempotent: retrying after a commit whose response was + lost would duplicate rows. A transient failure must surface immediately + without re-appending, even with retries budgeted.""" + monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None) + calls: list[str] = [] + client = _managed_client(max_retries=8) + client._runtime = _load_recording_runtime(calls) + + with pytest.raises(mc.HotdataTransientError): + client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append") + + assert calls == ["append"] # tried once, never retried + + +def test_idempotent_load_retries_on_transient(monkeypatch: pytest.MonkeyPatch) -> None: + """Idempotent modes still exhaust the retry budget on transient errors.""" + monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None) + calls: list[str] = [] + client = _managed_client(max_retries=3) + client._runtime = _load_recording_runtime(calls) + + with pytest.raises(mc.HotdataTransientError): + client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="replace") + + assert calls == ["replace", "replace", "replace"] # retried up to max_retries