Merge pull request #44 from hotdata-dev/fix/append-at-most-once

fix: run append loads at-most-once to prevent retry duplication

Author: shefeek-jinnah

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

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")

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,25 +204,30 @@ 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,
                 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

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