diff --git a/CHANGELOG.md b/CHANGELOG.md index b7016fc..422d02d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.7.3] - 2026-07-16 + +### Changed + +- `upload_parquet()` now delegates to the SDK's `hotdata.uploads.UploadsApi.upload_file()` instead of hand-rolling the session → PUT → finalize flow. Uploads gain concurrent part PUTs under a peak-memory budget, per-part retries, and ETag/size validation, making large uploads substantially faster. Errors still surface as `RuntimeError` with the underlying `ApiException` as the direct cause. + +### Fixed + +- `classify_sdk_error` now classifies HTTP 501 (Not Implemented) as terminal instead of transient — a permanent capability gap must not burn retries. + ## [0.7.2] - 2026-07-15 ### Removed diff --git a/hotdata_framework/client.py b/hotdata_framework/client.py index 543e0f8..62cad87 100644 --- a/hotdata_framework/client.py +++ b/hotdata_framework/client.py @@ -1,9 +1,7 @@ from __future__ import annotations import functools -import os import time -import urllib3 from collections.abc import Iterator from dataclasses import asdict, dataclass from typing import Any, Literal @@ -15,16 +13,15 @@ from hotdata.api.query_api import QueryApi from hotdata.api.query_runs_api import QueryRunsApi from hotdata.api.results_api import ResultsApi -from hotdata.api.uploads_api import UploadsApi +# The enriched wrapper (hotdata.uploads), NOT the generated hotdata.api class: +# it adds the full upload_file orchestration used by upload_parquet. +from hotdata.uploads import UploadError, UploadsApi from hotdata.exceptions import ApiException from hotdata.models.add_managed_table_request import AddManagedTableRequest from hotdata.models.async_query_response import AsyncQueryResponse from hotdata.models.create_database_request import CreateDatabaseRequest from hotdata.models.database_default_schema_decl import DatabaseDefaultSchemaDecl from hotdata.models.database_default_table_decl import DatabaseDefaultTableDecl -from hotdata.models.create_upload_request import CreateUploadRequest -from hotdata.models.finalize_upload_part import FinalizeUploadPart -from hotdata.models.finalize_upload_request import FinalizeUploadRequest from hotdata.models.load_managed_table_request import LoadManagedTableRequest from hotdata.models.query_request import QueryRequest from hotdata.models.query_response import QueryResponse @@ -309,74 +306,29 @@ def list_managed_tables( return rows def upload_parquet(self, path: str) -> str: + """Upload a parquet file via the SDK's upload orchestration. + + ``UploadsApi.upload_file`` owns the whole session -> storage PUT -> + finalize flow: concurrent part uploads under a peak-memory budget, + per-part retries, and ETag/size validation. Errors surface with the + underlying ``ApiException`` as the direct cause when there is one, so + retry classification keeps seeing the status code. + """ if not is_parquet_path(path): raise ValueError(f"Managed table loads require a parquet file (got {path!r})") - file_size = os.path.getsize(path) - try: - session = self.uploads().create_upload_session_handler( - CreateUploadRequest( - declared_size_bytes=file_size, - content_type="application/octet-stream", - ) - ) - except ApiException as e: - if e.status == 501: - raise RuntimeError( - "the server does not support presigned uploads (HTTP 501); " - "managed table loads require a storage backend that can issue " - "presigned URLs (the POST /v1/files fallback was removed in " - "hotdata-framework 0.7.2)" - ) from e - raise RuntimeError(api_error_message(e)) from e - http = urllib3.PoolManager() - parts: list[FinalizeUploadPart] | None = None - try: - if session.mode == "single": - with open(path, "rb") as f: - data = f.read() - resp = http.request( - "PUT", - session.url, - body=data, - headers={"Content-Length": str(file_size), **session.headers}, - ) - if resp.status not in (200, 201, 204): - raise RuntimeError(f"Storage PUT failed: HTTP {resp.status}") - else: - collected: list[FinalizeUploadPart] = [] - with open(path, "rb") as f: - for i, part_url in enumerate(session.part_urls): - chunk = f.read(session.part_size) - resp = http.request( - "PUT", - part_url, - body=chunk, - headers={ - "Content-Length": str(len(chunk)), - **session.headers, - }, - ) - if resp.status not in (200, 201, 204): - raise RuntimeError( - f"Part {i + 1} PUT failed: HTTP {resp.status}" - ) - collected.append( - FinalizeUploadPart( - part_number=i + 1, - e_tag=resp.headers["ETag"], - ) - ) - parts = collected - finally: - http.clear() try: - finalized = self.uploads().finalize_upload_handler( - upload_id=session.upload_id, - x_upload_finalize_token=session.finalize_token, - finalize_upload_request=FinalizeUploadRequest(parts=parts), + finalized = self.uploads().upload_file( + path, content_type="application/octet-stream" ) except ApiException as e: raise RuntimeError(api_error_message(e)) from e + except UploadError as e: + node: BaseException | None = e + while node is not None and not isinstance(node, ApiException): + node = node.__cause__ + if isinstance(node, ApiException): + raise RuntimeError(api_error_message(node)) from node + raise RuntimeError(str(e)) from e return finalized.upload_id def load_managed_table( diff --git a/hotdata_framework/errors.py b/hotdata_framework/errors.py index 5a5824b..89749b6 100644 --- a/hotdata_framework/errors.py +++ b/hotdata_framework/errors.py @@ -30,6 +30,10 @@ def classify_sdk_error(error: Exception) -> HotdataError: message = f"{message} — {' '.join(str(body).split())[:500]}" if status_code in (408, 409, 425, 429): return HotdataTransientError(message) + if status_code == 501: + # Not Implemented is a permanent capability gap (e.g. the storage + # backend cannot issue presigned URLs) — retrying cannot succeed. + return HotdataTerminalError(message) if 500 <= status_code <= 599: return HotdataTransientError(message) return HotdataTerminalError(message) diff --git a/pyproject.toml b/pyproject.toml index 7cf5cef..d5ecc92 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "hotdata-framework" -version = "0.7.2" +version = "0.7.3" description = "Python framework for building Hotdata integrations: workspace/session runtime, query execution, and managed databases" readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_databases.py b/tests/test_databases.py index 118e318..ab77958 100644 --- a/tests/test_databases.py +++ b/tests/test_databases.py @@ -1,8 +1,7 @@ from __future__ import annotations -import io from types import SimpleNamespace -from unittest.mock import MagicMock, mock_open, patch +from unittest.mock import patch import pytest from hotdata.exceptions import ApiException @@ -195,143 +194,61 @@ def test_upload_parquet_rejects_non_parquet(): client.upload_parquet("/tmp/data.csv") -def _mock_open_bytes(data: bytes) -> MagicMock: - """Return an open() mock whose file handle supports read(n) via BytesIO.""" - bio = io.BytesIO(data) - m = MagicMock() - m.__enter__ = lambda s: bio - m.__exit__ = MagicMock(return_value=False) - return MagicMock(return_value=m) - - -def _session(mode: str, **kw) -> SimpleNamespace: - defaults = dict( - upload_id="upl_sess", - finalize_token="tok", - headers={}, - part_size=None, - part_urls=None, - url=None, - ) - return SimpleNamespace(mode=mode, **{**defaults, **kw}) - - -def _http_resp(status: int = 200, etag: str = '"abc"') -> SimpleNamespace: - return SimpleNamespace(status=status, headers={"ETag": etag}) - - -def test_upload_parquet_multipart(): +def test_upload_parquet_delegates_to_sdk_upload_file(): client = _client() - data = b"PAR1" + b"\x00" * 6 # 10 bytes -> 2 parts of 5 - session = _session("multipart", part_size=5, part_urls=["https://s/1", "https://s/2"]) finalized = SimpleNamespace(upload_id="upl_final") - with ( - patch("builtins.open", _mock_open_bytes(data)), - patch("os.path.getsize", return_value=len(data)), - patch.object(client, "uploads") as uploads, - patch("hotdata_framework.client.urllib3.PoolManager") as MockPool, - ): - pool = MockPool.return_value - pool.request.return_value = _http_resp() - pool.clear.return_value = None - uploads.return_value.create_upload_session_handler.return_value = session - uploads.return_value.finalize_upload_handler.return_value = finalized - + with patch.object(client, "uploads") as uploads: + uploads.return_value.upload_file.return_value = finalized upload_id = client.upload_parquet("/tmp/data.parquet") assert upload_id == "upl_final" - assert pool.request.call_count == 2 - finalize_call = uploads.return_value.finalize_upload_handler.call_args - assert finalize_call.kwargs["upload_id"] == "upl_sess" - assert finalize_call.kwargs["x_upload_finalize_token"] == "tok" - parts = finalize_call.kwargs["finalize_upload_request"].parts - assert len(parts) == 2 - assert parts[0].part_number == 1 - assert parts[1].part_number == 2 + uploads.return_value.upload_file.assert_called_once_with( + "/tmp/data.parquet", content_type="application/octet-stream" + ) + +def test_upload_parquet_surfaces_api_cause_from_upload_error(): + # SessionCreateError et al. wrap the ApiException as __cause__; the wrapper + # must re-raise with that ApiException as the DIRECT cause so + # classify_sdk_error keeps seeing the status code. + from hotdata.uploads import SessionCreateError -def test_upload_parquet_single_put(): client = _client() - data = b"PAR1tiny" - session = _session("single", url="https://s/put") - finalized = SimpleNamespace(upload_id="upl_single") + api_exc = ApiException(status=501) + sdk_err = SessionCreateError("opening the upload session failed") + sdk_err.__cause__ = api_exc - with ( - patch("builtins.open", mock_open(read_data=data)), - patch("os.path.getsize", return_value=len(data)), - patch.object(client, "uploads") as uploads, - patch("hotdata_framework.client.urllib3.PoolManager") as MockPool, - ): - pool = MockPool.return_value - pool.request.return_value = _http_resp() - pool.clear.return_value = None - uploads.return_value.create_upload_session_handler.return_value = session - uploads.return_value.finalize_upload_handler.return_value = finalized + with patch.object(client, "uploads") as uploads: + uploads.return_value.upload_file.side_effect = sdk_err + with pytest.raises(RuntimeError) as exc_info: + client.upload_parquet("/tmp/data.parquet") - upload_id = client.upload_parquet("/tmp/data.parquet") + assert exc_info.value.__cause__ is api_exc - assert upload_id == "upl_single" - pool.request.assert_called_once() - call_args = pool.request.call_args - assert call_args.args[0] == "PUT" - assert call_args.args[1] == "https://s/put" +def test_upload_parquet_wraps_plain_upload_error(): + from hotdata.uploads import StorageError -def test_upload_parquet_raises_on_501(): client = _client() + sdk_err = StorageError(status=503, part_number=3, body="upstream unavailable") - with ( - patch("builtins.open", mock_open(read_data=b"PAR1")), - patch("os.path.getsize", return_value=4), - patch.object(client, "uploads") as uploads, - ): - err = ApiException(status=501) - uploads.return_value.create_upload_session_handler.side_effect = err - - with pytest.raises(RuntimeError, match="presigned uploads"): + with patch.object(client, "uploads") as uploads: + uploads.return_value.upload_file.side_effect = sdk_err + with pytest.raises(RuntimeError, match="503"): client.upload_parquet("/tmp/data.parquet") - uploads.return_value.upload_file.assert_not_called() - -def test_upload_parquet_multipart_reads_are_part_size_bounded(): - """The OOM fix: multipart mode must never read more than part_size at once.""" - client = _client() - part_size = 5 - data = b"PAR1" + b"\x00" * 9 # 13 bytes -> parts of 5, 5, 3 - read_sizes: list[int] = [] - - class _TrackingFile(io.BytesIO): - def read(self, n=-1): - read_sizes.append(n) - return super().read(n) - - bio = _TrackingFile(data) - handle = MagicMock() - handle.__enter__ = lambda s: bio - handle.__exit__ = MagicMock(return_value=False) - session = _session( - "multipart", part_size=part_size, part_urls=["https://s/1", "https://s/2", "https://s/3"] +def test_classify_501_is_terminal(): + from hotdata_framework.errors import ( + HotdataTerminalError, + HotdataTransientError, + classify_sdk_error, ) - finalized = SimpleNamespace(upload_id="upl_final") - - with ( - patch("builtins.open", MagicMock(return_value=handle)), - patch("os.path.getsize", return_value=len(data)), - patch.object(client, "uploads") as uploads, - patch("hotdata_framework.client.urllib3.PoolManager") as MockPool, - ): - pool = MockPool.return_value - pool.request.return_value = _http_resp() - pool.clear.return_value = None - uploads.return_value.create_upload_session_handler.return_value = session - uploads.return_value.finalize_upload_handler.return_value = finalized - - client.upload_parquet("/tmp/data.parquet") - assert read_sizes, "expected chunked reads" - assert all(n == part_size for n in read_sizes) + assert isinstance(classify_sdk_error(ApiException(status=501)), HotdataTerminalError) + # The rest of 5xx stays transient. + assert isinstance(classify_sdk_error(ApiException(status=503)), HotdataTransientError) def test_load_managed_table_with_upload_id(): diff --git a/uv.lock b/uv.lock index d9bc915..45187be 100644 --- a/uv.lock +++ b/uv.lock @@ -101,7 +101,7 @@ wheels = [ [[package]] name = "hotdata-framework" -version = "0.7.2" +version = "0.7.3" source = { editable = "." } dependencies = [ { name = "hotdata" },