diff --git a/CHANGELOG.md b/CHANGELOG.md index f77430a..44334a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- `ManagedDatabaseClient.fetch_table` now carries the `X-Database-Id` scope header on the result poll, the query-run poll, and the Arrow fetch — not only on the query submit. Results of database-scoped queries are themselves database-scoped, so every read against an existing synced table (merge/append loads, dlt state restore) failed with `400: Bad Request` once the table had data (dlthubworker#70). +- API error messages now include the response body (flattened, truncated to 500 chars). `400: Bad Request` alone hid the server's actual explanation. ## [0.6.0] - 2026-06-30 diff --git a/hotdata_framework/databases.py b/hotdata_framework/databases.py index dc50a67..06266d6 100644 --- a/hotdata_framework/databases.py +++ b/hotdata_framework/databases.py @@ -58,4 +58,9 @@ def managed_database_from_detail(detail: Any) -> ManagedDatabase: def api_error_message(exc: ApiException) -> str: - return exc.reason or str(exc) + reason = exc.reason or str(exc) + # Keep the response body: it carries the API's actual explanation. + body = getattr(exc, "body", None) + if body: + return f"{reason}: {' '.join(str(body).split())[:500]}" + return reason diff --git a/hotdata_framework/errors.py b/hotdata_framework/errors.py index fe2f058..5a5824b 100644 --- a/hotdata_framework/errors.py +++ b/hotdata_framework/errors.py @@ -23,6 +23,11 @@ def classify_sdk_error(error: Exception) -> HotdataError: if isinstance(error, ApiException): status_code = int(error.status or 0) message = f"{status_code}: {error.reason or 'unknown error'}" + # The response body is where the API explains itself (e.g. which + # header is missing) — without it "400: Bad Request" is undebuggable. + body = getattr(error, "body", None) + if body: + message = f"{message} — {' '.join(str(body).split())[:500]}" if status_code in (408, 409, 425, 429): return HotdataTransientError(message) if 500 <= status_code <= 599: diff --git a/hotdata_framework/managed_client.py b/hotdata_framework/managed_client.py index 4545de3..4a229f4 100644 --- a/hotdata_framework/managed_client.py +++ b/hotdata_framework/managed_client.py @@ -110,10 +110,26 @@ def operation() -> pa.Table | None: result_id = self._query_database_scoped(sql, database_id=db.id) if result_id is None: return None - return ArrowResultsApi(self._runtime.api).get_result_arrow(result_id) + return self._fetch_result_arrow(result_id, database_id=db.id) return self._request_with_retry(operation) + def _fetch_result_arrow(self, result_id: str, *, database_id: str) -> pa.Table: + """Fetch a ready result as Arrow, carrying the database scope. + + Results of database-scoped queries are themselves database-scoped — + the results endpoints reject requests without an ``X-Database-Id`` + header. The Arrow helper takes no per-call headers, so the header is + set as a client default for the duration of the call and removed + after (this client is not shared across threads). + """ + api = self._runtime.api + api.set_default_header("X-Database-Id", database_id) + try: + return ArrowResultsApi(api).get_result_arrow(result_id) + finally: + api.default_headers.pop("X-Database-Id", None) + def _poll( self, fetch: Callable[[], S], @@ -146,26 +162,34 @@ def _query_database_scoped(self, sql: str, *, database_id: str) -> str | None: # under ``result_id``; that result may be ``processing`` when the # inline preview returns, so wait for ``ready`` before the caller # fetches it as Arrow. - return self._wait_result_ready(raw.result_id) + return self._wait_result_ready(raw.result_id, database_id=database_id) if isinstance(raw, AsyncQueryResponse): - return self._wait_result_ready(self._await_query_run(raw.query_run_id)) + run_result = self._await_query_run(raw.query_run_id, database_id=database_id) + return self._wait_result_ready(run_result, database_id=database_id) return None - def _await_query_run(self, query_run_id: str) -> str | None: + def _await_query_run(self, query_run_id: str, *, database_id: str) -> str | None: runs = QueryRunsApi(self._runtime.api) run = self._poll( - lambda: runs.get_query_run(query_run_id), + # Runs (like results) of database-scoped queries are database-scoped. + lambda: runs.get_query_run( + query_run_id, _headers={"X-Database-Id": database_id} + ), is_ready=lambda r: r.status == "succeeded", describe="Query", ) return run.result_id - def _wait_result_ready(self, result_id: str | None) -> str | None: + def _wait_result_ready(self, result_id: str | None, *, database_id: str) -> str | None: if result_id is None: return None results = ResultsApi(self._runtime.api) self._poll( - lambda: results.get_result(result_id), + # The stored result of a database-scoped query 400s without the + # database scope header. + lambda: results.get_result( + result_id, _headers={"X-Database-Id": database_id} + ), is_ready=lambda r: r.status == "ready", describe=f"Result {result_id}", ) diff --git a/tests/test_errors.py b/tests/test_errors.py new file mode 100644 index 0000000..ed344af --- /dev/null +++ b/tests/test_errors.py @@ -0,0 +1,49 @@ +"""Error message construction: the API's response body must survive. + +"400: Bad Request" alone is undebuggable; the body carries the server's +actual explanation (e.g. which header was missing). Regression for the +opaque load failures in dlthubworker#70. +""" + +from __future__ import annotations + +from hotdata.rest import ApiException + +from hotdata_framework.databases import api_error_message +from hotdata_framework.errors import ( + HotdataTerminalError, + HotdataTransientError, + classify_sdk_error, +) + +BODY = '{"error":{"code":"BAD_REQUEST","message":"X-Database-Id header is required"}}' + + +def test_classify_sdk_error_includes_response_body() -> None: + err = classify_sdk_error(ApiException(status=400, reason="Bad Request", body=BODY)) + assert isinstance(err, HotdataTerminalError) + assert "400: Bad Request" in str(err) + assert "X-Database-Id header is required" in str(err) + + +def test_classify_sdk_error_without_body_keeps_short_form() -> None: + err = classify_sdk_error(ApiException(status=409, reason="Conflict")) + assert isinstance(err, HotdataTransientError) + assert str(err) == "409: Conflict" + + +def test_classify_sdk_error_truncates_and_flattens_body() -> None: + noisy = "x\n" * 1000 + err = classify_sdk_error(ApiException(status=500, reason="ISE", body=noisy)) + assert "\n" not in str(err) + assert len(str(err)) < 600 + + +def test_api_error_message_includes_body() -> None: + msg = api_error_message(ApiException(status=400, reason="Bad Request", body=BODY)) + assert msg.startswith("Bad Request: ") + assert "X-Database-Id header is required" in msg + + +def test_api_error_message_without_body() -> None: + assert api_error_message(ApiException(status=404, reason="Not Found")) == "Not Found" diff --git a/tests/test_managed_client.py b/tests/test_managed_client.py index 9dd70e4..1feba84 100644 --- a/tests/test_managed_client.py +++ b/tests/test_managed_client.py @@ -50,7 +50,7 @@ class FakeResultsApi: def __init__(self, api: object) -> None: pass - def get_result(self, result_id: str) -> Any: + def get_result(self, result_id: str, **kwargs: Any) -> Any: status = next(statuses) calls.append(f"get_result:{status}") return SimpleNamespace(status=status, result_id=result_id, error_message=None) @@ -75,13 +75,7 @@ def get_result_arrow(self, result_id: str) -> pa.Table: max_retries=1, retry_backoff_seconds=0.0, ) - client._runtime = SimpleNamespace( # type: ignore[assignment] - api=object(), - resolve_managed_database=lambda name: SimpleNamespace(id="db1", default_connection_id="c"), - list_managed_tables=lambda database, schema=None: [ - SimpleNamespace(table="orders", synced=True) - ], - ) + client._runtime = _fake_runtime() table = client.fetch_table(database="mydb", schema="public", table="orders") @@ -91,3 +85,86 @@ def get_result_arrow(self, result_id: str) -> pa.Table: assert "get_result:processing" in calls assert "get_result:ready" in calls assert calls.index("arrow") > calls.index("get_result:ready") + + +class _FakeApiClient: + """Just enough of the generated ApiClient's default-header surface.""" + + def __init__(self) -> None: + self.default_headers: dict[str, str] = {} + + def set_default_header(self, name: str, value: str) -> None: + self.default_headers[name] = value + + +def _fake_runtime(api: object | None = None) -> SimpleNamespace: + return SimpleNamespace( + api=api if api is not None else _FakeApiClient(), + resolve_managed_database=lambda name: SimpleNamespace(id="db1", default_connection_id="c"), + list_managed_tables=lambda database, schema=None: [ + SimpleNamespace(table="orders", synced=True) + ], + ) + + +def test_fetch_table_carries_database_scope_on_result_reads( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Results (and runs) of a database-scoped query are database-scoped: + the results endpoints 400 with "X-Database-Id header is required" when + the header is missing. ``fetch_table`` must carry the database id on the + result poll and the Arrow fetch, not only on the query submit. + + Regression: reruns/append loads against an existing synced table failed + with an opaque ``400: Bad Request`` (dlthubworker#70) because both reads + omitted the header. + """ + seen_headers: list[dict[str, Any] | None] = [] + arrow_headers: list[str | None] = [] + fake_api = _FakeApiClient() + + class FakeQueryApi: + def __init__(self, api: object) -> None: + pass + + def query(self, request: object, *, x_database_id: str) -> QueryResponse: + assert x_database_id == "db1" + return _query_response("rslt1") + + class FakeResultsApi: + def __init__(self, api: object) -> None: + pass + + def get_result(self, result_id: str, *, _headers: dict[str, Any] | None = None) -> Any: + seen_headers.append(_headers) + return SimpleNamespace(status="ready", result_id=result_id, error_message=None) + + class FakeArrowResultsApi: + def __init__(self, api: _FakeApiClient) -> None: + self._api = api + + def get_result_arrow(self, result_id: str) -> pa.Table: + # The scoped default header must be present DURING the fetch. + arrow_headers.append(self._api.default_headers.get("X-Database-Id")) + return pa.table({"id": [1]}) + + monkeypatch.setattr(mc, "QueryApi", FakeQueryApi) + monkeypatch.setattr(mc, "ResultsApi", FakeResultsApi) + monkeypatch.setattr(mc, "ArrowResultsApi", FakeArrowResultsApi) + + client = mc.ManagedDatabaseClient( + api_key="k", + workspace_id="w", + api_base_url="https://example.test", + max_retries=1, + retry_backoff_seconds=0.0, + ) + client._runtime = _fake_runtime(fake_api) + + table = client.fetch_table(database="mydb", schema="public", table="orders") + + assert table is not None + assert seen_headers == [{"X-Database-Id": "db1"}] + assert arrow_headers == ["db1"] + # The scoped header is removed once the fetch completes. + assert "X-Database-Id" not in fake_api.default_headers