Address review: reuse property helpers and align timeout with Java client

Author: rahulsmahadev

mkdocs/docs/configuration.md

@@ -418,24 +418,29 @@ When server-side planning returns `storage-credentials` on a completed plan, PyI
 The REST Catalog uses `requests` with no retries and no timeout by default, so transient
 5xx / network failures bubble up immediately and slow servers can hang the client indefinitely.
 Set the `rest.client.*` catalog properties to opt in to a request timeout and a retry policy.
+The property names mirror the Java REST client so a single catalog configuration can serve both.
 
 ```yaml
 catalog:
   default:
     uri: http://rest-catalog/ws/
-    rest.client.request-timeout: 60       # seconds, applied to the whole request
-    rest.client.max-retries: 5            # number of retry attempts on transient failures
-    rest.client.retry-backoff-factor: 1.0 # exponential backoff between retries
+    rest.client.connection-timeout-ms: 5000  # milliseconds, time allowed to establish a connection
+    rest.client.socket-timeout-ms: 60000     # milliseconds, time allowed between bytes once connected
+    rest.client.max-retries: 5               # number of retry attempts on transient failures
+    rest.client.retry-backoff-factor: 1.0    # exponential backoff between retries
 ```
 
 | Key                                     | Example  | Description                                                                                                        |
 | --------------------------------------- | -------- | ------------------------------------------------------------------------------------------------------------------ |
-| rest.client.request-timeout             | 60       | Timeout in seconds, applied as a single value to the whole request. Must be a positive number.                      |
+| rest.client.connection-timeout-ms       | 5000     | Time to establish a connection, in milliseconds. Must be a positive number.                                        |
+| rest.client.socket-timeout-ms           | 60000    | Time allowed between bytes once a connection is established, in milliseconds. Must be a positive number.            |
 | rest.client.max-retries                 | 5        | Number of retry attempts for transient failures. Must be non-negative.                                             |
 | rest.client.retry-backoff-factor        | 1.0      | Backoff factor between retry attempts. Must be non-negative. See [`urllib3` Retry docs](https://urllib3.readthedocs.io/en/latest/reference/urllib3.util.html#urllib3.util.Retry) for the formula. |
 
-Retries are applied to idempotent methods only (`GET`, `HEAD`, `OPTIONS`, `PUT`, `DELETE`) and to the
-transient HTTP status codes `429`, `500`, `502`, `503`, `504`. Other failures are not retried.
+`requests` cannot split the connection and socket timeouts, so the two values are summed and applied
+as a single request timeout (floored to whole seconds). Retries are applied to idempotent methods
+only (`GET`, `HEAD`, `OPTIONS`, `PUT`, `DELETE`) and to the transient HTTP status codes `429`, `500`,
+`502`, `503`, `504`. Other failures are not retried.
 
 #### Headers in REST Catalog
 

pyiceberg/catalog/rest/__init__.py

@@ -19,12 +19,11 @@
 import logging
 import time
 from collections import deque
-from collections.abc import Callable, Mapping
+from collections.abc import Mapping
 from enum import Enum
 from typing import (
     TYPE_CHECKING,
     Any,
-    TypeVar,
 )
 from urllib.parse import quote, unquote
 
@@ -98,7 +97,13 @@
 from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel, Identifier, Properties
 from pyiceberg.types import transform_dict_value_to_str
 from pyiceberg.utils.deprecated import deprecation_message
-from pyiceberg.utils.properties import get_first_property_value, get_header_properties, property_as_bool, property_as_int
+from pyiceberg.utils.properties import (
+    get_first_property_value,
+    get_header_properties,
+    property_as_bool,
+    property_as_float,
+    property_as_int,
+)
 from pyiceberg.view import View
 from pyiceberg.view.metadata import ViewMetadata, ViewVersion
 
@@ -278,7 +283,8 @@ class ScanPlanningMode(Enum):
 SIGV4_SERVICE = "rest.signing-name"
 SIGV4_MAX_RETRIES = "rest.sigv4.max-retries"
 SIGV4_MAX_RETRIES_DEFAULT = 10
-REST_CLIENT_REQUEST_TIMEOUT = "rest.client.request-timeout"
+REST_CLIENT_CONNECTION_TIMEOUT_MS = "rest.client.connection-timeout-ms"
+REST_CLIENT_SOCKET_TIMEOUT_MS = "rest.client.socket-timeout-ms"
 REST_CLIENT_MAX_RETRIES = "rest.client.max-retries"
 REST_CLIENT_RETRY_BACKOFF_FACTOR = "rest.client.retry-backoff-factor"
 # Hard-coded internally so users cannot misconfigure the retry policy
@@ -453,29 +459,6 @@ class ListViewsResponse(IcebergBaseModel):
 _PLANNING_RESPONSE_ADAPTER = TypeAdapter(PlanningResponse)
 
 
-_T = TypeVar("_T", int, float)
-
-
-def _parse_connection_property(
-    properties: Properties,
-    property_name: str,
-    converter: Callable[[Any], _T],
-    type_description: str,
-    is_invalid: Callable[[_T], bool],
-    range_description: str,
-) -> _T | None:
-    raw_value = properties.get(property_name)
-    if raw_value is None:
-        return None
-    try:
-        value = converter(raw_value)
-    except (TypeError, ValueError) as e:
-        raise ValueError(f"`{property_name}` must be {type_description}, got: {raw_value!r}") from e
-    if is_invalid(value):
-        raise ValueError(f"`{property_name}` must be {range_description}, got: {value}")
-    return value
-
-
 class _RetryTimeoutHTTPAdapter(HTTPAdapter):
     """HTTPAdapter that applies a default per-request timeout.
 
@@ -508,37 +491,30 @@ def _create_connection_adapter(properties: Properties) -> _RetryTimeoutHTTPAdapt
     Returns None when no connection properties are supplied, leaving the default
     Session behavior unchanged. Raises ValueError on invalid input.
     """
-    if not any(
-        property_name in properties
-        for property_name in (REST_CLIENT_REQUEST_TIMEOUT, REST_CLIENT_MAX_RETRIES, REST_CLIENT_RETRY_BACKOFF_FACTOR)
-    ):
-        return None
+    connection_timeout_ms = property_as_int(properties, REST_CLIENT_CONNECTION_TIMEOUT_MS)
+    if connection_timeout_ms is not None and connection_timeout_ms <= 0:
+        raise ValueError(f"`{REST_CLIENT_CONNECTION_TIMEOUT_MS}` must be a positive number, got: {connection_timeout_ms}")
 
-    timeout = _parse_connection_property(
-        properties,
-        REST_CLIENT_REQUEST_TIMEOUT,
-        float,
-        "a number",
-        lambda value: value <= 0,
-        "a positive number",
-    )
+    socket_timeout_ms = property_as_int(properties, REST_CLIENT_SOCKET_TIMEOUT_MS)
+    if socket_timeout_ms is not None and socket_timeout_ms <= 0:
+        raise ValueError(f"`{REST_CLIENT_SOCKET_TIMEOUT_MS}` must be a positive number, got: {socket_timeout_ms}")
 
-    retries = _parse_connection_property(
-        properties,
-        REST_CLIENT_MAX_RETRIES,
-        int,
-        "an integer",
-        lambda value: value < 0,
-        "non-negative",
-    )
-    backoff_factor = _parse_connection_property(
-        properties,
-        REST_CLIENT_RETRY_BACKOFF_FACTOR,
-        float,
-        "a number",
-        lambda value: value < 0,
-        "non-negative",
-    )
+    retries = property_as_int(properties, REST_CLIENT_MAX_RETRIES)
+    if retries is not None and retries < 0:
+        raise ValueError(f"`{REST_CLIENT_MAX_RETRIES}` must be non-negative, got: {retries}")
+
+    backoff_factor = property_as_float(properties, REST_CLIENT_RETRY_BACKOFF_FACTOR)
+    if backoff_factor is not None and backoff_factor < 0:
+        raise ValueError(f"`{REST_CLIENT_RETRY_BACKOFF_FACTOR}` must be non-negative, got: {backoff_factor}")
+
+    if all(value is None for value in (connection_timeout_ms, socket_timeout_ms, retries, backoff_factor)):
+        return None
+
+    # requests uses a single timeout and cannot split connect vs socket, so follow the Java client
+    # and sum the two (milliseconds), flooring to whole seconds.
+    timeout: float | None = None
+    if connection_timeout_ms is not None or socket_timeout_ms is not None:
+        timeout = ((connection_timeout_ms or 0) + (socket_timeout_ms or 0)) // 1000
 
     return _RetryTimeoutHTTPAdapter(
         timeout=timeout,

tests/catalog/test_rest.py

@@ -37,9 +37,10 @@
     EMPTY_BODY_SHA256,
     OAUTH2_SERVER_URI,
     PAGE_SIZE,
+    REST_CLIENT_CONNECTION_TIMEOUT_MS,
     REST_CLIENT_MAX_RETRIES,
-    REST_CLIENT_REQUEST_TIMEOUT,
     REST_CLIENT_RETRY_BACKOFF_FACTOR,
+    REST_CLIENT_SOCKET_TIMEOUT_MS,
     SIGV4_MAX_RETRIES,
     SIGV4_MAX_RETRIES_DEFAULT,
     SNAPSHOT_LOADING_MODE,
@@ -2300,17 +2301,18 @@ def test_session_with_connection_timeout_and_retries(rest_mock: Mocker) -> None:
     catalog_properties = {
         "uri": TEST_URI,
         "token": TEST_TOKEN,
-        REST_CLIENT_REQUEST_TIMEOUT: 60,
-        REST_CLIENT_MAX_RETRIES: 5,
-        REST_CLIENT_RETRY_BACKOFF_FACTOR: 1.0,
+        REST_CLIENT_CONNECTION_TIMEOUT_MS: "5000",
+        REST_CLIENT_SOCKET_TIMEOUT_MS: "60000",
+        REST_CLIENT_MAX_RETRIES: "5",
+        REST_CLIENT_RETRY_BACKOFF_FACTOR: "1.0",
     }
-    catalog = RestCatalog("rest", **catalog_properties)  # type: ignore
+    catalog = RestCatalog("rest", **catalog_properties)
 
     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._timeout == 65  # (5000 + 60000) ms floored to whole seconds
     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.
@@ -2323,13 +2325,26 @@ def test_session_with_connection_timeout_only(rest_mock: Mocker) -> None:
     catalog_properties = {
         "uri": TEST_URI,
         "token": TEST_TOKEN,
-        REST_CLIENT_REQUEST_TIMEOUT: "30",
+        REST_CLIENT_CONNECTION_TIMEOUT_MS: "5000",
+    }
+    catalog = RestCatalog("rest", **catalog_properties)
+    adapter = catalog._session.adapters["https://"]
+    assert isinstance(adapter, _RetryTimeoutHTTPAdapter)
+    assert adapter._timeout == 5  # 5000 ms floored to whole seconds
+    # Default retry policy (total=0) is a no-op when only a timeout is configured.
+    assert adapter.max_retries.total == 0
+
+
+def test_session_with_socket_timeout_only(rest_mock: Mocker) -> None:
+    catalog_properties = {
+        "uri": TEST_URI,
+        "token": TEST_TOKEN,
+        REST_CLIENT_SOCKET_TIMEOUT_MS: "60000",
     }
     catalog = RestCatalog("rest", **catalog_properties)
     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._timeout == 60  # 60000 ms floored to whole seconds
     assert adapter.max_retries.total == 0
 
 
@@ -2390,12 +2405,12 @@ def test_session_retries_on_transient_5xx_then_succeeds() -> None:
     with _local_rest_server_503_then_200(num_failures=3) as server:
         catalog = RestCatalog(
             "rest",
-            **{  # type: ignore
+            **{
                 "uri": f"http://127.0.0.1:{server['port']}/",
                 "token": TEST_TOKEN,
                 # backoff-factor=0 keeps the test fast; retries=3 covers three 503s + the eventual 200.
-                REST_CLIENT_MAX_RETRIES: 3,
-                REST_CLIENT_RETRY_BACKOFF_FACTOR: 0,
+                REST_CLIENT_MAX_RETRIES: "3",
+                REST_CLIENT_RETRY_BACKOFF_FACTOR: "0",
             },
         )
         assert catalog.list_namespaces() == [("foo",)]
@@ -2409,11 +2424,11 @@ def test_session_exhausted_retries_surfaces_typed_exception() -> None:
     with _local_rest_server_503_then_200(num_failures=100) as server:
         catalog = RestCatalog(
             "rest",
-            **{  # type: ignore
+            **{
                 "uri": f"http://127.0.0.1:{server['port']}/",
                 "token": TEST_TOKEN,
-                REST_CLIENT_MAX_RETRIES: 2,
-                REST_CLIENT_RETRY_BACKOFF_FACTOR: 0,
+                REST_CLIENT_MAX_RETRIES: "2",
+                REST_CLIENT_RETRY_BACKOFF_FACTOR: "0",
             },
         )
         with pytest.raises(ServiceUnavailableError):
@@ -2426,20 +2441,30 @@ def test_session_with_invalid_connection_timeout_raises(rest_mock: Mocker) -> No
     catalog_properties = {
         "uri": TEST_URI,
         "token": TEST_TOKEN,
-        REST_CLIENT_REQUEST_TIMEOUT: -1,
+        REST_CLIENT_CONNECTION_TIMEOUT_MS: "-1",
     }
-    with pytest.raises(ValueError, match="`rest.client.request-timeout` must be a positive number"):
-        RestCatalog("rest", **catalog_properties)  # type: ignore
+    with pytest.raises(ValueError, match="`rest.client.connection-timeout-ms` must be a positive number"):
+        RestCatalog("rest", **catalog_properties)
+
+
+def test_session_with_invalid_socket_timeout_raises(rest_mock: Mocker) -> None:
+    catalog_properties = {
+        "uri": TEST_URI,
+        "token": TEST_TOKEN,
+        REST_CLIENT_SOCKET_TIMEOUT_MS: "0",
+    }
+    with pytest.raises(ValueError, match="`rest.client.socket-timeout-ms` must be a positive number"):
+        RestCatalog("rest", **catalog_properties)
 
 
 def test_session_with_invalid_connection_retries_raises(rest_mock: Mocker) -> None:
     catalog_properties = {
         "uri": TEST_URI,
         "token": TEST_TOKEN,
-        REST_CLIENT_MAX_RETRIES: -1,
+        REST_CLIENT_MAX_RETRIES: "-1",
     }
     with pytest.raises(ValueError, match="`rest.client.max-retries` must be non-negative"):
-        RestCatalog("rest", **catalog_properties)  # type: ignore
+        RestCatalog("rest", **catalog_properties)
 
 
 def test_rest_catalog_with_basic_auth_type(rest_mock: Mocker) -> None: