rahulsmahadev commented on code in PR #3418:
URL: https://github.com/apache/iceberg-python/pull/3418#discussion_r3463729581
##########
pyiceberg/catalog/rest/__init__.py:
##########
@@ -396,6 +403,87 @@ class ListViewsResponse(IcebergBaseModel):
_PLANNING_RESPONSE_ADAPTER = TypeAdapter(PlanningResponse)
+class _RetryTimeoutHTTPAdapter(HTTPAdapter):
+ """HTTPAdapter that applies a default per-request timeout.
+
+ requests does not provide a way to set a default timeout on a Session;
+ without this adapter, every call would have to thread `timeout=` through.
+ The adapter applies `self._timeout` whenever a per-call timeout is not set.
+ """
+
+ def __init__(self, timeout: float | None = None, max_retries: Retry | int
| None = None) -> None:
+ self._timeout = timeout
+ if max_retries is not None:
+ super().__init__(max_retries=max_retries)
+ else:
+ super().__init__()
+
+ def send(
+ self,
+ request: PreparedRequest,
+ stream: bool = False,
+ timeout: None | float | tuple[float, float] | tuple[float, None] =
None,
+ verify: bool | str = True,
+ cert: None | bytes | str | tuple[bytes | str, bytes | str] = None,
+ proxies: Mapping[str, str] | None = None,
+ ) -> Response:
+ if timeout is None:
+ timeout = self._timeout
+ return super().send(request, stream=stream, timeout=timeout,
verify=verify, cert=cert, proxies=proxies)
+
+
+def _create_connection_adapter(properties: Properties) ->
_RetryTimeoutHTTPAdapter | None:
+ """Build a connection adapter from the optional `connection.*` properties.
+
+ Returns None when no `connection` block is supplied, leaving the default
+ Session behavior unchanged. Raises ValueError on invalid input.
+ """
+ connection_config = properties.get(CONNECTION)
+ if not connection_config:
+ return None
+ if not isinstance(connection_config, dict):
+ raise ValueError(f"`{CONNECTION}` must be a mapping, got:
{type(connection_config).__name__}")
+
+ timeout: float | None = None
+ if (raw_timeout := connection_config.get(CONNECTION_TIMEOUT)) is not None:
+ try:
+ timeout = float(raw_timeout)
+ except (TypeError, ValueError) as e:
+ raise ValueError(f"`{CONNECTION}.{CONNECTION_TIMEOUT}` must be a
number, got: {raw_timeout!r}") from e
+ if timeout <= 0:
+ raise ValueError(f"`{CONNECTION}.{CONNECTION_TIMEOUT}` must be a
positive number, got: {timeout}")
+
+ # `retries` and `backoff_factor` default to 0 (a no-op Retry) so the user
can set only
+ # one or the other without forcing the rest of the policy to be specified
explicitly.
+ retries = 0
+ if (raw_retries := connection_config.get(CONNECTION_RETRIES)) is not None:
+ try:
+ retries = int(raw_retries)
+ except (TypeError, ValueError) as e:
+ raise ValueError(f"`{CONNECTION}.{CONNECTION_RETRIES}` must be an
integer, got: {raw_retries!r}") from e
+ if retries < 0:
+ raise ValueError(f"`{CONNECTION}.{CONNECTION_RETRIES}` must be
non-negative, got: {retries}")
+
+ backoff_factor = 0.0
+ if (raw_backoff := connection_config.get(CONNECTION_BACKOFF_FACTOR)) is
not None:
+ try:
+ backoff_factor = float(raw_backoff)
+ except (TypeError, ValueError) as e:
+ raise ValueError(f"`{CONNECTION}.{CONNECTION_BACKOFF_FACTOR}` must
be a number, got: {raw_backoff!r}") from e
+ if backoff_factor < 0:
+ raise ValueError(f"`{CONNECTION}.{CONNECTION_BACKOFF_FACTOR}` must
be non-negative, got: {backoff_factor}")
+
+ return _RetryTimeoutHTTPAdapter(
Review Comment:
Good catch — fixed in 9a799c6. Added `raise_on_status=False` so urllib3
returns the final 5xx response instead of raising `MaxRetryError` /
`RetryError` on exhaustion. The response then flows through
`_handle_non_200_response` and is mapped to the typed exception
(`ServiceUnavailableError` for 503, etc.). Safe because `status_forcelist` is
hard-coded to transient codes only — 4xx codes are never retried and reach the
same mapping unchanged.
##########
tests/catalog/test_rest.py:
##########
@@ -2287,6 +2025,139 @@ def test_request_session_with_ssl_client_cert() -> None:
assert "Could not find the TLS certificate file, invalid path:
path_to_client_cert" in str(e.value)
+def test_session_without_connection_config_uses_default_adapter(rest_mock:
Mocker) -> None:
+ catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+ for adapter in catalog._session.adapters.values():
+ assert not isinstance(adapter, _RetryTimeoutHTTPAdapter)
+
+
+def test_session_with_connection_timeout_and_retries(rest_mock: Mocker) ->
None:
+ catalog_properties = {
+ "uri": TEST_URI,
+ "token": TEST_TOKEN,
+ CONNECTION: {
+ CONNECTION_TIMEOUT: 60,
+ CONNECTION_RETRIES: 5,
+ CONNECTION_BACKOFF_FACTOR: 1.0,
+ },
+ }
+ catalog = RestCatalog("rest", **catalog_properties) # type: ignore
+
+ https_adapter = catalog._session.adapters["https://";]
+ http_adapter = catalog._session.adapters["http://";]
+ assert isinstance(https_adapter, _RetryTimeoutHTTPAdapter)
+ assert https_adapter is http_adapter
+ assert https_adapter._timeout == 60.0
+ assert https_adapter.max_retries.total == 5
+ assert https_adapter.max_retries.backoff_factor == 1.0
+ # Internal retry policy: transient codes and idempotent methods only.
+ assert https_adapter.max_retries.status_forcelist == [429, 500, 502, 503,
504]
+ allowed_methods = https_adapter.max_retries.allowed_methods or frozenset()
+ assert set(allowed_methods) == {"GET", "HEAD", "OPTIONS"}
+
+
+def test_session_with_connection_timeout_only(rest_mock: Mocker) -> None:
+ catalog_properties = {
+ "uri": TEST_URI,
+ "token": TEST_TOKEN,
+ CONNECTION: {CONNECTION_TIMEOUT: "30"},
+ }
+ catalog = RestCatalog("rest", **catalog_properties) # type: ignore
+ adapter = catalog._session.adapters["https://";]
+ assert isinstance(adapter, _RetryTimeoutHTTPAdapter)
+ assert adapter._timeout == 30.0
+ # Default retry policy (total=0) is a no-op when only timeout is
configured.
+ assert adapter.max_retries.total == 0
+
+
+@contextmanager
+def _local_rest_server_503_then_200(num_failures: int) -> Iterator[dict[str,
Any]]:
+ """Stand up a loopback HTTP server that returns `num_failures` 503s for
`/v1/namespaces` then a 200.
+
+ Used in place of `requests_mock`, which replaces the HTTPAdapter and would
bypass the retry logic.
+
+ Yields a dict with `port` and `namespace_calls` keys (the latter is
updated in-place as requests arrive).
+ """
+ import json
+ import threading
+ from http.server import BaseHTTPRequestHandler, HTTPServer
+
+ state: dict[str, Any] = {"namespace_calls": 0}
+ config_body = json.dumps(
+ {"defaults": {}, "overrides": {}, "endpoints": [str(endpoint) for
endpoint in TEST_SUPPORTED_ENDPOINTS]}
+ ).encode()
+
+ class _Handler(BaseHTTPRequestHandler):
+ def do_GET(self) -> None:
+ if self.path.endswith("/v1/config"):
+ self._respond(200, config_body)
+ elif self.path.endswith("/v1/namespaces"):
+ state["namespace_calls"] += 1
+ if state["namespace_calls"] <= num_failures:
+ self._respond(503, b"")
+ else:
+ self._respond(200, json.dumps({"namespaces":
[["foo"]]}).encode())
+ else:
+ self._respond(404, b"")
+
+ def _respond(self, status: int, body: bytes) -> None:
+ self.send_response(status)
+ self.send_header("Content-Type", "application/json")
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ if body:
+ self.wfile.write(body)
+
+ def log_message(self, format: str, *args: Any) -> None: # silence
default access logs
+ pass
+
+ server = HTTPServer(("127.0.0.1", 0), _Handler)
+ state["port"] = server.server_address[1]
+ server_thread = threading.Thread(target=server.serve_forever, daemon=True)
+ server_thread.start()
+ try:
+ yield state
+ finally:
+ server.shutdown()
+ server.server_close()
+
+
+def test_session_retries_on_transient_5xx_then_succeeds() -> None:
Review Comment:
Added `test_session_exhausted_retries_surfaces_typed_exception` in 9a799c6 —
drives the retry loop to exhaustion against a server that always returns 503
and asserts `ServiceUnavailableError` is raised (not `RetryError`), plus
verifies that exactly `retries + 1` requests were made. This is the regression
guard for the fix above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
