From 889a39b5c12d18a532d9625142741260c114ddee Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Thu, 17 Sep 2026 16:36:42 +0000 Subject: [PATCH 01/11] feat(http): add default connect/read timeouts to HttpClient and make timeouts retryable Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../error_handlers/default_error_mapping.py | 24 ++- .../sources/streams/http/http_client.py | 26 +++ .../test_default_error_handler.py | 10 ++ .../test_http_status_error_handler.py | 9 + .../sources/streams/http/test_http_client.py | 170 +++++++++++++++++- 5 files changed, 236 insertions(+), 3 deletions(-) diff --git a/airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py b/airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py index 45716768fd..b5b075b44d 100644 --- a/airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py +++ b/airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py @@ -4,7 +4,14 @@ from typing import Mapping, Type, Union -from requests.exceptions import InvalidSchema, InvalidURL, RequestException +from requests.exceptions import ( + ConnectTimeout, + InvalidSchema, + InvalidURL, + ReadTimeout, + RequestException, + Timeout, +) from airbyte_cdk.models import FailureType from airbyte_cdk.sources.streams.http.error_handlers.response_models import ( @@ -23,6 +30,21 @@ failure_type=FailureType.transient_error, error_message="Invalid URL specified or DNS error occurred: The endpoint that data is being requested from is not a valid URL. Exception: requests.exceptions.InvalidURL", ), + ConnectTimeout: ErrorResolution( + response_action=ResponseAction.RETRY, + failure_type=FailureType.transient_error, + error_message="The request timed out while connecting to the server. Exception: requests.exceptions.ConnectTimeout", + ), + ReadTimeout: ErrorResolution( + response_action=ResponseAction.RETRY, + failure_type=FailureType.transient_error, + error_message="The request timed out while waiting for the server to respond. Exception: requests.exceptions.ReadTimeout", + ), + Timeout: ErrorResolution( + response_action=ResponseAction.RETRY, + failure_type=FailureType.transient_error, + error_message="The request timed out. Exception: requests.exceptions.Timeout", + ), RequestException: ErrorResolution( response_action=ResponseAction.RETRY, failure_type=FailureType.transient_error, diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index c9008d1e64..9faaa26fad 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -72,6 +72,21 @@ MessageRepresentationAirbyteTracedErrors = AirbyteTracedException BODY_REQUEST_METHODS = ("GET", "POST", "PUT", "PATCH") +DEFAULT_CONNECT_TIMEOUT_SECONDS: float = 30.0 +DEFAULT_READ_TIMEOUT_SECONDS: float = 300.0 +ENV_HTTP_CONNECT_TIMEOUT_SECONDS = "AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS" +ENV_HTTP_READ_TIMEOUT_SECONDS = "AIRBYTE_HTTP_READ_TIMEOUT_SECONDS" + + +def default_request_timeout() -> Tuple[float, float]: + """Returns the `(connect, read)` timeout in seconds applied to requests that do not set their own. + + Both values can be overridden with the `AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS` and + `AIRBYTE_HTTP_READ_TIMEOUT_SECONDS` environment variables. + """ + connect = float(os.getenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, DEFAULT_CONNECT_TIMEOUT_SECONDS)) + read = float(os.getenv(ENV_HTTP_READ_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS)) + return (connect, read) def monkey_patched_get_item(self, key): # type: ignore # this interface is a copy/paste from the requests_cache lib @@ -121,7 +136,13 @@ def __init__( error_message_parser: Optional[ErrorMessageParser] = None, disable_retries: bool = False, message_repository: Optional[MessageRepository] = None, + request_timeout: Optional[Union[float, Tuple[float, float]]] = None, ): + """Initialize the HTTP client. + + Args: + request_timeout: The default connect/read timeout in seconds for requests. + """ self._name = name self._api_budget: APIBudget = api_budget or APIBudget(policies=[]) if session: @@ -151,6 +172,9 @@ def __init__( self._disable_retries = disable_retries self._message_repository = message_repository self._authenticator_update_failed = False + self._request_timeout: Union[float, Tuple[float, float]] = ( + request_timeout if request_timeout is not None else default_request_timeout() + ) @property def cache_filename(self) -> str: @@ -411,6 +435,8 @@ def _send( exc: Optional[requests.RequestException] = None try: + if "timeout" not in request_kwargs: + request_kwargs = {**request_kwargs, "timeout": self._request_timeout} response = self._session.send(request, **request_kwargs) except requests.RequestException as e: exc = e diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py index 526e5760fc..f07d2ac6e3 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py @@ -6,6 +6,7 @@ import pytest import requests +from requests.exceptions import ReadTimeout from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy import ( ConstantBackoffStrategy, @@ -291,6 +292,15 @@ def test_default_error_handler_with_unmapped_http_code(): assert actual_error_resolution.response_action == ResponseAction.RETRY +def test_default_error_handler_with_read_timeout(): + error_handler = DefaultErrorHandler(config={}, parameters={}) + + actual_error_resolution = error_handler.interpret_response(ReadTimeout()) + + assert actual_error_resolution.response_action == ResponseAction.RETRY + assert actual_error_resolution.failure_type == FailureType.transient_error + + def test_predicate_takes_precedent_over_default_mapped_error(): response_mock = create_response(404, json_body={"error": "test"}) diff --git a/unit_tests/sources/streams/http/error_handlers/test_http_status_error_handler.py b/unit_tests/sources/streams/http/error_handlers/test_http_status_error_handler.py index b49bc17f89..1b468c0661 100644 --- a/unit_tests/sources/streams/http/error_handlers/test_http_status_error_handler.py +++ b/unit_tests/sources/streams/http/error_handlers/test_http_status_error_handler.py @@ -5,6 +5,7 @@ import pytest import requests +from requests.exceptions import ConnectTimeout, ReadTimeout, Timeout from airbyte_cdk.models import FailureType from airbyte_cdk.sources.streams.http.error_handlers import ( @@ -82,6 +83,14 @@ def test_given_requests_exception_returns_retry_action_as_transient_error(): assert error_resolution.failure_type +@pytest.mark.parametrize("exception", [ReadTimeout(), ConnectTimeout(), Timeout()]) +def test_given_timeout_exception_returns_retry_action_as_transient_error(exception): + error_resolution = HttpStatusErrorHandler(logger).interpret_response(exception) + + assert error_resolution.response_action == ResponseAction.RETRY + assert error_resolution.failure_type == FailureType.transient_error + + def test_given_unmapped_exception_returns_retry_action_as_system_error(): error_resolution = HttpStatusErrorHandler(logger).interpret_response(Exception()) diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 2c642ab9b1..1519c19cd7 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -2,9 +2,11 @@ import logging import os +import socket +import threading import time from datetime import timedelta -from unittest.mock import MagicMock, patch +from unittest.mock import ANY, MagicMock, patch import pytest import requests @@ -26,7 +28,13 @@ RequestBodyException, UserDefinedBackoffException, ) -from airbyte_cdk.sources.streams.http.http_client import MessageRepresentationAirbyteTracedErrors +from airbyte_cdk.sources.streams.http.http_client import ( + DEFAULT_CONNECT_TIMEOUT_SECONDS, + DEFAULT_READ_TIMEOUT_SECONDS, + ENV_HTTP_CONNECT_TIMEOUT_SECONDS, + ENV_HTTP_READ_TIMEOUT_SECONDS, + MessageRepresentationAirbyteTracedErrors, +) from airbyte_cdk.sources.streams.http.requests_native_auth import TokenAuthenticator from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -198,6 +206,164 @@ def test_valid_basic_send_request(mocker): assert returned_response == mocked_response +def _mock_success_session(): + session = MagicMock(spec=requests.Session) + session.merge_environment_settings.return_value = {} + session.prepare_request.return_value = requests.Request("GET", "https://test.example").prepare() + response = MagicMock(spec=requests.Response) + response.status_code = 200 + response.ok = True + response.headers = {} + session.send.return_value = response + return session + + +def test_send_applies_default_timeout(): + session = _mock_success_session() + client = HttpClient(name="test", logger=MagicMock(), session=session) + + client.send_request("GET", "https://test.example", request_kwargs={}) + + session.send.assert_called_once_with( + ANY, + timeout=(DEFAULT_CONNECT_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS), + ) + + +def test_request_kwargs_timeout_overrides_default(): + session = _mock_success_session() + client = HttpClient(name="test", logger=MagicMock(), session=session) + + client.send_request("GET", "https://test.example", request_kwargs={"timeout": 7}) + client.send_request("GET", "https://test.example", request_kwargs={"timeout": None}) + + assert session.send.call_args_list[0].kwargs["timeout"] == 7 + assert session.send.call_args_list[1].kwargs["timeout"] is None + + +def test_constructor_request_timeout_overrides_default(): + session = _mock_success_session() + client = HttpClient( + name="test", + logger=MagicMock(), + session=session, + request_timeout=(1, 2), + ) + + client.send_request("GET", "https://test.example", request_kwargs={}) + + assert session.send.call_args.kwargs["timeout"] == (1, 2) + + +def test_env_vars_override_default_timeout(monkeypatch): + monkeypatch.setenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, "5") + monkeypatch.setenv(ENV_HTTP_READ_TIMEOUT_SECONDS, "9") + session = _mock_success_session() + client = HttpClient(name="test", logger=MagicMock(), session=session) + + client.send_request("GET", "https://test.example", request_kwargs={}) + + assert session.send.call_args.kwargs["timeout"] == (5.0, 9.0) + + +def test_hanging_server_times_out_retries_and_raises(): + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen() + server.settimeout(0.1) + port = server.getsockname()[1] + stop_server = threading.Event() + accepted_connections = [] + accepted_three_connections = threading.Event() + + def accept_connections(): + while not stop_server.is_set(): + try: + connection, _ = server.accept() + except TimeoutError: + continue + except OSError: + break + accepted_connections.append(connection) + if len(accepted_connections) >= 3: + accepted_three_connections.set() + + server_thread = threading.Thread(target=accept_connections, daemon=True) + server_thread.start() + + try: + client = HttpClient( + name="test", + logger=MagicMock(), + error_handler=HttpStatusErrorHandler(logger=MagicMock(), max_retries=2), + backoff_strategy=CustomBackoffStrategy(backoff_time_value=0.01), + request_timeout=(5, 0.3), + ) + started_at = time.monotonic() + + with pytest.raises(AirbyteTracedException) as exc_info: + client.send_request("GET", f"http://127.0.0.1:{port}/", request_kwargs={}) + + elapsed = time.monotonic() - started_at + accepted_three_connections.wait(timeout=1) + assert elapsed < 10 + assert len(accepted_connections) >= 3 + assert exc_info.value.failure_type == FailureType.transient_error + assert "ReadTimeout" in (exc_info.value.internal_message or "") + finally: + stop_server.set() + server.close() + server_thread.join(timeout=1) + for connection in accepted_connections: + connection.close() + + +def test_slow_but_responding_server_succeeds(): + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen() + port = server.getsockname()[1] + accepted_connections = [] + response_body = b'{"ok": true}' + response = ( + b"HTTP/1.1 200 OK\r\n" + b"Content-Type: application/json\r\n" + + f"Content-Length: {len(response_body)}\r\n".encode() + + b"Connection: close\r\n" + + b"\r\n" + + response_body + ) + + def respond_slowly(): + connection, _ = server.accept() + accepted_connections.append(connection) + connection.recv(65536) + time.sleep(1.0) + connection.sendall(response) + + server_thread = threading.Thread(target=respond_slowly, daemon=True) + server_thread.start() + + try: + client = HttpClient( + name="test", + logger=MagicMock(), + request_timeout=(5, 5), + ) + _, returned_response = client.send_request( + "GET", f"http://127.0.0.1:{port}/", request_kwargs={} + ) + + assert returned_response.status_code == 200 + assert returned_response.json() == {"ok": True} + assert len(accepted_connections) == 1 + finally: + server.close() + server_thread.join(timeout=2) + for connection in accepted_connections: + connection.close() + + def test_send_raises_airbyte_traced_exception_with_fail_response_action(): mocked_session = MagicMock(spec=requests.Session) http_client = HttpClient( From 0638337b9b6be4e912e5de8f273c574007adf50f Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Thu, 17 Sep 2026 16:37:42 +0000 Subject: [PATCH 02/11] docs(http): describe request_timeout in HttpClient docstring Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- airbyte_cdk/sources/streams/http/http_client.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 9faaa26fad..1f7091428f 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -138,10 +138,11 @@ def __init__( message_repository: Optional[MessageRepository] = None, request_timeout: Optional[Union[float, Tuple[float, float]]] = None, ): - """Initialize the HTTP client. + """Initializes the HTTP client. - Args: - request_timeout: The default connect/read timeout in seconds for requests. + `request_timeout` is the `(connect, read)` timeout in seconds (or a single value for both) + applied to every request whose `request_kwargs` do not set `timeout`; it defaults to + `default_request_timeout()`. """ self._name = name self._api_budget: APIBudget = api_budget or APIBudget(policies=[]) From 64c851b5dfd2f0832ca0670dbac9a1b6dadbc2ff Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Wed, 23 Sep 2026 11:30:32 +0000 Subject: [PATCH 03/11] fix(http): apply default request timeout to auth and unstructured-parser requests Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- airbyte_cdk/sources/declarative/auth/token.py | 3 ++ .../file_types/unstructured_parser.py | 7 +++- .../sources/streams/http/http_client.py | 22 ++++------- .../sources/streams/http/request_timeout.py | 22 +++++++++++ .../requests_native_auth/abstract_oauth.py | 2 + .../auth/test_session_token_auth.py | 39 +++++++++++++++++++ .../file_types/test_unstructured_parser.py | 28 ++++++++++++- .../test_requests_native_auth.py | 23 +++++++++++ 8 files changed, 129 insertions(+), 17 deletions(-) create mode 100644 airbyte_cdk/sources/streams/http/request_timeout.py diff --git a/airbyte_cdk/sources/declarative/auth/token.py b/airbyte_cdk/sources/declarative/auth/token.py index caecf9d2ce..6b4533674d 100644 --- a/airbyte_cdk/sources/declarative/auth/token.py +++ b/airbyte_cdk/sources/declarative/auth/token.py @@ -17,6 +17,7 @@ RequestOption, RequestOptionType, ) +from airbyte_cdk.sources.streams.http.request_timeout import default_request_timeout from airbyte_cdk.sources.types import Config @@ -164,6 +165,7 @@ def get_new_session_token(api_url: str, username: str, password: str, response_k f"{api_url}", headers={"Content-Type": "application/json"}, json={"username": username, "password": password}, + timeout=default_request_timeout(), ) response.raise_for_status() if not response.ok: @@ -250,6 +252,7 @@ def is_valid_session_token(self) -> bool: response = requests.get( f"{self._api_url.eval(self.config)}{self._validate_session_url.eval(self.config)}", headers={self.auth_header: self._session_token.eval(self.config)}, + timeout=default_request_timeout(), ) response.raise_for_status() except requests.exceptions.HTTPError as e: diff --git a/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py b/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py index f55675e0ad..2796f6bf34 100644 --- a/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py @@ -36,6 +36,7 @@ from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser from airbyte_cdk.sources.file_based.remote_file import RemoteFile from airbyte_cdk.sources.file_based.schema_helpers import SchemaType +from airbyte_cdk.sources.streams.http.request_timeout import default_request_timeout from airbyte_cdk.utils import is_cloud_environment from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -338,7 +339,11 @@ def _read_file_remotely( file_data = {"files": ("filename", file_handle, FILETYPE_TO_MIMETYPE[filetype])} response = requests.post( - f"{format.api_url}/general/v0/general", headers=headers, data=data, files=file_data + f"{format.api_url}/general/v0/general", + headers=headers, + data=data, + files=file_data, + timeout=default_request_timeout(), ) if response.status_code == 422: diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 1f7091428f..3a250efddf 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -50,6 +50,13 @@ rate_limit_default_backoff_handler, user_defined_backoff_handler, ) +from airbyte_cdk.sources.streams.http.request_timeout import ( + DEFAULT_CONNECT_TIMEOUT_SECONDS, + DEFAULT_READ_TIMEOUT_SECONDS, + ENV_HTTP_CONNECT_TIMEOUT_SECONDS, + ENV_HTTP_READ_TIMEOUT_SECONDS, + default_request_timeout, +) # Imported from the leaf module rather than the package: `protocols` pulls in nothing from the # CDK, so this import cannot cycle no matter what else lands in `requests_native_auth` -- an @@ -72,21 +79,6 @@ MessageRepresentationAirbyteTracedErrors = AirbyteTracedException BODY_REQUEST_METHODS = ("GET", "POST", "PUT", "PATCH") -DEFAULT_CONNECT_TIMEOUT_SECONDS: float = 30.0 -DEFAULT_READ_TIMEOUT_SECONDS: float = 300.0 -ENV_HTTP_CONNECT_TIMEOUT_SECONDS = "AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS" -ENV_HTTP_READ_TIMEOUT_SECONDS = "AIRBYTE_HTTP_READ_TIMEOUT_SECONDS" - - -def default_request_timeout() -> Tuple[float, float]: - """Returns the `(connect, read)` timeout in seconds applied to requests that do not set their own. - - Both values can be overridden with the `AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS` and - `AIRBYTE_HTTP_READ_TIMEOUT_SECONDS` environment variables. - """ - connect = float(os.getenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, DEFAULT_CONNECT_TIMEOUT_SECONDS)) - read = float(os.getenv(ENV_HTTP_READ_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS)) - return (connect, read) def monkey_patched_get_item(self, key): # type: ignore # this interface is a copy/paste from the requests_cache lib diff --git a/airbyte_cdk/sources/streams/http/request_timeout.py b/airbyte_cdk/sources/streams/http/request_timeout.py new file mode 100644 index 0000000000..95dc82ae89 --- /dev/null +++ b/airbyte_cdk/sources/streams/http/request_timeout.py @@ -0,0 +1,22 @@ +# +# Copyright (c) 2023 Airbyte, Inc., all rights reserved. +# + +import os +from typing import Tuple + +DEFAULT_CONNECT_TIMEOUT_SECONDS: float = 30.0 +DEFAULT_READ_TIMEOUT_SECONDS: float = 300.0 +ENV_HTTP_CONNECT_TIMEOUT_SECONDS = "AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS" +ENV_HTTP_READ_TIMEOUT_SECONDS = "AIRBYTE_HTTP_READ_TIMEOUT_SECONDS" + + +def default_request_timeout() -> Tuple[float, float]: + """Returns the `(connect, read)` timeout in seconds applied to requests that do not set their own. + + Both values can be overridden with the `AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS` and + `AIRBYTE_HTTP_READ_TIMEOUT_SECONDS` environment variables. + """ + connect = float(os.getenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, DEFAULT_CONNECT_TIMEOUT_SECONDS)) + read = float(os.getenv(ENV_HTTP_READ_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS)) + return (connect, read) diff --git a/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py b/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py index e236463814..cde7ff5f30 100644 --- a/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py +++ b/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py @@ -17,6 +17,7 @@ from airbyte_cdk.models import FailureType, Level from airbyte_cdk.sources.http_logger import format_http_message from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository +from airbyte_cdk.sources.streams.http.request_timeout import default_request_timeout from airbyte_cdk.utils import AirbyteTracedException from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets, filter_secrets from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse @@ -414,6 +415,7 @@ def _make_handled_request(self) -> Any: data=self.build_refresh_request_body(), headers=self.build_refresh_request_headers(), params=self.build_refresh_request_query_params(), + timeout=default_request_timeout(), ) if not response.ok: diff --git a/unit_tests/sources/declarative/auth/test_session_token_auth.py b/unit_tests/sources/declarative/auth/test_session_token_auth.py index c4b15eb825..59af108f90 100644 --- a/unit_tests/sources/declarative/auth/test_session_token_auth.py +++ b/unit_tests/sources/declarative/auth/test_session_token_auth.py @@ -195,3 +195,42 @@ def test_get_new_session_token(requests_mock): config["session_token_response_key"], ) assert session_token == "some session id" + + +def test_get_new_session_token_applies_default_timeout(mocker): + response = mocker.Mock(ok=True) + response.json.return_value = {"id": "some session id"} + mocked_post = mocker.patch( + "airbyte_cdk.sources.declarative.auth.token.requests.post", return_value=response + ) + + get_new_session_token( + "https://airbyte.example/session-timeout-test", + username, + password, + session_token_response_key, + ) + + assert mocked_post.call_args.kwargs["timeout"] == (30.0, 300.0) + + +def test_is_valid_session_token_applies_default_timeout(mocker): + response = mocker.Mock() + mocked_get = mocker.patch( + "airbyte_cdk.sources.declarative.auth.token.requests.get", return_value=response + ) + authenticator = LegacySessionTokenAuthenticator( + config=config, + parameters=parameters, + api_url=input_instance_api_url, + username=input_username, + password=input_password, + session_token=input_session_token, + header=header, + session_token_response_key=session_token_response_key, + login_url=login_url, + validate_session_url=validate_session_url, + ) + + assert authenticator.is_valid_session_token() + assert mocked_get.call_args.kwargs["timeout"] == (30.0, 300.0) diff --git a/unit_tests/sources/file_based/file_types/test_unstructured_parser.py b/unit_tests/sources/file_based/file_types/test_unstructured_parser.py index 374a02eed4..47930875f9 100644 --- a/unit_tests/sources/file_based/file_types/test_unstructured_parser.py +++ b/unit_tests/sources/file_based/file_types/test_unstructured_parser.py @@ -671,6 +671,32 @@ def test_parse_records_remotely( ) if expected_requests: - requests_mock.post.assert_has_calls(expected_requests) + requests_mock.post.assert_has_calls( + [ + call( + *expected_call.args, + **{**expected_call.kwargs, "timeout": (30.0, 300.0)}, + ) + if expected_call.args + else expected_call + for expected_call in expected_requests + ] + ) else: requests_mock.post.assert_not_called() + + +def test_read_file_remotely_applies_default_timeout(mocker): + response = mocker.Mock(status_code=200) + response.json.return_value = [] + mocked_post = mocker.patch( + "airbyte_cdk.sources.file_based.file_types.unstructured_parser.requests.post", + return_value=response, + ) + format_config = APIProcessingConfigModel(mode="api", api_key="test") + + UnstructuredParser()._read_file_remotely( + MagicMock(), format_config, FileType.PDF, "auto", MagicMock() + ) + + assert mocked_post.call_args.kwargs["timeout"] == (30.0, 300.0) diff --git a/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py b/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py index 98450e7f01..fca6d7bf02 100644 --- a/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py +++ b/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py @@ -1198,6 +1198,29 @@ def test_refresh_access_token(self, mocker, connector_config): "new_refresh_token", ) + def test_refresh_access_token_applies_default_timeout(self, mocker, connector_config): + authenticator = SingleUseRefreshTokenOauth2Authenticator( + connector_config, + token_refresh_endpoint="https://refresh_endpoint.com", + client_id=connector_config["credentials"]["client_id"], + client_secret=connector_config["credentials"]["client_secret"], + ) + resp.status_code = 200 + mocker.patch.object( + resp, + "json", + return_value={ + authenticator.get_access_token_name(): "new_access_token", + authenticator.get_expires_in_name(): "42", + authenticator.get_refresh_token_name(): "new_refresh_token", + }, + ) + mocked_request = mocker.patch.object(requests, "request", return_value=resp) + + authenticator.refresh_access_token() + + assert mocked_request.call_args.kwargs["timeout"] == (30.0, 300.0) + def test_send_refresh_request_as_query_params_picks_up_rotated_refresh_token( self, mocker, connector_config ): From 5353132bf8b926ce6d15478447d3678b98a1f0de Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Wed, 23 Sep 2026 11:33:30 +0000 Subject: [PATCH 04/11] fix(http): remove unused timeout re-exports Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- airbyte_cdk/sources/streams/http/http_client.py | 4 ---- unit_tests/sources/streams/http/test_http_client.py | 4 +++- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 3a250efddf..77c8363a15 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -51,10 +51,6 @@ user_defined_backoff_handler, ) from airbyte_cdk.sources.streams.http.request_timeout import ( - DEFAULT_CONNECT_TIMEOUT_SECONDS, - DEFAULT_READ_TIMEOUT_SECONDS, - ENV_HTTP_CONNECT_TIMEOUT_SECONDS, - ENV_HTTP_READ_TIMEOUT_SECONDS, default_request_timeout, ) diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 1519c19cd7..081ed973ba 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -29,11 +29,13 @@ UserDefinedBackoffException, ) from airbyte_cdk.sources.streams.http.http_client import ( + MessageRepresentationAirbyteTracedErrors, +) +from airbyte_cdk.sources.streams.http.request_timeout import ( DEFAULT_CONNECT_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS, ENV_HTTP_CONNECT_TIMEOUT_SECONDS, ENV_HTTP_READ_TIMEOUT_SECONDS, - MessageRepresentationAirbyteTracedErrors, ) from airbyte_cdk.sources.streams.http.requests_native_auth import TokenAuthenticator from airbyte_cdk.utils.traced_exception import AirbyteTracedException From 0c915f3b819fc47432a75eddf3d01263c4c393e3 Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Fri, 25 Sep 2026 05:01:14 +0000 Subject: [PATCH 05/11] fix(file-based): bound only the connect phase of the Unstructured API call Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../file_types/unstructured_parser.py | 9 +++- .../sources/streams/http/request_timeout.py | 11 +++++ .../file_types/test_unstructured_parser.py | 47 +++++++++++++++++-- .../streams/http/test_request_timeout.py | 18 +++++++ 4 files changed, 80 insertions(+), 5 deletions(-) create mode 100644 unit_tests/sources/streams/http/test_request_timeout.py diff --git a/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py b/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py index 2796f6bf34..c5a159311b 100644 --- a/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py @@ -36,7 +36,7 @@ from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser from airbyte_cdk.sources.file_based.remote_file import RemoteFile from airbyte_cdk.sources.file_based.schema_helpers import SchemaType -from airbyte_cdk.sources.streams.http.request_timeout import default_request_timeout +from airbyte_cdk.sources.streams.http.request_timeout import connect_only_request_timeout from airbyte_cdk.utils import is_cloud_environment from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -338,12 +338,17 @@ def _read_file_remotely( file_data = {"files": ("filename", file_handle, FILETYPE_TO_MIMETYPE[filetype])} + # The hosted partition endpoint is synchronous: the response arrives only once the whole + # document has been processed, so a read timeout would be a limit on document processing + # time (hi_res runs at roughly seconds per page). Only the connect phase is bounded. + # A ReadTimeout is also a RequestException, which the max_tries=5 backoff wrapper around + # this call would retry by re-uploading the whole document up to five times. response = requests.post( f"{format.api_url}/general/v0/general", headers=headers, data=data, files=file_data, - timeout=default_request_timeout(), + timeout=connect_only_request_timeout(), ) if response.status_code == 422: diff --git a/airbyte_cdk/sources/streams/http/request_timeout.py b/airbyte_cdk/sources/streams/http/request_timeout.py index 95dc82ae89..d784a7123c 100644 --- a/airbyte_cdk/sources/streams/http/request_timeout.py +++ b/airbyte_cdk/sources/streams/http/request_timeout.py @@ -20,3 +20,14 @@ def default_request_timeout() -> Tuple[float, float]: connect = float(os.getenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, DEFAULT_CONNECT_TIMEOUT_SECONDS)) read = float(os.getenv(ENV_HTTP_READ_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS)) return (connect, read) + + +def connect_only_request_timeout() -> Tuple[float, None]: + """Returns a `(connect, None)` timeout: the connection attempt is bounded, the read is not. + + For endpoints whose response time is the server-side processing time of the request + (e.g. synchronous document partitioning), a read timeout would be a processing-time + limit rather than a liveness check. + """ + connect, _ = default_request_timeout() + return (connect, None) diff --git a/unit_tests/sources/file_based/file_types/test_unstructured_parser.py b/unit_tests/sources/file_based/file_types/test_unstructured_parser.py index 47930875f9..96bc8c436b 100644 --- a/unit_tests/sources/file_based/file_types/test_unstructured_parser.py +++ b/unit_tests/sources/file_based/file_types/test_unstructured_parser.py @@ -3,7 +3,9 @@ # import asyncio +import time from datetime import datetime +from io import BytesIO from unittest import mock from unittest.mock import MagicMock, call, mock_open, patch @@ -22,6 +24,7 @@ from airbyte_cdk.sources.file_based.exceptions import RecordParseError from airbyte_cdk.sources.file_based.file_types import UnstructuredParser from airbyte_cdk.sources.file_based.remote_file import RemoteFile +from airbyte_cdk.sources.streams.http.request_timeout import ENV_HTTP_CONNECT_TIMEOUT_SECONDS from airbyte_cdk.utils.traced_exception import AirbyteTracedException FILE_URI = "path/to/file.xyz" @@ -675,7 +678,7 @@ def test_parse_records_remotely( [ call( *expected_call.args, - **{**expected_call.kwargs, "timeout": (30.0, 300.0)}, + **{**expected_call.kwargs, "timeout": (30.0, None)}, ) if expected_call.args else expected_call @@ -686,7 +689,7 @@ def test_parse_records_remotely( requests_mock.post.assert_not_called() -def test_read_file_remotely_applies_default_timeout(mocker): +def test_read_file_remotely_applies_connect_only_timeout(mocker): response = mocker.Mock(status_code=200) response.json.return_value = [] mocked_post = mocker.patch( @@ -699,4 +702,42 @@ def test_read_file_remotely_applies_default_timeout(mocker): MagicMock(), format_config, FileType.PDF, "auto", MagicMock() ) - assert mocked_post.call_args.kwargs["timeout"] == (30.0, 300.0) + assert mocked_post.call_args.kwargs["timeout"] == (30.0, None) + + +def test_read_file_remotely_connect_timeout_honours_env_var(mocker, monkeypatch): + response = mocker.Mock(status_code=200) + response.json.return_value = [] + mocked_post = mocker.patch( + "airbyte_cdk.sources.file_based.file_types.unstructured_parser.requests.post", + return_value=response, + ) + monkeypatch.setenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, "5") + format_config = APIProcessingConfigModel(mode="api", api_key="test") + + UnstructuredParser()._read_file_remotely( + MagicMock(), format_config, FileType.PDF, "auto", MagicMock() + ) + + assert mocked_post.call_args.kwargs["timeout"] == (5.0, None) + + +def test_read_file_remotely_hanging_connect_fails_fast(monkeypatch): + monkeypatch.setenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, "0.5") + format_config = APIProcessingConfigModel( + mode="api", + api_key="test", + api_url="http://10.255.255.1:9", + ) + + started_at = time.monotonic() + try: + UnstructuredParser()._read_file_remotely( + BytesIO(b"document"), format_config, FileType.PDF, "auto", MagicMock() + ) + except requests.exceptions.ConnectTimeout: + assert time.monotonic() - started_at < 5 + except requests.exceptions.ConnectionError as exc: + pytest.skip(f"Sandbox refused the connection before timeout: {exc}") + else: + pytest.fail("Expected the non-routable connection to fail") diff --git a/unit_tests/sources/streams/http/test_request_timeout.py b/unit_tests/sources/streams/http/test_request_timeout.py new file mode 100644 index 0000000000..6a9f9b6861 --- /dev/null +++ b/unit_tests/sources/streams/http/test_request_timeout.py @@ -0,0 +1,18 @@ +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. + +from airbyte_cdk.sources.streams.http.request_timeout import ( + ENV_HTTP_CONNECT_TIMEOUT_SECONDS, + connect_only_request_timeout, +) + + +def test_connect_only_request_timeout_defaults(monkeypatch): + monkeypatch.delenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, raising=False) + + assert connect_only_request_timeout() == (30.0, None) + + +def test_connect_only_request_timeout_honours_env_var(monkeypatch): + monkeypatch.setenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, "7") + + assert connect_only_request_timeout() == (7.0, None) From 20ec792924fb2f6427b4edcbb1d9ecca8991115b Mon Sep 17 00:00:00 2001 From: ZaneHyattAB Date: Mon, 5 Oct 2026 12:50:32 -0700 Subject: [PATCH 06/11] refactor(http): drop timeout env var overrides, keep fixed defaults Nothing sets AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS / AIRBYTE_HTTP_READ_TIMEOUT_SECONDS in Cloud and users cannot, so they were not a usable override. Removing them also removes an uncaught ValueError on an empty or non-numeric value. Overrides remain per request (request_kwargs['timeout']) and per client (HttpClient(request_timeout=...)). Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/sources/streams/http/request_timeout.py | 13 ++----------- .../file_types/test_unstructured_parser.py | 8 ++++---- unit_tests/sources/streams/http/test_http_client.py | 10 +++------- .../sources/streams/http/test_request_timeout.py | 12 ++++++------ 4 files changed, 15 insertions(+), 28 deletions(-) diff --git a/airbyte_cdk/sources/streams/http/request_timeout.py b/airbyte_cdk/sources/streams/http/request_timeout.py index d784a7123c..1016a54ae8 100644 --- a/airbyte_cdk/sources/streams/http/request_timeout.py +++ b/airbyte_cdk/sources/streams/http/request_timeout.py @@ -2,24 +2,15 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # -import os from typing import Tuple DEFAULT_CONNECT_TIMEOUT_SECONDS: float = 30.0 DEFAULT_READ_TIMEOUT_SECONDS: float = 300.0 -ENV_HTTP_CONNECT_TIMEOUT_SECONDS = "AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS" -ENV_HTTP_READ_TIMEOUT_SECONDS = "AIRBYTE_HTTP_READ_TIMEOUT_SECONDS" def default_request_timeout() -> Tuple[float, float]: - """Returns the `(connect, read)` timeout in seconds applied to requests that do not set their own. - - Both values can be overridden with the `AIRBYTE_HTTP_CONNECT_TIMEOUT_SECONDS` and - `AIRBYTE_HTTP_READ_TIMEOUT_SECONDS` environment variables. - """ - connect = float(os.getenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, DEFAULT_CONNECT_TIMEOUT_SECONDS)) - read = float(os.getenv(ENV_HTTP_READ_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS)) - return (connect, read) + """Returns the `(connect, read)` timeout in seconds applied to requests that do not set their own.""" + return (DEFAULT_CONNECT_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS) def connect_only_request_timeout() -> Tuple[float, None]: diff --git a/unit_tests/sources/file_based/file_types/test_unstructured_parser.py b/unit_tests/sources/file_based/file_types/test_unstructured_parser.py index 96bc8c436b..7cf4cb25e2 100644 --- a/unit_tests/sources/file_based/file_types/test_unstructured_parser.py +++ b/unit_tests/sources/file_based/file_types/test_unstructured_parser.py @@ -24,7 +24,7 @@ from airbyte_cdk.sources.file_based.exceptions import RecordParseError from airbyte_cdk.sources.file_based.file_types import UnstructuredParser from airbyte_cdk.sources.file_based.remote_file import RemoteFile -from airbyte_cdk.sources.streams.http.request_timeout import ENV_HTTP_CONNECT_TIMEOUT_SECONDS +from airbyte_cdk.sources.streams.http import request_timeout from airbyte_cdk.utils.traced_exception import AirbyteTracedException FILE_URI = "path/to/file.xyz" @@ -705,14 +705,14 @@ def test_read_file_remotely_applies_connect_only_timeout(mocker): assert mocked_post.call_args.kwargs["timeout"] == (30.0, None) -def test_read_file_remotely_connect_timeout_honours_env_var(mocker, monkeypatch): +def test_read_file_remotely_connect_timeout_follows_default(mocker, monkeypatch): response = mocker.Mock(status_code=200) response.json.return_value = [] mocked_post = mocker.patch( "airbyte_cdk.sources.file_based.file_types.unstructured_parser.requests.post", return_value=response, ) - monkeypatch.setenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, "5") + monkeypatch.setattr(request_timeout, "DEFAULT_CONNECT_TIMEOUT_SECONDS", 5.0) format_config = APIProcessingConfigModel(mode="api", api_key="test") UnstructuredParser()._read_file_remotely( @@ -723,7 +723,7 @@ def test_read_file_remotely_connect_timeout_honours_env_var(mocker, monkeypatch) def test_read_file_remotely_hanging_connect_fails_fast(monkeypatch): - monkeypatch.setenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, "0.5") + monkeypatch.setattr(request_timeout, "DEFAULT_CONNECT_TIMEOUT_SECONDS", 0.5) format_config = APIProcessingConfigModel( mode="api", api_key="test", diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index c81ea38d5c..aa5334e830 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -39,8 +39,6 @@ from airbyte_cdk.sources.streams.http.request_timeout import ( DEFAULT_CONNECT_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS, - ENV_HTTP_CONNECT_TIMEOUT_SECONDS, - ENV_HTTP_READ_TIMEOUT_SECONDS, ) from airbyte_cdk.sources.streams.http.request_window_split_exception import ( RequestWindowSplitRequiredException, @@ -269,15 +267,13 @@ def test_constructor_request_timeout_overrides_default(): assert session.send.call_args.kwargs["timeout"] == (1, 2) -def test_env_vars_override_default_timeout(monkeypatch): - monkeypatch.setenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, "5") - monkeypatch.setenv(ENV_HTTP_READ_TIMEOUT_SECONDS, "9") +def test_constructor_single_value_request_timeout_is_passed_through(): session = _mock_success_session() - client = HttpClient(name="test", logger=MagicMock(), session=session) + client = HttpClient(name="test", logger=MagicMock(), session=session, request_timeout=10) client.send_request("GET", "https://test.example", request_kwargs={}) - assert session.send.call_args.kwargs["timeout"] == (5.0, 9.0) + assert session.send.call_args.kwargs["timeout"] == 10 def test_hanging_server_times_out_retries_and_raises(): diff --git a/unit_tests/sources/streams/http/test_request_timeout.py b/unit_tests/sources/streams/http/test_request_timeout.py index 6a9f9b6861..3a39d26e16 100644 --- a/unit_tests/sources/streams/http/test_request_timeout.py +++ b/unit_tests/sources/streams/http/test_request_timeout.py @@ -1,18 +1,18 @@ # Copyright (c) 2025 Airbyte, Inc., all rights reserved. +from airbyte_cdk.sources.streams.http import request_timeout from airbyte_cdk.sources.streams.http.request_timeout import ( - ENV_HTTP_CONNECT_TIMEOUT_SECONDS, connect_only_request_timeout, + default_request_timeout, ) -def test_connect_only_request_timeout_defaults(monkeypatch): - monkeypatch.delenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, raising=False) - +def test_request_timeout_defaults(): + assert default_request_timeout() == (30.0, 300.0) assert connect_only_request_timeout() == (30.0, None) -def test_connect_only_request_timeout_honours_env_var(monkeypatch): - monkeypatch.setenv(ENV_HTTP_CONNECT_TIMEOUT_SECONDS, "7") +def test_connect_only_request_timeout_follows_default_connect_timeout(monkeypatch): + monkeypatch.setattr(request_timeout, "DEFAULT_CONNECT_TIMEOUT_SECONDS", 7.0) assert connect_only_request_timeout() == (7.0, None) From d7ea733387f0dd70ff3f28e2df80dab7d8f99172 Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Wed, 7 Oct 2026 18:23:34 +0000 Subject: [PATCH 07/11] revert(http): remove fleet-wide default timeouts --- airbyte_cdk/sources/declarative/auth/token.py | 3 - .../file_types/unstructured_parser.py | 12 +- .../error_handlers/default_error_mapping.py | 24 +-- .../sources/streams/http/http_client.py | 15 -- .../sources/streams/http/request_timeout.py | 24 --- .../requests_native_auth/abstract_oauth.py | 2 - .../auth/test_session_token_auth.py | 39 ----- .../test_default_error_handler.py | 10 -- .../file_types/test_unstructured_parser.py | 69 +------- .../test_http_status_error_handler.py | 9 - .../test_requests_native_auth.py | 23 --- .../sources/streams/http/test_http_client.py | 163 +----------------- .../streams/http/test_request_timeout.py | 18 -- 13 files changed, 4 insertions(+), 407 deletions(-) delete mode 100644 airbyte_cdk/sources/streams/http/request_timeout.py delete mode 100644 unit_tests/sources/streams/http/test_request_timeout.py diff --git a/airbyte_cdk/sources/declarative/auth/token.py b/airbyte_cdk/sources/declarative/auth/token.py index 6b4533674d..caecf9d2ce 100644 --- a/airbyte_cdk/sources/declarative/auth/token.py +++ b/airbyte_cdk/sources/declarative/auth/token.py @@ -17,7 +17,6 @@ RequestOption, RequestOptionType, ) -from airbyte_cdk.sources.streams.http.request_timeout import default_request_timeout from airbyte_cdk.sources.types import Config @@ -165,7 +164,6 @@ def get_new_session_token(api_url: str, username: str, password: str, response_k f"{api_url}", headers={"Content-Type": "application/json"}, json={"username": username, "password": password}, - timeout=default_request_timeout(), ) response.raise_for_status() if not response.ok: @@ -252,7 +250,6 @@ def is_valid_session_token(self) -> bool: response = requests.get( f"{self._api_url.eval(self.config)}{self._validate_session_url.eval(self.config)}", headers={self.auth_header: self._session_token.eval(self.config)}, - timeout=default_request_timeout(), ) response.raise_for_status() except requests.exceptions.HTTPError as e: diff --git a/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py b/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py index c5a159311b..f55675e0ad 100644 --- a/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/unstructured_parser.py @@ -36,7 +36,6 @@ from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser from airbyte_cdk.sources.file_based.remote_file import RemoteFile from airbyte_cdk.sources.file_based.schema_helpers import SchemaType -from airbyte_cdk.sources.streams.http.request_timeout import connect_only_request_timeout from airbyte_cdk.utils import is_cloud_environment from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -338,17 +337,8 @@ def _read_file_remotely( file_data = {"files": ("filename", file_handle, FILETYPE_TO_MIMETYPE[filetype])} - # The hosted partition endpoint is synchronous: the response arrives only once the whole - # document has been processed, so a read timeout would be a limit on document processing - # time (hi_res runs at roughly seconds per page). Only the connect phase is bounded. - # A ReadTimeout is also a RequestException, which the max_tries=5 backoff wrapper around - # this call would retry by re-uploading the whole document up to five times. response = requests.post( - f"{format.api_url}/general/v0/general", - headers=headers, - data=data, - files=file_data, - timeout=connect_only_request_timeout(), + f"{format.api_url}/general/v0/general", headers=headers, data=data, files=file_data ) if response.status_code == 422: diff --git a/airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py b/airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py index b5b075b44d..45716768fd 100644 --- a/airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py +++ b/airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py @@ -4,14 +4,7 @@ from typing import Mapping, Type, Union -from requests.exceptions import ( - ConnectTimeout, - InvalidSchema, - InvalidURL, - ReadTimeout, - RequestException, - Timeout, -) +from requests.exceptions import InvalidSchema, InvalidURL, RequestException from airbyte_cdk.models import FailureType from airbyte_cdk.sources.streams.http.error_handlers.response_models import ( @@ -30,21 +23,6 @@ failure_type=FailureType.transient_error, error_message="Invalid URL specified or DNS error occurred: The endpoint that data is being requested from is not a valid URL. Exception: requests.exceptions.InvalidURL", ), - ConnectTimeout: ErrorResolution( - response_action=ResponseAction.RETRY, - failure_type=FailureType.transient_error, - error_message="The request timed out while connecting to the server. Exception: requests.exceptions.ConnectTimeout", - ), - ReadTimeout: ErrorResolution( - response_action=ResponseAction.RETRY, - failure_type=FailureType.transient_error, - error_message="The request timed out while waiting for the server to respond. Exception: requests.exceptions.ReadTimeout", - ), - Timeout: ErrorResolution( - response_action=ResponseAction.RETRY, - failure_type=FailureType.transient_error, - error_message="The request timed out. Exception: requests.exceptions.Timeout", - ), RequestException: ErrorResolution( response_action=ResponseAction.RETRY, failure_type=FailureType.transient_error, diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 6a7a1db785..50d03039cc 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -54,9 +54,6 @@ rate_limit_default_backoff_handler, user_defined_backoff_handler, ) -from airbyte_cdk.sources.streams.http.request_timeout import ( - default_request_timeout, -) from airbyte_cdk.sources.streams.http.request_window_split_exception import ( RequestWindowSplitRequiredException, ) @@ -163,14 +160,7 @@ def __init__( error_message_parser: Optional[ErrorMessageParser] = None, disable_retries: bool = False, message_repository: Optional[MessageRepository] = None, - request_timeout: Optional[Union[float, Tuple[float, float]]] = None, ): - """Initializes the HTTP client. - - `request_timeout` is the `(connect, read)` timeout in seconds (or a single value for both) - applied to every request whose `request_kwargs` do not set `timeout`; it defaults to - `default_request_timeout()`. - """ self._name = name self._api_budget: APIBudget = api_budget or APIBudget(policies=[]) if session: @@ -201,9 +191,6 @@ def __init__( self._disable_retries = disable_retries self._message_repository = message_repository self._authenticator_update_failed = False - self._request_timeout: Union[float, Tuple[float, float]] = ( - request_timeout if request_timeout is not None else default_request_timeout() - ) @property def cache_filename(self) -> str: @@ -467,8 +454,6 @@ def _send( exc: Optional[requests.RequestException] = None try: - if "timeout" not in request_kwargs: - request_kwargs = {**request_kwargs, "timeout": self._request_timeout} response = self._session.send(request, **request_kwargs) except requests.RequestException as e: exc = e diff --git a/airbyte_cdk/sources/streams/http/request_timeout.py b/airbyte_cdk/sources/streams/http/request_timeout.py deleted file mode 100644 index 1016a54ae8..0000000000 --- a/airbyte_cdk/sources/streams/http/request_timeout.py +++ /dev/null @@ -1,24 +0,0 @@ -# -# Copyright (c) 2023 Airbyte, Inc., all rights reserved. -# - -from typing import Tuple - -DEFAULT_CONNECT_TIMEOUT_SECONDS: float = 30.0 -DEFAULT_READ_TIMEOUT_SECONDS: float = 300.0 - - -def default_request_timeout() -> Tuple[float, float]: - """Returns the `(connect, read)` timeout in seconds applied to requests that do not set their own.""" - return (DEFAULT_CONNECT_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS) - - -def connect_only_request_timeout() -> Tuple[float, None]: - """Returns a `(connect, None)` timeout: the connection attempt is bounded, the read is not. - - For endpoints whose response time is the server-side processing time of the request - (e.g. synchronous document partitioning), a read timeout would be a processing-time - limit rather than a liveness check. - """ - connect, _ = default_request_timeout() - return (connect, None) diff --git a/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py b/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py index 2d431f1b13..8fa300b176 100644 --- a/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py +++ b/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py @@ -17,7 +17,6 @@ from airbyte_cdk.models import FailureType, Level from airbyte_cdk.sources.http_logger import format_http_message from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository -from airbyte_cdk.sources.streams.http.request_timeout import default_request_timeout from airbyte_cdk.utils import AirbyteTracedException from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets, filter_secrets from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse @@ -435,7 +434,6 @@ def _make_handled_request(self) -> Any: data=self.build_refresh_request_body(), headers=self.build_refresh_request_headers(), params=self.build_refresh_request_query_params(), - timeout=default_request_timeout(), ) if not response.ok: diff --git a/unit_tests/sources/declarative/auth/test_session_token_auth.py b/unit_tests/sources/declarative/auth/test_session_token_auth.py index 59af108f90..c4b15eb825 100644 --- a/unit_tests/sources/declarative/auth/test_session_token_auth.py +++ b/unit_tests/sources/declarative/auth/test_session_token_auth.py @@ -195,42 +195,3 @@ def test_get_new_session_token(requests_mock): config["session_token_response_key"], ) assert session_token == "some session id" - - -def test_get_new_session_token_applies_default_timeout(mocker): - response = mocker.Mock(ok=True) - response.json.return_value = {"id": "some session id"} - mocked_post = mocker.patch( - "airbyte_cdk.sources.declarative.auth.token.requests.post", return_value=response - ) - - get_new_session_token( - "https://airbyte.example/session-timeout-test", - username, - password, - session_token_response_key, - ) - - assert mocked_post.call_args.kwargs["timeout"] == (30.0, 300.0) - - -def test_is_valid_session_token_applies_default_timeout(mocker): - response = mocker.Mock() - mocked_get = mocker.patch( - "airbyte_cdk.sources.declarative.auth.token.requests.get", return_value=response - ) - authenticator = LegacySessionTokenAuthenticator( - config=config, - parameters=parameters, - api_url=input_instance_api_url, - username=input_username, - password=input_password, - session_token=input_session_token, - header=header, - session_token_response_key=session_token_response_key, - login_url=login_url, - validate_session_url=validate_session_url, - ) - - assert authenticator.is_valid_session_token() - assert mocked_get.call_args.kwargs["timeout"] == (30.0, 300.0) diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py index f07d2ac6e3..526e5760fc 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py @@ -6,7 +6,6 @@ import pytest import requests -from requests.exceptions import ReadTimeout from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy import ( ConstantBackoffStrategy, @@ -292,15 +291,6 @@ def test_default_error_handler_with_unmapped_http_code(): assert actual_error_resolution.response_action == ResponseAction.RETRY -def test_default_error_handler_with_read_timeout(): - error_handler = DefaultErrorHandler(config={}, parameters={}) - - actual_error_resolution = error_handler.interpret_response(ReadTimeout()) - - assert actual_error_resolution.response_action == ResponseAction.RETRY - assert actual_error_resolution.failure_type == FailureType.transient_error - - def test_predicate_takes_precedent_over_default_mapped_error(): response_mock = create_response(404, json_body={"error": "test"}) diff --git a/unit_tests/sources/file_based/file_types/test_unstructured_parser.py b/unit_tests/sources/file_based/file_types/test_unstructured_parser.py index 7cf4cb25e2..374a02eed4 100644 --- a/unit_tests/sources/file_based/file_types/test_unstructured_parser.py +++ b/unit_tests/sources/file_based/file_types/test_unstructured_parser.py @@ -3,9 +3,7 @@ # import asyncio -import time from datetime import datetime -from io import BytesIO from unittest import mock from unittest.mock import MagicMock, call, mock_open, patch @@ -24,7 +22,6 @@ from airbyte_cdk.sources.file_based.exceptions import RecordParseError from airbyte_cdk.sources.file_based.file_types import UnstructuredParser from airbyte_cdk.sources.file_based.remote_file import RemoteFile -from airbyte_cdk.sources.streams.http import request_timeout from airbyte_cdk.utils.traced_exception import AirbyteTracedException FILE_URI = "path/to/file.xyz" @@ -674,70 +671,6 @@ def test_parse_records_remotely( ) if expected_requests: - requests_mock.post.assert_has_calls( - [ - call( - *expected_call.args, - **{**expected_call.kwargs, "timeout": (30.0, None)}, - ) - if expected_call.args - else expected_call - for expected_call in expected_requests - ] - ) + requests_mock.post.assert_has_calls(expected_requests) else: requests_mock.post.assert_not_called() - - -def test_read_file_remotely_applies_connect_only_timeout(mocker): - response = mocker.Mock(status_code=200) - response.json.return_value = [] - mocked_post = mocker.patch( - "airbyte_cdk.sources.file_based.file_types.unstructured_parser.requests.post", - return_value=response, - ) - format_config = APIProcessingConfigModel(mode="api", api_key="test") - - UnstructuredParser()._read_file_remotely( - MagicMock(), format_config, FileType.PDF, "auto", MagicMock() - ) - - assert mocked_post.call_args.kwargs["timeout"] == (30.0, None) - - -def test_read_file_remotely_connect_timeout_follows_default(mocker, monkeypatch): - response = mocker.Mock(status_code=200) - response.json.return_value = [] - mocked_post = mocker.patch( - "airbyte_cdk.sources.file_based.file_types.unstructured_parser.requests.post", - return_value=response, - ) - monkeypatch.setattr(request_timeout, "DEFAULT_CONNECT_TIMEOUT_SECONDS", 5.0) - format_config = APIProcessingConfigModel(mode="api", api_key="test") - - UnstructuredParser()._read_file_remotely( - MagicMock(), format_config, FileType.PDF, "auto", MagicMock() - ) - - assert mocked_post.call_args.kwargs["timeout"] == (5.0, None) - - -def test_read_file_remotely_hanging_connect_fails_fast(monkeypatch): - monkeypatch.setattr(request_timeout, "DEFAULT_CONNECT_TIMEOUT_SECONDS", 0.5) - format_config = APIProcessingConfigModel( - mode="api", - api_key="test", - api_url="http://10.255.255.1:9", - ) - - started_at = time.monotonic() - try: - UnstructuredParser()._read_file_remotely( - BytesIO(b"document"), format_config, FileType.PDF, "auto", MagicMock() - ) - except requests.exceptions.ConnectTimeout: - assert time.monotonic() - started_at < 5 - except requests.exceptions.ConnectionError as exc: - pytest.skip(f"Sandbox refused the connection before timeout: {exc}") - else: - pytest.fail("Expected the non-routable connection to fail") diff --git a/unit_tests/sources/streams/http/error_handlers/test_http_status_error_handler.py b/unit_tests/sources/streams/http/error_handlers/test_http_status_error_handler.py index 1b468c0661..b49bc17f89 100644 --- a/unit_tests/sources/streams/http/error_handlers/test_http_status_error_handler.py +++ b/unit_tests/sources/streams/http/error_handlers/test_http_status_error_handler.py @@ -5,7 +5,6 @@ import pytest import requests -from requests.exceptions import ConnectTimeout, ReadTimeout, Timeout from airbyte_cdk.models import FailureType from airbyte_cdk.sources.streams.http.error_handlers import ( @@ -83,14 +82,6 @@ def test_given_requests_exception_returns_retry_action_as_transient_error(): assert error_resolution.failure_type -@pytest.mark.parametrize("exception", [ReadTimeout(), ConnectTimeout(), Timeout()]) -def test_given_timeout_exception_returns_retry_action_as_transient_error(exception): - error_resolution = HttpStatusErrorHandler(logger).interpret_response(exception) - - assert error_resolution.response_action == ResponseAction.RETRY - assert error_resolution.failure_type == FailureType.transient_error - - def test_given_unmapped_exception_returns_retry_action_as_system_error(): error_resolution = HttpStatusErrorHandler(logger).interpret_response(Exception()) diff --git a/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py b/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py index 41b2e99c45..923292cc25 100644 --- a/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py +++ b/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py @@ -1495,29 +1495,6 @@ def test_set_token_expiry_date_stores_a_date_string_as_is(self, connector_config assert connector_config["credentials"]["token_expiry_date"] == "2023-01-01T00:00:00Z" assert authenticator.get_token_expiry_date() == AirbyteDateTime(year=2023, month=1, day=1) - def test_refresh_access_token_applies_default_timeout(self, mocker, connector_config): - authenticator = SingleUseRefreshTokenOauth2Authenticator( - connector_config, - token_refresh_endpoint="https://refresh_endpoint.com", - client_id=connector_config["credentials"]["client_id"], - client_secret=connector_config["credentials"]["client_secret"], - ) - resp.status_code = 200 - mocker.patch.object( - resp, - "json", - return_value={ - authenticator.get_access_token_name(): "new_access_token", - authenticator.get_expires_in_name(): "42", - authenticator.get_refresh_token_name(): "new_refresh_token", - }, - ) - mocked_request = mocker.patch.object(requests, "request", return_value=resp) - - authenticator.refresh_access_token() - - assert mocked_request.call_args.kwargs["timeout"] == (30.0, 300.0) - def test_send_refresh_request_as_query_params_picks_up_rotated_refresh_token( self, mocker, connector_config ): diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index eaa1738f44..ca960f95bb 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -4,11 +4,10 @@ import logging import os import re -import socket import threading import time from datetime import timedelta -from unittest.mock import ANY, MagicMock, patch +from unittest.mock import MagicMock, patch import pytest import requests @@ -37,10 +36,6 @@ from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionRequiredException, ) -from airbyte_cdk.sources.streams.http.request_timeout import ( - DEFAULT_CONNECT_TIMEOUT_SECONDS, - DEFAULT_READ_TIMEOUT_SECONDS, -) from airbyte_cdk.sources.streams.http.request_window_split_exception import ( RequestWindowSplitRequiredException, ) @@ -220,162 +215,6 @@ def test_valid_basic_send_request(mocker): assert returned_response == mocked_response -def _mock_success_session(): - session = MagicMock(spec=requests.Session) - session.merge_environment_settings.return_value = {} - session.prepare_request.return_value = requests.Request("GET", "https://test.example").prepare() - response = MagicMock(spec=requests.Response) - response.status_code = 200 - response.ok = True - response.headers = {} - session.send.return_value = response - return session - - -def test_send_applies_default_timeout(): - session = _mock_success_session() - client = HttpClient(name="test", logger=MagicMock(), session=session) - - client.send_request("GET", "https://test.example", request_kwargs={}) - - session.send.assert_called_once_with( - ANY, - timeout=(DEFAULT_CONNECT_TIMEOUT_SECONDS, DEFAULT_READ_TIMEOUT_SECONDS), - ) - - -def test_request_kwargs_timeout_overrides_default(): - session = _mock_success_session() - client = HttpClient(name="test", logger=MagicMock(), session=session) - - client.send_request("GET", "https://test.example", request_kwargs={"timeout": 7}) - client.send_request("GET", "https://test.example", request_kwargs={"timeout": None}) - - assert session.send.call_args_list[0].kwargs["timeout"] == 7 - assert session.send.call_args_list[1].kwargs["timeout"] is None - - -def test_constructor_request_timeout_overrides_default(): - session = _mock_success_session() - client = HttpClient( - name="test", - logger=MagicMock(), - session=session, - request_timeout=(1, 2), - ) - - client.send_request("GET", "https://test.example", request_kwargs={}) - - assert session.send.call_args.kwargs["timeout"] == (1, 2) - - -def test_constructor_single_value_request_timeout_is_passed_through(): - session = _mock_success_session() - client = HttpClient(name="test", logger=MagicMock(), session=session, request_timeout=10) - - client.send_request("GET", "https://test.example", request_kwargs={}) - - assert session.send.call_args.kwargs["timeout"] == 10 - - -def test_hanging_server_times_out_retries_and_raises(): - server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - server.bind(("127.0.0.1", 0)) - server.listen() - server.settimeout(0.1) - port = server.getsockname()[1] - stop_server = threading.Event() - accepted_connections = [] - accepted_three_connections = threading.Event() - - def accept_connections(): - while not stop_server.is_set(): - try: - connection, _ = server.accept() - except TimeoutError: - continue - except OSError: - break - accepted_connections.append(connection) - if len(accepted_connections) >= 3: - accepted_three_connections.set() - - server_thread = threading.Thread(target=accept_connections, daemon=True) - server_thread.start() - - try: - client = HttpClient( - name="test", - logger=MagicMock(), - error_handler=HttpStatusErrorHandler(logger=MagicMock(), max_retries=2), - backoff_strategy=CustomBackoffStrategy(backoff_time_value=0.01), - request_timeout=(5, 0.3), - ) - started_at = time.monotonic() - - with pytest.raises(AirbyteTracedException) as exc_info: - client.send_request("GET", f"http://127.0.0.1:{port}/", request_kwargs={}) - - elapsed = time.monotonic() - started_at - accepted_three_connections.wait(timeout=1) - assert elapsed < 10 - assert len(accepted_connections) >= 3 - assert exc_info.value.failure_type == FailureType.transient_error - assert "ReadTimeout" in (exc_info.value.internal_message or "") - finally: - stop_server.set() - server.close() - server_thread.join(timeout=1) - for connection in accepted_connections: - connection.close() - - -def test_slow_but_responding_server_succeeds(): - server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - server.bind(("127.0.0.1", 0)) - server.listen() - port = server.getsockname()[1] - accepted_connections = [] - response_body = b'{"ok": true}' - response = ( - b"HTTP/1.1 200 OK\r\n" - b"Content-Type: application/json\r\n" - + f"Content-Length: {len(response_body)}\r\n".encode() - + b"Connection: close\r\n" - + b"\r\n" - + response_body - ) - - def respond_slowly(): - connection, _ = server.accept() - accepted_connections.append(connection) - connection.recv(65536) - time.sleep(1.0) - connection.sendall(response) - - server_thread = threading.Thread(target=respond_slowly, daemon=True) - server_thread.start() - - try: - client = HttpClient( - name="test", - logger=MagicMock(), - request_timeout=(5, 5), - ) - _, returned_response = client.send_request( - "GET", f"http://127.0.0.1:{port}/", request_kwargs={} - ) - - assert returned_response.status_code == 200 - assert returned_response.json() == {"ok": True} - assert len(accepted_connections) == 1 - finally: - server.close() - server_thread.join(timeout=2) - for connection in accepted_connections: - connection.close() - - def test_send_raises_airbyte_traced_exception_with_fail_response_action(): mocked_session = MagicMock(spec=requests.Session) http_client = HttpClient( diff --git a/unit_tests/sources/streams/http/test_request_timeout.py b/unit_tests/sources/streams/http/test_request_timeout.py deleted file mode 100644 index 3a39d26e16..0000000000 --- a/unit_tests/sources/streams/http/test_request_timeout.py +++ /dev/null @@ -1,18 +0,0 @@ -# Copyright (c) 2025 Airbyte, Inc., all rights reserved. - -from airbyte_cdk.sources.streams.http import request_timeout -from airbyte_cdk.sources.streams.http.request_timeout import ( - connect_only_request_timeout, - default_request_timeout, -) - - -def test_request_timeout_defaults(): - assert default_request_timeout() == (30.0, 300.0) - assert connect_only_request_timeout() == (30.0, None) - - -def test_connect_only_request_timeout_follows_default_connect_timeout(monkeypatch): - monkeypatch.setattr(request_timeout, "DEFAULT_CONNECT_TIMEOUT_SECONDS", 7.0) - - assert connect_only_request_timeout() == (7.0, None) From 5379652a2389a2fc8d890c414cc7d3bfc63fbfff Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Wed, 7 Oct 2026 18:32:06 +0000 Subject: [PATCH 08/11] feat(low-code): opt-in TCP keepalive and request timeouts on HttpRequester --- .../declarative_component_schema.yaml | 21 +++ .../models/declarative_component_schema.py | 20 ++ .../parsers/model_to_component_factory.py | 3 + .../declarative/requesters/http_requester.py | 10 + .../sources/streams/http/http_client.py | 28 ++- .../sources/streams/http/tcp_keepalive.py | 77 ++++++++ .../test_model_to_component_factory.py | 86 +++++++++ .../requesters/test_http_requester.py | 38 ++++ .../sources/streams/http/test_http_client.py | 117 ++++++++++++ .../streams/http/test_tcp_keepalive.py | 172 ++++++++++++++++++ 10 files changed, 566 insertions(+), 6 deletions(-) create mode 100644 airbyte_cdk/sources/streams/http/tcp_keepalive.py create mode 100644 unit_tests/sources/streams/http/test_tcp_keepalive.py diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index dfabccd6b0..5cf001b1ba 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2778,6 +2778,27 @@ definitions: description: Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records). type: boolean default: false + connect_timeout_in_seconds: + title: Connect Timeout (seconds) + description: Maximum time to establish the connection (TCP + TLS handshake). When unset, no connect timeout is applied. + type: number + exclusiveMinimum: 0 + examples: + - 30 + read_timeout_in_seconds: + title: Read Timeout (seconds) + description: The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes. + type: number + exclusiveMinimum: 0 + examples: + - 300 + use_tcp_keepalive: + title: Use TCP Keepalive + description: Enables TCP keepalive (60 s idle, 10 s probe interval, 6 probes) on this requester's connections, so dead or NAT-dropped connections are detected and fail instead of hanging forever. Does not fail slow but alive servers; use read_timeout_in_seconds to bound those. + type: boolean + default: false + examples: + - true $parameters: type: object additionalProperties: true diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 5581be3757..2596ad8897 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -3204,6 +3204,26 @@ class HttpRequester(BaseModelWithDeprecations): description="Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).", title="Use Cache", ) + connect_timeout_in_seconds: Optional[float] = Field( + None, + description="Maximum time to establish the connection (TCP + TLS handshake). When unset, no connect timeout is applied.", + examples=[30], + gt=0, + title="Connect Timeout (seconds)", + ) + read_timeout_in_seconds: Optional[float] = Field( + None, + description="The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes.", + examples=[300], + gt=0, + title="Read Timeout (seconds)", + ) + use_tcp_keepalive: Optional[bool] = Field( + False, + description="Enables TCP keepalive (60 s idle, 10 s probe interval, 6 probes) on this requester's connections, so dead or NAT-dropped connections are detected and fail instead of hanging forever. Does not fail slow but alive servers; use read_timeout_in_seconds to bound those.", + examples=[True], + title="Use TCP Keepalive", + ) parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 97a1f4ce2d..219e0f70ad 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -2803,6 +2803,9 @@ def create_http_requester( parameters=model.parameters or {}, message_repository=self._message_repository, use_cache=should_use_cache, + connect_timeout_in_seconds=model.connect_timeout_in_seconds, + read_timeout_in_seconds=model.read_timeout_in_seconds, + use_tcp_keepalive=model.use_tcp_keepalive or False, decoder=decoder, stream_response=decoder.is_stream_response() if decoder else False, ) diff --git a/airbyte_cdk/sources/declarative/requesters/http_requester.py b/airbyte_cdk/sources/declarative/requesters/http_requester.py index 3ce4c85408..bd1468bc7b 100644 --- a/airbyte_cdk/sources/declarative/requesters/http_requester.py +++ b/airbyte_cdk/sources/declarative/requesters/http_requester.py @@ -66,6 +66,9 @@ class HttpRequester(Requester): disable_retries: bool = False message_repository: MessageRepository = NoopMessageRepository() use_cache: bool = False + connect_timeout_in_seconds: Optional[float] = None + read_timeout_in_seconds: Optional[float] = None + use_tcp_keepalive: bool = False _exit_on_rate_limit: bool = False stream_response: bool = False decoder: Decoder = field(default_factory=lambda: JsonDecoder(parameters={})) @@ -114,6 +117,13 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None: backoff_strategy=backoff_strategies, disable_retries=self.disable_retries, message_repository=self.message_repository, + request_timeout=( + (self.connect_timeout_in_seconds, self.read_timeout_in_seconds) + if self.connect_timeout_in_seconds is not None + or self.read_timeout_in_seconds is not None + else None + ), + use_tcp_keepalive=self.use_tcp_keepalive, ) @property diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 50d03039cc..8da499aa32 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -65,6 +65,7 @@ ResponseAwareAuthenticator, TokenRotatingAuthenticator, ) +from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter from airbyte_cdk.sources.utils.types import JsonType from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH @@ -160,6 +161,8 @@ def __init__( error_message_parser: Optional[ErrorMessageParser] = None, disable_retries: bool = False, message_repository: Optional[MessageRepository] = None, + request_timeout: Optional[Union[float, Tuple[Optional[float], Optional[float]]]] = None, + use_tcp_keepalive: bool = False, ): self._name = name self._api_budget: APIBudget = api_budget or APIBudget(policies=[]) @@ -168,12 +171,21 @@ def __init__( else: self._use_cache = use_cache self._session = self._request_session() - self._session.mount( - "https://", - requests.adapters.HTTPAdapter( - pool_connections=MAX_CONNECTION_POOL_SIZE, pool_maxsize=MAX_CONNECTION_POOL_SIZE - ), - ) + if use_tcp_keepalive: + tcp_keepalive_adapter = TcpKeepaliveHTTPAdapter( + pool_connections=MAX_CONNECTION_POOL_SIZE, + pool_maxsize=MAX_CONNECTION_POOL_SIZE, + ) + self._session.mount("https://", tcp_keepalive_adapter) + self._session.mount("http://", tcp_keepalive_adapter) + else: + self._session.mount( + "https://", + requests.adapters.HTTPAdapter( + pool_connections=MAX_CONNECTION_POOL_SIZE, + pool_maxsize=MAX_CONNECTION_POOL_SIZE, + ), + ) if isinstance(authenticator, AuthBase): self._session.auth = authenticator self._logger = logger @@ -191,6 +203,7 @@ def __init__( self._disable_retries = disable_retries self._message_repository = message_repository self._authenticator_update_failed = False + self._request_timeout = request_timeout @property def cache_filename(self) -> str: @@ -453,6 +466,9 @@ def _send( response: Optional[requests.Response] = None exc: Optional[requests.RequestException] = None + if self._request_timeout is not None and "timeout" not in request_kwargs: + request_kwargs = {**request_kwargs, "timeout": self._request_timeout} + try: response = self._session.send(request, **request_kwargs) except requests.RequestException as e: diff --git a/airbyte_cdk/sources/streams/http/tcp_keepalive.py b/airbyte_cdk/sources/streams/http/tcp_keepalive.py new file mode 100644 index 0000000000..7693182861 --- /dev/null +++ b/airbyte_cdk/sources/streams/http/tcp_keepalive.py @@ -0,0 +1,77 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +import functools +import socket +from typing import List, Tuple, cast + +import requests +import urllib3.connection + +TCP_KEEPALIVE_IDLE_SECONDS = 60 +TCP_KEEPALIVE_INTERVAL_SECONDS = 10 +TCP_KEEPALIVE_PROBE_COUNT = 6 + + +@functools.lru_cache(maxsize=None) +def tcp_keepalive_socket_options() -> Tuple[Tuple[int, int, int], ...]: + """Socket options enabling TCP keepalive, probed for what this OS supports. + + Returns urllib3's default options plus SO_KEEPALIVE and the platform's idle / + interval / probe-count tuning options (TCP_KEEPIDLE on Linux/Windows, + TCP_KEEPALIVE on macOS). Each option is tried once on a throwaway socket and + dropped if the OS rejects it, because urllib3 applies these at connect time + and an unsupported option would fail every connection. Never raises. + """ + base_options = cast( + List[Tuple[int, int, int]], + list(urllib3.connection.HTTPConnection.default_socket_options), + ) + + idle_option = getattr(socket, "TCP_KEEPIDLE", None) or getattr(socket, "TCP_KEEPALIVE", None) + candidates: List[Tuple[int, int, int]] = [ + (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1), + ] + if idle_option is not None: + candidates.append((socket.IPPROTO_TCP, idle_option, TCP_KEEPALIVE_IDLE_SECONDS)) + for name, value in ( + ("TCP_KEEPINTVL", TCP_KEEPALIVE_INTERVAL_SECONDS), + ("TCP_KEEPCNT", TCP_KEEPALIVE_PROBE_COUNT), + ): + option = getattr(socket, name, None) + if option is not None: + candidates.append((socket.IPPROTO_TCP, option, value)) + + try: + probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + except OSError: + return tuple(base_options) + try: + supported = base_options + for level, optname, value in candidates: + try: + probe.setsockopt(level, optname, value) + except OSError: + if (level, optname) == (socket.SOL_SOCKET, socket.SO_KEEPALIVE): + # Without SO_KEEPALIVE the tuning options are useless. + return tuple(base_options) + continue + supported.append((level, optname, value)) + return tuple(supported) + finally: + probe.close() + + +class TcpKeepaliveHTTPAdapter(requests.adapters.HTTPAdapter): + """HTTPAdapter that enables TCP keepalive on every connection it pools.""" + + def init_poolmanager(self, *args, **pool_kwargs): # type: ignore[no-untyped-def] + pool_kwargs.setdefault("socket_options", tcp_keepalive_socket_options()) + super().init_poolmanager(*args, **pool_kwargs) # type: ignore[no-untyped-call] + + def proxy_manager_for(self, proxy, **proxy_kwargs): # type: ignore[no-untyped-def] + # requests does not forward init_poolmanager kwargs to proxy managers, + # so without this override proxied connections get no keepalive. + proxy_kwargs.setdefault("socket_options", tcp_keepalive_socket_options()) + return super().proxy_manager_for(proxy, **proxy_kwargs) # type: ignore[no-untyped-call] diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 56e1ad76f0..e54f23e8e8 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -274,6 +274,7 @@ from airbyte_cdk.sources.streams.http.requests_native_auth.oauth import ( SingleUseRefreshTokenOauth2Authenticator, ) +from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter from airbyte_cdk.sources.types import StreamSlice from airbyte_cdk.utils import AirbyteTracedException from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse @@ -10096,3 +10097,88 @@ def test_create_file_uploader_with_a_combined_download_target_extractor(): ) assert isinstance(file_uploader.download_target_extractor, CombinedExtractor) + + +@pytest.mark.parametrize( + "manifest_fields, expected_timeout", + [ + pytest.param( + {"connect_timeout_in_seconds": 30, "read_timeout_in_seconds": 300}, + (30.0, 300.0), + id="both_timeouts", + ), + pytest.param( + {"read_timeout_in_seconds": 300}, + (None, 300.0), + id="read_only", + ), + pytest.param( + {"connect_timeout_in_seconds": 30}, + (30.0, None), + id="connect_only", + ), + pytest.param({}, None, id="no_timeouts"), + ], +) +def test_create_http_requester_with_request_timeouts(manifest_fields, expected_timeout): + requester = factory.create_component( + model_type=HttpRequesterModel, + component_definition={ + "type": "HttpRequester", + "url_base": "https://api.test.com", + "path": "/data", + "http_method": "GET", + **manifest_fields, + }, + config=input_config, + name="name", + decoder=None, + ) + + assert requester._http_client._request_timeout == expected_timeout + if expected_timeout is None: + assert ( + type(requester._http_client._session.adapters["https://"]) + is requests.adapters.HTTPAdapter + ) + assert ( + type(requester._http_client._session.adapters["http://"]) + is requests.adapters.HTTPAdapter + ) + + +def test_create_http_requester_with_tcp_keepalive(): + requester = factory.create_component( + model_type=HttpRequesterModel, + component_definition={ + "type": "HttpRequester", + "url_base": "https://api.test.com", + "path": "/data", + "http_method": "GET", + "use_tcp_keepalive": True, + }, + config=input_config, + name="name", + decoder=None, + ) + + assert isinstance(requester._http_client._session.adapters["https://"], TcpKeepaliveHTTPAdapter) + assert isinstance(requester._http_client._session.adapters["http://"], TcpKeepaliveHTTPAdapter) + + +@pytest.mark.parametrize( + "manifest_fields", + [ + pytest.param({"read_timeout_in_seconds": 0}, id="read_zero"), + pytest.param({"connect_timeout_in_seconds": -1}, id="connect_negative"), + pytest.param({"read_timeout_in_seconds": -5}, id="read_negative"), + ], +) +def test_http_requester_rejects_non_positive_timeouts(manifest_fields): + with pytest.raises(ValidationError): + HttpRequesterModel( + type="HttpRequester", + url_base="https://api.test.com", + path="/data", + **manifest_fields, + ) diff --git a/unit_tests/sources/declarative/requesters/test_http_requester.py b/unit_tests/sources/declarative/requesters/test_http_requester.py index 211b348908..88eabad54f 100644 --- a/unit_tests/sources/declarative/requesters/test_http_requester.py +++ b/unit_tests/sources/declarative/requesters/test_http_requester.py @@ -53,6 +53,8 @@ def factory( disable_retries: bool = False, message_repository: Optional[MessageRepository] = None, use_cache: bool = False, + connect_timeout_in_seconds: Optional[float] = None, + read_timeout_in_seconds: Optional[float] = None, ) -> HttpRequester: return HttpRequester( name=name, @@ -68,6 +70,8 @@ def factory( disable_retries=disable_retries, message_repository=message_repository or MagicMock(), use_cache=use_cache, + connect_timeout_in_seconds=connect_timeout_in_seconds, + read_timeout_in_seconds=read_timeout_in_seconds, ) return factory @@ -960,3 +964,37 @@ def test_http_requester_with_mock_api_budget(http_requester_factory, monkeypatch assert response.status_code == 200 assert mock_budget.acquire_call.call_count == 1 + + +def test_default_requester_sends_no_timeout_kwarg(http_requester_factory): + requester = http_requester_factory(url_base="https://example.com", path="test") + + assert requester._http_client._request_timeout is None + + dummy_response = requests.Response() + dummy_response.status_code = 200 + send_mock = MagicMock(return_value=dummy_response) + requester._http_client._session.send = send_mock + + requester.send_request() + + assert send_mock.call_count == 1 + assert "timeout" not in send_mock.call_args.kwargs + + +def test_requester_with_timeouts_sends_timeout_tuple(http_requester_factory): + requester = http_requester_factory( + url_base="https://example.com", + path="test", + connect_timeout_in_seconds=30, + read_timeout_in_seconds=300, + ) + + dummy_response = requests.Response() + dummy_response.status_code = 200 + send_mock = MagicMock(return_value=dummy_response) + requester._http_client._session.send = send_mock + + requester.send_request() + + assert send_mock.call_args.kwargs["timeout"] == (30, 300) diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index ca960f95bb..31d4ff9a1c 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -7,6 +7,7 @@ import threading import time from datetime import timedelta +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from unittest.mock import MagicMock, patch import pytest @@ -16,6 +17,7 @@ from airbyte_cdk.models import FailureType, Level from airbyte_cdk.sources.declarative.auth.oauth import DeclarativeOauth2Authenticator +from airbyte_cdk.sources.http_config import MAX_CONNECTION_POOL_SIZE from airbyte_cdk.sources.http_logger import format_http_message from airbyte_cdk.sources.message import InMemoryMessageRepository from airbyte_cdk.sources.streams.call_rate import CachedLimiterSession, LimiterSession @@ -43,6 +45,7 @@ Oauth2Authenticator, TokenAuthenticator, ) +from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH from airbyte_cdk.utils.datetime_helpers import ab_datetime_now from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -2011,3 +2014,117 @@ def test_given_split_request_window_action_then_log_the_response_as_an_auxiliary "Stream 'test' request window rejected, retrying with a smaller window" ) assert "no records" in logged[0]["http"]["description"] + + +@pytest.mark.parametrize("use_cache", [True, False]) +def test_default_session_mounts_plain_http_adapter(use_cache): + http_client = HttpClient(name="test", logger=MagicMock(), use_cache=use_cache) + + assert type(http_client._session.adapters["https://"]) is requests.adapters.HTTPAdapter + assert not isinstance(http_client._session.adapters["http://"], TcpKeepaliveHTTPAdapter) + assert type(http_client._session.adapters["http://"]) is requests.adapters.HTTPAdapter + assert http_client._session.adapters["https://"]._pool_connections == MAX_CONNECTION_POOL_SIZE + assert http_client._session.adapters["https://"]._pool_maxsize == MAX_CONNECTION_POOL_SIZE + + +@pytest.mark.parametrize("use_cache", [True, False]) +def test_opted_in_session_mounts_tcp_keepalive_adapter(use_cache): + http_client = HttpClient( + name="test", logger=MagicMock(), use_cache=use_cache, use_tcp_keepalive=True + ) + + for prefix in ("https://", "http://"): + adapter = http_client._session.adapters[prefix] + assert isinstance(adapter, TcpKeepaliveHTTPAdapter) + assert adapter._pool_connections == MAX_CONNECTION_POOL_SIZE + assert adapter._pool_maxsize == MAX_CONNECTION_POOL_SIZE + + +def test_passed_in_session_is_not_modified_even_with_tcp_keepalive(): + mocked_session = MagicMock(spec=requests.Session) + HttpClient( + name="test", + logger=MagicMock(), + session=mocked_session, + use_tcp_keepalive=True, + ) + + mocked_session.mount.assert_not_called() + + +def _session_with_200(): + mocked_session = MagicMock(spec=requests.Session) + mocked_response = requests.Response() + mocked_response.status_code = 200 + mocked_session.send.return_value = mocked_response + return mocked_session + + +def test_no_timeout_kwarg_by_default(): + mocked_session = _session_with_200() + http_client = HttpClient(name="test", logger=MagicMock(), session=mocked_session) + + http_client.send_request(http_method="get", url="https://airbyte.io", request_kwargs={}) + + assert "timeout" not in mocked_session.send.call_args.kwargs + + +def test_request_timeout_is_passed_to_session_send(): + mocked_session = _session_with_200() + http_client = HttpClient( + name="test", logger=MagicMock(), session=mocked_session, request_timeout=(5, 10) + ) + + http_client.send_request(http_method="get", url="https://airbyte.io", request_kwargs={}) + + assert mocked_session.send.call_args.kwargs["timeout"] == (5, 10) + + +def test_explicit_timeout_in_request_kwargs_wins(): + mocked_session = _session_with_200() + http_client = HttpClient( + name="test", logger=MagicMock(), session=mocked_session, request_timeout=(5, 10) + ) + + http_client.send_request( + http_method="get", + url="https://airbyte.io", + request_kwargs={"timeout": 1}, + ) + + assert mocked_session.send.call_args.kwargs["timeout"] == 1 + + +def test_read_timeout_is_retried_then_raises(): + attempts = [] + + class SlowHandler(BaseHTTPRequestHandler): + def do_GET(self): + attempts.append(1) + time.sleep(2) + self.send_response(200) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), SlowHandler) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + http_client = HttpClient( + name="test", + logger=MagicMock(), + error_handler=HttpStatusErrorHandler(logger=MagicMock(), max_retries=1), + request_timeout=(1, 0.5), + ) + with pytest.raises(AirbyteTracedException): + http_client.send_request( + http_method="get", + url=f"http://127.0.0.1:{server.server_port}/", + request_kwargs={}, + ) + assert len(attempts) == 2 + finally: + server.shutdown() + server.server_close() diff --git a/unit_tests/sources/streams/http/test_tcp_keepalive.py b/unit_tests/sources/streams/http/test_tcp_keepalive.py new file mode 100644 index 0000000000..87f2b1cda1 --- /dev/null +++ b/unit_tests/sources/streams/http/test_tcp_keepalive.py @@ -0,0 +1,172 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +import socket +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest.mock import MagicMock + +import pytest +import urllib3.connection + +from airbyte_cdk.sources.streams.http.http_client import HttpClient +from airbyte_cdk.sources.streams.http.tcp_keepalive import ( + TCP_KEEPALIVE_IDLE_SECONDS, + TCP_KEEPALIVE_INTERVAL_SECONDS, + TCP_KEEPALIVE_PROBE_COUNT, + TcpKeepaliveHTTPAdapter, + tcp_keepalive_socket_options, +) + + +@pytest.fixture(autouse=True) +def clear_options_cache(): + tcp_keepalive_socket_options.cache_clear() + yield + tcp_keepalive_socket_options.cache_clear() + + +def _ok_server(): + class OkHandler(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(200) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), OkHandler) + threading.Thread(target=server.serve_forever, daemon=True).start() + return server + + +def _capture_socket_options(monkeypatch): + captured_options = [] + real_connect = urllib3.connection.HTTPConnection.connect + + def capture_connect(connection): + real_connect(connection) + # The socket may be closed by the time the caller inspects it, so read + # the options while the connection is still open. + captured_options.append( + ( + connection.sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE), + connection.sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE), + ) + ) + + monkeypatch.setattr(urllib3.connection.HTTPConnection, "connect", capture_connect) + return captured_options + + +def test_socket_options_returns_tuple(): + assert isinstance(tcp_keepalive_socket_options(), tuple) + + +@pytest.mark.skipif(not hasattr(socket, "TCP_KEEPIDLE"), reason="requires TCP_KEEPIDLE") +def test_socket_options_contains_keepalive_tuning(): + options = tcp_keepalive_socket_options() + + assert (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) in options + assert (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) in options + assert (socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, TCP_KEEPALIVE_IDLE_SECONDS) in options + assert ( + socket.IPPROTO_TCP, + socket.TCP_KEEPINTVL, + TCP_KEEPALIVE_INTERVAL_SECONDS, + ) in options + assert (socket.IPPROTO_TCP, socket.TCP_KEEPCNT, TCP_KEEPALIVE_PROBE_COUNT) in options + + +def test_socket_options_degrade_when_idle_option_missing(monkeypatch): + monkeypatch.delattr(socket, "TCP_KEEPIDLE", raising=False) + monkeypatch.delattr(socket, "TCP_KEEPALIVE", raising=False) + + options = tcp_keepalive_socket_options() + + assert (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) in options + assert not any(value == TCP_KEEPALIVE_IDLE_SECONDS for _, _, value in options) + if hasattr(socket, "TCP_KEEPINTVL"): + assert ( + socket.IPPROTO_TCP, + socket.TCP_KEEPINTVL, + TCP_KEEPALIVE_INTERVAL_SECONDS, + ) in options + if hasattr(socket, "TCP_KEEPCNT"): + assert ( + socket.IPPROTO_TCP, + socket.TCP_KEEPCNT, + TCP_KEEPALIVE_PROBE_COUNT, + ) in options + + +def test_socket_options_drop_rejected_option(monkeypatch): + if not hasattr(socket, "TCP_KEEPCNT"): + pytest.skip("requires TCP_KEEPCNT") + + real_socket_cls = socket.socket + + class FakeSocket: + def setsockopt(self, level, optname, value): + if optname == socket.TCP_KEEPCNT: + raise OSError("unsupported") + + def close(self): + pass + + monkeypatch.setattr(socket, "socket", lambda *a, **kw: FakeSocket()) + + options = tcp_keepalive_socket_options() + + assert (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) in options + assert not any(optname == socket.TCP_KEEPCNT for _, optname, _ in options) + monkeypatch.undo() + assert isinstance(real_socket_cls(socket.AF_INET, socket.SOCK_STREAM), real_socket_cls) + + +def test_adapter_passes_options_to_pool_and_proxy_managers(): + adapter = TcpKeepaliveHTTPAdapter() + + expected = tcp_keepalive_socket_options() + assert adapter.poolmanager.connection_pool_kw["socket_options"] == expected + proxy_manager = adapter.proxy_manager_for("http://proxy.invalid:3128") + assert proxy_manager.connection_pool_kw["socket_options"] == expected + + +@pytest.mark.skipif(not hasattr(socket, "TCP_KEEPIDLE"), reason="requires Linux") +def test_opted_in_http_client_connection_has_keepalive_enabled(monkeypatch): + captured_options = _capture_socket_options(monkeypatch) + + server = _ok_server() + try: + http_client = HttpClient(name="test", logger=MagicMock(), use_tcp_keepalive=True) + http_client.send_request( + http_method="GET", + url=f"http://127.0.0.1:{server.server_port}/", + request_kwargs={"stream": True}, + ) + assert captured_options == [(1, TCP_KEEPALIVE_IDLE_SECONDS)] + finally: + server.shutdown() + server.server_close() + + +@pytest.mark.skipif(not hasattr(socket, "TCP_KEEPIDLE"), reason="requires Linux") +def test_default_http_client_connection_has_no_keepalive(monkeypatch): + captured_options = _capture_socket_options(monkeypatch) + + server = _ok_server() + try: + http_client = HttpClient(name="test", logger=MagicMock()) + http_client.send_request( + http_method="GET", + url=f"http://127.0.0.1:{server.server_port}/", + request_kwargs={"stream": True}, + ) + assert captured_options + assert captured_options[0][0] == 0 + finally: + server.shutdown() + server.server_close() From df2201c88fc9dae1ee7d42d4354c750c9eae1892 Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Wed, 7 Oct 2026 18:39:13 +0000 Subject: [PATCH 09/11] fix(http): declare urllib3 as a direct dependency for deptry --- poetry.lock | 74 +++++++++++++++++++++++++------------------------- pyproject.toml | 1 + 2 files changed, 38 insertions(+), 37 deletions(-) diff --git a/poetry.lock b/poetry.lock index dbaf9d8732..92f30930f6 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.1.4 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.5.1 and should not be changed by hand. [[package]] name = "aiohappyeyeballs" @@ -175,11 +175,11 @@ description = "High level compatibility layer for multiple asynchronous event lo optional = false python-versions = ">=3.9" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "anyio-4.9.0-py3-none-any.whl", hash = "sha256:9f76d541cad6e36af7beb62e978876f3b41e3e04f2c1fbf0884604c0a9c4d93c"}, {file = "anyio-4.9.0.tar.gz", hash = "sha256:673c0c244e15788651a4ff38710fea9675823028a6f08a5eda409e0c9840a028"}, ] +markers = {main = "python_version <= \"3.13.0\" and (extra == \"vector-db-based\" or extra == \"manifest-server\")", dev = "python_version <= \"3.13.0\""} [package.dependencies] exceptiongroup = {version = ">=1.0.2", markers = "python_version < \"3.11\""} @@ -681,11 +681,11 @@ description = "Cross-platform colored terminal text." optional = false python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7" groups = ["main", "dev"] -markers = "(platform_system == \"Windows\" or sys_platform == \"win32\") and python_version <= \"3.13.0\"" files = [ {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"}, {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"}, ] +markers = {main = "(sys_platform == \"win32\" or platform_system == \"Windows\") and (extra == \"dev\" or platform_system == \"Windows\") and python_version <= \"3.13.0\"", dev = "(platform_system == \"Windows\" or sys_platform == \"win32\") and python_version <= \"3.13.0\""} [[package]] name = "contourpy" @@ -1181,10 +1181,10 @@ files = [ [package.dependencies] bytecode = [ - {version = ">=0.15.1", markers = "python_version ~= \"3.12.0\""}, - {version = ">=0.14.0", markers = "python_version ~= \"3.11.0\""}, {version = ">=0.13.0", markers = "python_version < \"3.11\""}, {version = ">=0.16.0", markers = "python_version >= \"3.13.0\""}, + {version = ">=0.14.0", markers = "python_version ~= \"3.11.0\""}, + {version = ">=0.15.1", markers = "python_version ~= \"3.12.0\""}, ] envier = ">=0.6.1,<0.7.0" legacy-cgi = {version = ">=2.0.0", markers = "python_version >= \"3.13.0\""} @@ -1338,7 +1338,7 @@ files = [ ] [package.dependencies] -pydantic = ">=1.7.4,<1.8 || >1.8,<1.8.1 || >1.8.1,<2.0.0 || >2.0.0,<2.0.1 || >2.0.1,<2.1.0 || >2.1.0,<3.0.0" +pydantic = ">=1.7.4,!=1.8,!=1.8.1,!=2.0.0,!=2.0.1,!=2.1.0,<3.0.0" starlette = ">=0.40.0,<0.48.0" typing-extensions = ">=4.8.0" @@ -1714,25 +1714,25 @@ files = [ google-auth = ">=2.14.1,<3.0.0" googleapis-common-protos = ">=1.56.2,<2.0.0" grpcio = [ - {version = ">=1.49.1,<2.0dev", optional = true, markers = "python_version >= \"3.11\" and extra == \"grpc\""}, - {version = ">=1.33.2,<2.0dev", optional = true, markers = "python_version < \"3.11\" and extra == \"grpc\""}, + {version = ">=1.33.2,<2.0", optional = true, markers = "python_version < \"3.11\" and extra == \"grpc\""}, + {version = ">=1.49.1,<2.0", optional = true, markers = "python_version >= \"3.11\" and extra == \"grpc\""}, ] grpcio-status = [ - {version = ">=1.49.1,<2.0.dev0", optional = true, markers = "python_version >= \"3.11\" and extra == \"grpc\""}, - {version = ">=1.33.2,<2.0.dev0", optional = true, markers = "extra == \"grpc\""}, + {version = ">=1.33.2,<2.0", optional = true, markers = "extra == \"grpc\""}, + {version = ">=1.49.1,<2.0", optional = true, markers = "python_version >= \"3.11\" and extra == \"grpc\""}, ] proto-plus = [ {version = ">=1.22.3,<2.0.0"}, {version = ">=1.25.0,<2.0.0", markers = "python_version >= \"3.13\""}, ] -protobuf = ">=3.19.5,<3.20.0 || >3.20.0,<3.20.1 || >3.20.1,<4.21.0 || >4.21.0,<4.21.1 || >4.21.1,<4.21.2 || >4.21.2,<4.21.3 || >4.21.3,<4.21.4 || >4.21.4,<4.21.5 || >4.21.5,<7.0.0" +protobuf = ">=3.19.5,!=3.20.0,!=3.20.1,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5,<7.0.0" requests = ">=2.18.0,<3.0.0" [package.extras] -async-rest = ["google-auth[aiohttp] (>=2.35.0,<3.0.dev0)"] -grpc = ["grpcio (>=1.33.2,<2.0dev)", "grpcio (>=1.49.1,<2.0dev) ; python_version >= \"3.11\"", "grpcio-status (>=1.33.2,<2.0.dev0)", "grpcio-status (>=1.49.1,<2.0.dev0) ; python_version >= \"3.11\""] -grpcgcp = ["grpcio-gcp (>=0.2.2,<1.0.dev0)"] -grpcio-gcp = ["grpcio-gcp (>=0.2.2,<1.0.dev0)"] +async-rest = ["google-auth[aiohttp] (>=2.35.0,<3.0)"] +grpc = ["grpcio (>=1.33.2,<2.0)", "grpcio (>=1.49.1,<2.0) ; python_version >= \"3.11\"", "grpcio-status (>=1.33.2,<2.0)", "grpcio-status (>=1.49.1,<2.0) ; python_version >= \"3.11\""] +grpcgcp = ["grpcio-gcp (>=0.2.2,<1.0)"] +grpcio-gcp = ["grpcio-gcp (>=0.2.2,<1.0)"] [[package]] name = "google-auth" @@ -1776,14 +1776,14 @@ files = [ ] [package.dependencies] -google-api-core = {version = ">=1.34.1,<2.0.dev0 || >=2.11.dev0,<3.0.0", extras = ["grpc"]} -google-auth = ">=2.14.1,<2.24.0 || >2.24.0,<2.25.0 || >2.25.0,<3.0.0" +google-api-core = {version = ">=1.34.1,<2.0 || >=2.11.dev0,<3.0.0", extras = ["grpc"]} +google-auth = ">=2.14.1,!=2.24.0,!=2.25.0,<3.0.0" grpc-google-iam-v1 = ">=0.14.0,<1.0.0" proto-plus = [ {version = ">=1.22.3,<2.0.0"}, {version = ">=1.25.0,<2.0.0", markers = "python_version >= \"3.13\""}, ] -protobuf = ">=3.20.2,<4.21.0 || >4.21.0,<4.21.1 || >4.21.1,<4.21.2 || >4.21.2,<4.21.3 || >4.21.3,<4.21.4 || >4.21.4,<4.21.5 || >4.21.5,<7.0.0" +protobuf = ">=3.20.2,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5,<7.0.0" [[package]] name = "googleapis-common-protos" @@ -1800,7 +1800,7 @@ files = [ [package.dependencies] grpcio = {version = ">=1.44.0,<2.0.0", optional = true, markers = "extra == \"grpc\""} -protobuf = ">=3.20.2,<4.21.1 || >4.21.1,<4.21.2 || >4.21.2,<4.21.3 || >4.21.3,<4.21.4 || >4.21.4,<4.21.5 || >4.21.5,<7.0.0" +protobuf = ">=3.20.2,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5,<7.0.0" [package.extras] grpc = ["grpcio (>=1.44.0,<2.0.0)"] @@ -1954,7 +1954,7 @@ files = [ [package.dependencies] googleapis-common-protos = {version = ">=1.56.0,<2.0.0", extras = ["grpc"]} grpcio = ">=1.44.0,<2.0.0" -protobuf = ">=3.20.2,<4.21.1 || >4.21.1,<4.21.2 || >4.21.2,<4.21.3 || >4.21.3,<4.21.4 || >4.21.4,<4.21.5 || >4.21.5,<7.0.0" +protobuf = ">=3.20.2,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5,<7.0.0" [[package]] name = "grpcio" @@ -2046,11 +2046,11 @@ description = "A pure-Python, bring-your-own-I/O implementation of HTTP/1.1" optional = false python-versions = ">=3.8" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86"}, {file = "h11-0.16.0.tar.gz", hash = "sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1"}, ] +markers = {main = "python_version <= \"3.13.0\" and (extra == \"vector-db-based\" or extra == \"manifest-server\")", dev = "python_version <= \"3.13.0\""} [[package]] name = "hf-xet" @@ -2081,11 +2081,11 @@ description = "A minimal low-level HTTP client." optional = false python-versions = ">=3.8" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55"}, {file = "httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8"}, ] +markers = {main = "python_version <= \"3.13.0\" and extra == \"vector-db-based\"", dev = "python_version <= \"3.13.0\""} [package.dependencies] certifi = "*" @@ -2104,11 +2104,11 @@ description = "The next generation HTTP client." optional = false python-versions = ">=3.8" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad"}, {file = "httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc"}, ] +markers = {main = "python_version <= \"3.13.0\" and extra == \"vector-db-based\"", dev = "python_version <= \"3.13.0\""} [package.dependencies] anyio = "*" @@ -2330,11 +2330,11 @@ description = "brain-dead simple config-ini parsing" optional = false python-versions = ">=3.7" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "iniconfig-2.0.0-py3-none-any.whl", hash = "sha256:b6a85871a79d2e3b22d2d1b94ac2824226a63c6b741c88f7ae975f18b6778374"}, {file = "iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3"}, ] +markers = {main = "python_version <= \"3.13.0\" and extra == \"dev\"", dev = "python_version <= \"3.13.0\""} [[package]] name = "isodate" @@ -2441,7 +2441,7 @@ files = [ [package.dependencies] attrs = ">=22.2.0" -jsonschema-specifications = ">=2023.03.6" +jsonschema-specifications = ">=2023.3.6" referencing = ">=0.28.4" rpds-py = ">=0.7.1" @@ -2622,7 +2622,7 @@ pydantic-settings = ">=2.10.1,<3.0.0" PyYAML = ">=5.3.0,<7.0.0" requests = ">=2.32.5,<3.0.0" SQLAlchemy = ">=1.4.0,<3.0.0" -tenacity = ">=8.1.0,<8.4.0 || >8.4.0,<10.0.0" +tenacity = ">=8.1.0,!=8.4.0,<10.0.0" [[package]] name = "langchain-core" @@ -2643,7 +2643,7 @@ langsmith = ">=0.3.45,<1.0.0" packaging = ">=23.2.0,<26.0.0" pydantic = ">=2.7.4,<3.0.0" pyyaml = ">=5.3.0,<7.0.0" -tenacity = ">=8.1.0,<8.4.0 || >8.4.0,<10.0.0" +tenacity = ">=8.1.0,!=8.4.0,<10.0.0" typing-extensions = ">=4.7.0,<5.0.0" uuid-utils = ">=0.12.0,<1.0" @@ -3898,9 +3898,9 @@ files = [ [package.dependencies] numpy = [ - {version = ">=1.26.0", markers = "python_version >= \"3.12\""}, - {version = ">=1.23.2", markers = "python_version == \"3.11\""}, {version = ">=1.22.4", markers = "python_version < \"3.11\""}, + {version = ">=1.23.2", markers = "python_version == \"3.11\""}, + {version = ">=1.26.0", markers = "python_version >= \"3.12\""}, ] python-dateutil = ">=2.8.2" pytz = ">=2020.1" @@ -4149,11 +4149,11 @@ description = "plugin and hook calling mechanisms for python" optional = false python-versions = ">=3.8" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "pluggy-1.5.0-py3-none-any.whl", hash = "sha256:44e1ad92c8ca002de6377e165f3e0f1be63266ab4d554740532335b9d75ea669"}, {file = "pluggy-1.5.0.tar.gz", hash = "sha256:2cffa88e94fdc978c4c574f15f9e59b7f4201d439195c3715ca9e2486f1d0cf1"}, ] +markers = {main = "python_version <= \"3.13.0\" and extra == \"dev\"", dev = "python_version <= \"3.13.0\""} [package.extras] dev = ["pre-commit", "tox"] @@ -4558,7 +4558,7 @@ files = [ ] [package.dependencies] -typing-extensions = ">=4.6.0,<4.7.0 || >4.7.0" +typing-extensions = ">=4.6.0,!=4.7.0" [[package]] name = "pydantic-settings" @@ -4723,11 +4723,11 @@ description = "pytest: simple powerful testing with Python" optional = false python-versions = ">=3.7" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "pytest-7.4.4-py3-none-any.whl", hash = "sha256:b090cdf5ed60bf4c45261be03239c2c1c22df034fbffe691abe93cd80cea01d8"}, {file = "pytest-7.4.4.tar.gz", hash = "sha256:2cf0005922c6ace4a3e2ec8b4080eb0d9753fdc93107415332f50ce9e7994280"}, ] +markers = {main = "python_version <= \"3.13.0\" and extra == \"dev\"", dev = "python_version <= \"3.13.0\""} [package.dependencies] colorama = {version = "*", markers = "sys_platform == \"win32\""} @@ -5960,11 +5960,11 @@ description = "Sniff out which async library your code is running under" optional = false python-versions = ">=3.7" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2"}, {file = "sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc"}, ] +markers = {main = "python_version <= \"3.13.0\" and (extra == \"vector-db-based\" or extra == \"manifest-server\")", dev = "python_version <= \"3.13.0\""} [[package]] name = "soupsieve" @@ -6288,7 +6288,7 @@ files = [ {file = "tomli-2.2.1-py3-none-any.whl", hash = "sha256:cb55c73c5f4408779d0cf3eef9f762b9c9f147a77de7b258bef0a5628adc85cc"}, {file = "tomli-2.2.1.tar.gz", hash = "sha256:cd45e1dc79c835ce60f7404ec8119f2eb06d38b1deba146f07ced3bbc44505ff"}, ] -markers = {main = "python_version == \"3.10\"", dev = "python_version <= \"3.13.0\""} +markers = {main = "extra == \"dev\" and python_version == \"3.10\"", dev = "python_version <= \"3.13.0\""} [[package]] name = "tqdm" @@ -6372,11 +6372,11 @@ description = "Typing stubs for requests" optional = false python-versions = ">=3.8" groups = ["main", "dev"] -markers = "python_version <= \"3.13.0\"" files = [ {file = "types-requests-2.32.0.20241016.tar.gz", hash = "sha256:0d9cad2f27515d0e3e3da7134a1b6f28fb97129d86b867f24d9c726452634d95"}, {file = "types_requests-2.32.0.20241016-py3-none-any.whl", hash = "sha256:4195d62d6d3e043a4eaaf08ff8a62184584d2e8684e9d2aa178c7915a7da3747"}, ] +markers = {main = "python_version <= \"3.13.0\" and extra == \"vector-db-based\"", dev = "python_version <= \"3.13.0\""} [package.dependencies] urllib3 = ">=2" @@ -7161,7 +7161,7 @@ files = [ ] [package.extras] -cffi = ["cffi (>=1.17,<2.0) ; platform_python_implementation != \"PyPy\" and python_version < \"3.14\"", "cffi (>=2.0.0b) ; platform_python_implementation != \"PyPy\" and python_version >= \"3.14\""] +cffi = ["cffi (>=1.17,<2.0) ; platform_python_implementation != \"PyPy\" and python_version < \"3.14\"", "cffi (>=2.0.0b0) ; platform_python_implementation != \"PyPy\" and python_version >= \"3.14\""] [extras] dev = ["pytest"] @@ -7173,4 +7173,4 @@ vector-db-based = ["cohere", "langchain_community", "langchain_core", "langchain [metadata] lock-version = "2.1" python-versions = ">=3.10,<3.14" -content-hash = "16588f0ec5fed9bd99bed0f8b40d13971708a8cf3f9c1a24ab6f4bdb358cd97c" +content-hash = "9bc4751e256775be70984fad5bc335e472ec06d9e19c640cdfc169ca0b82fec9" diff --git a/pyproject.toml b/pyproject.toml index 0abbe73123..a5b89f7617 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,7 @@ requests_cache = "*" rich = "*" # Use whichever version meets other constraints rich-click = "^1.8.8" typing-extensions = "*" # Transitive dependency used directly in code +urllib3 = "*" # Transitive dependency used directly in code wcmatch = "10.0" # Extras depedencies avro = { version = ">=1.11.2,<1.13.0", optional = true } # TODO: Move into dev dependencies if only used in tests From 18ecc5735edf0b35958c81ff53a493e21a14cb95 Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Wed, 7 Oct 2026 19:51:38 +0000 Subject: [PATCH 10/11] fix(deps): relock with Poetry 2.1.4 to avoid unrelated lockfile churn --- poetry.lock | 72 ++++++++++++++++++++++++++--------------------------- 1 file changed, 36 insertions(+), 36 deletions(-) diff --git a/poetry.lock b/poetry.lock index 92f30930f6..b51dea4b72 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.5.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.1.4 and should not be changed by hand. [[package]] name = "aiohappyeyeballs" @@ -175,11 +175,11 @@ description = "High level compatibility layer for multiple asynchronous event lo optional = false python-versions = ">=3.9" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "anyio-4.9.0-py3-none-any.whl", hash = "sha256:9f76d541cad6e36af7beb62e978876f3b41e3e04f2c1fbf0884604c0a9c4d93c"}, {file = "anyio-4.9.0.tar.gz", hash = "sha256:673c0c244e15788651a4ff38710fea9675823028a6f08a5eda409e0c9840a028"}, ] -markers = {main = "python_version <= \"3.13.0\" and (extra == \"vector-db-based\" or extra == \"manifest-server\")", dev = "python_version <= \"3.13.0\""} [package.dependencies] exceptiongroup = {version = ">=1.0.2", markers = "python_version < \"3.11\""} @@ -681,11 +681,11 @@ description = "Cross-platform colored terminal text." optional = false python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7" groups = ["main", "dev"] +markers = "(platform_system == \"Windows\" or sys_platform == \"win32\") and python_version <= \"3.13.0\"" files = [ {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"}, {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"}, ] -markers = {main = "(sys_platform == \"win32\" or platform_system == \"Windows\") and (extra == \"dev\" or platform_system == \"Windows\") and python_version <= \"3.13.0\"", dev = "(platform_system == \"Windows\" or sys_platform == \"win32\") and python_version <= \"3.13.0\""} [[package]] name = "contourpy" @@ -1181,10 +1181,10 @@ files = [ [package.dependencies] bytecode = [ + {version = ">=0.15.1", markers = "python_version ~= \"3.12.0\""}, + {version = ">=0.14.0", markers = "python_version ~= \"3.11.0\""}, {version = ">=0.13.0", markers = "python_version < \"3.11\""}, {version = ">=0.16.0", markers = "python_version >= \"3.13.0\""}, - {version = ">=0.14.0", markers = "python_version ~= \"3.11.0\""}, - {version = ">=0.15.1", markers = "python_version ~= \"3.12.0\""}, ] envier = ">=0.6.1,<0.7.0" legacy-cgi = {version = ">=2.0.0", markers = "python_version >= \"3.13.0\""} @@ -1338,7 +1338,7 @@ files = [ ] [package.dependencies] -pydantic = ">=1.7.4,!=1.8,!=1.8.1,!=2.0.0,!=2.0.1,!=2.1.0,<3.0.0" +pydantic = ">=1.7.4,<1.8 || >1.8,<1.8.1 || >1.8.1,<2.0.0 || >2.0.0,<2.0.1 || >2.0.1,<2.1.0 || >2.1.0,<3.0.0" starlette = ">=0.40.0,<0.48.0" typing-extensions = ">=4.8.0" @@ -1714,25 +1714,25 @@ files = [ google-auth = ">=2.14.1,<3.0.0" googleapis-common-protos = ">=1.56.2,<2.0.0" grpcio = [ - {version = ">=1.33.2,<2.0", optional = true, markers = "python_version < \"3.11\" and extra == \"grpc\""}, - {version = ">=1.49.1,<2.0", optional = true, markers = "python_version >= \"3.11\" and extra == \"grpc\""}, + {version = ">=1.49.1,<2.0dev", optional = true, markers = "python_version >= \"3.11\" and extra == \"grpc\""}, + {version = ">=1.33.2,<2.0dev", optional = true, markers = "python_version < \"3.11\" and extra == \"grpc\""}, ] grpcio-status = [ - {version = ">=1.33.2,<2.0", optional = true, markers = "extra == \"grpc\""}, - {version = ">=1.49.1,<2.0", optional = true, markers = "python_version >= \"3.11\" and extra == \"grpc\""}, + {version = ">=1.49.1,<2.0.dev0", optional = true, markers = "python_version >= \"3.11\" and extra == \"grpc\""}, + {version = ">=1.33.2,<2.0.dev0", optional = true, markers = "extra == \"grpc\""}, ] proto-plus = [ {version = ">=1.22.3,<2.0.0"}, {version = ">=1.25.0,<2.0.0", markers = "python_version >= \"3.13\""}, ] -protobuf = ">=3.19.5,!=3.20.0,!=3.20.1,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5,<7.0.0" +protobuf = ">=3.19.5,<3.20.0 || >3.20.0,<3.20.1 || >3.20.1,<4.21.0 || >4.21.0,<4.21.1 || >4.21.1,<4.21.2 || >4.21.2,<4.21.3 || >4.21.3,<4.21.4 || >4.21.4,<4.21.5 || >4.21.5,<7.0.0" requests = ">=2.18.0,<3.0.0" [package.extras] -async-rest = ["google-auth[aiohttp] (>=2.35.0,<3.0)"] -grpc = ["grpcio (>=1.33.2,<2.0)", "grpcio (>=1.49.1,<2.0) ; python_version >= \"3.11\"", "grpcio-status (>=1.33.2,<2.0)", "grpcio-status (>=1.49.1,<2.0) ; python_version >= \"3.11\""] -grpcgcp = ["grpcio-gcp (>=0.2.2,<1.0)"] -grpcio-gcp = ["grpcio-gcp (>=0.2.2,<1.0)"] +async-rest = ["google-auth[aiohttp] (>=2.35.0,<3.0.dev0)"] +grpc = ["grpcio (>=1.33.2,<2.0dev)", "grpcio (>=1.49.1,<2.0dev) ; python_version >= \"3.11\"", "grpcio-status (>=1.33.2,<2.0.dev0)", "grpcio-status (>=1.49.1,<2.0.dev0) ; python_version >= \"3.11\""] +grpcgcp = ["grpcio-gcp (>=0.2.2,<1.0.dev0)"] +grpcio-gcp = ["grpcio-gcp (>=0.2.2,<1.0.dev0)"] [[package]] name = "google-auth" @@ -1776,14 +1776,14 @@ files = [ ] [package.dependencies] -google-api-core = {version = ">=1.34.1,<2.0 || >=2.11.dev0,<3.0.0", extras = ["grpc"]} -google-auth = ">=2.14.1,!=2.24.0,!=2.25.0,<3.0.0" +google-api-core = {version = ">=1.34.1,<2.0.dev0 || >=2.11.dev0,<3.0.0", extras = ["grpc"]} +google-auth = ">=2.14.1,<2.24.0 || >2.24.0,<2.25.0 || >2.25.0,<3.0.0" grpc-google-iam-v1 = ">=0.14.0,<1.0.0" proto-plus = [ {version = ">=1.22.3,<2.0.0"}, {version = ">=1.25.0,<2.0.0", markers = "python_version >= \"3.13\""}, ] -protobuf = ">=3.20.2,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5,<7.0.0" +protobuf = ">=3.20.2,<4.21.0 || >4.21.0,<4.21.1 || >4.21.1,<4.21.2 || >4.21.2,<4.21.3 || >4.21.3,<4.21.4 || >4.21.4,<4.21.5 || >4.21.5,<7.0.0" [[package]] name = "googleapis-common-protos" @@ -1800,7 +1800,7 @@ files = [ [package.dependencies] grpcio = {version = ">=1.44.0,<2.0.0", optional = true, markers = "extra == \"grpc\""} -protobuf = ">=3.20.2,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5,<7.0.0" +protobuf = ">=3.20.2,<4.21.1 || >4.21.1,<4.21.2 || >4.21.2,<4.21.3 || >4.21.3,<4.21.4 || >4.21.4,<4.21.5 || >4.21.5,<7.0.0" [package.extras] grpc = ["grpcio (>=1.44.0,<2.0.0)"] @@ -1954,7 +1954,7 @@ files = [ [package.dependencies] googleapis-common-protos = {version = ">=1.56.0,<2.0.0", extras = ["grpc"]} grpcio = ">=1.44.0,<2.0.0" -protobuf = ">=3.20.2,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5,<7.0.0" +protobuf = ">=3.20.2,<4.21.1 || >4.21.1,<4.21.2 || >4.21.2,<4.21.3 || >4.21.3,<4.21.4 || >4.21.4,<4.21.5 || >4.21.5,<7.0.0" [[package]] name = "grpcio" @@ -2046,11 +2046,11 @@ description = "A pure-Python, bring-your-own-I/O implementation of HTTP/1.1" optional = false python-versions = ">=3.8" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86"}, {file = "h11-0.16.0.tar.gz", hash = "sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1"}, ] -markers = {main = "python_version <= \"3.13.0\" and (extra == \"vector-db-based\" or extra == \"manifest-server\")", dev = "python_version <= \"3.13.0\""} [[package]] name = "hf-xet" @@ -2081,11 +2081,11 @@ description = "A minimal low-level HTTP client." optional = false python-versions = ">=3.8" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55"}, {file = "httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8"}, ] -markers = {main = "python_version <= \"3.13.0\" and extra == \"vector-db-based\"", dev = "python_version <= \"3.13.0\""} [package.dependencies] certifi = "*" @@ -2104,11 +2104,11 @@ description = "The next generation HTTP client." optional = false python-versions = ">=3.8" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad"}, {file = "httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc"}, ] -markers = {main = "python_version <= \"3.13.0\" and extra == \"vector-db-based\"", dev = "python_version <= \"3.13.0\""} [package.dependencies] anyio = "*" @@ -2330,11 +2330,11 @@ description = "brain-dead simple config-ini parsing" optional = false python-versions = ">=3.7" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "iniconfig-2.0.0-py3-none-any.whl", hash = "sha256:b6a85871a79d2e3b22d2d1b94ac2824226a63c6b741c88f7ae975f18b6778374"}, {file = "iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3"}, ] -markers = {main = "python_version <= \"3.13.0\" and extra == \"dev\"", dev = "python_version <= \"3.13.0\""} [[package]] name = "isodate" @@ -2441,7 +2441,7 @@ files = [ [package.dependencies] attrs = ">=22.2.0" -jsonschema-specifications = ">=2023.3.6" +jsonschema-specifications = ">=2023.03.6" referencing = ">=0.28.4" rpds-py = ">=0.7.1" @@ -2622,7 +2622,7 @@ pydantic-settings = ">=2.10.1,<3.0.0" PyYAML = ">=5.3.0,<7.0.0" requests = ">=2.32.5,<3.0.0" SQLAlchemy = ">=1.4.0,<3.0.0" -tenacity = ">=8.1.0,!=8.4.0,<10.0.0" +tenacity = ">=8.1.0,<8.4.0 || >8.4.0,<10.0.0" [[package]] name = "langchain-core" @@ -2643,7 +2643,7 @@ langsmith = ">=0.3.45,<1.0.0" packaging = ">=23.2.0,<26.0.0" pydantic = ">=2.7.4,<3.0.0" pyyaml = ">=5.3.0,<7.0.0" -tenacity = ">=8.1.0,!=8.4.0,<10.0.0" +tenacity = ">=8.1.0,<8.4.0 || >8.4.0,<10.0.0" typing-extensions = ">=4.7.0,<5.0.0" uuid-utils = ">=0.12.0,<1.0" @@ -3898,9 +3898,9 @@ files = [ [package.dependencies] numpy = [ - {version = ">=1.22.4", markers = "python_version < \"3.11\""}, - {version = ">=1.23.2", markers = "python_version == \"3.11\""}, {version = ">=1.26.0", markers = "python_version >= \"3.12\""}, + {version = ">=1.23.2", markers = "python_version == \"3.11\""}, + {version = ">=1.22.4", markers = "python_version < \"3.11\""}, ] python-dateutil = ">=2.8.2" pytz = ">=2020.1" @@ -4149,11 +4149,11 @@ description = "plugin and hook calling mechanisms for python" optional = false python-versions = ">=3.8" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "pluggy-1.5.0-py3-none-any.whl", hash = "sha256:44e1ad92c8ca002de6377e165f3e0f1be63266ab4d554740532335b9d75ea669"}, {file = "pluggy-1.5.0.tar.gz", hash = "sha256:2cffa88e94fdc978c4c574f15f9e59b7f4201d439195c3715ca9e2486f1d0cf1"}, ] -markers = {main = "python_version <= \"3.13.0\" and extra == \"dev\"", dev = "python_version <= \"3.13.0\""} [package.extras] dev = ["pre-commit", "tox"] @@ -4558,7 +4558,7 @@ files = [ ] [package.dependencies] -typing-extensions = ">=4.6.0,!=4.7.0" +typing-extensions = ">=4.6.0,<4.7.0 || >4.7.0" [[package]] name = "pydantic-settings" @@ -4723,11 +4723,11 @@ description = "pytest: simple powerful testing with Python" optional = false python-versions = ">=3.7" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "pytest-7.4.4-py3-none-any.whl", hash = "sha256:b090cdf5ed60bf4c45261be03239c2c1c22df034fbffe691abe93cd80cea01d8"}, {file = "pytest-7.4.4.tar.gz", hash = "sha256:2cf0005922c6ace4a3e2ec8b4080eb0d9753fdc93107415332f50ce9e7994280"}, ] -markers = {main = "python_version <= \"3.13.0\" and extra == \"dev\"", dev = "python_version <= \"3.13.0\""} [package.dependencies] colorama = {version = "*", markers = "sys_platform == \"win32\""} @@ -5960,11 +5960,11 @@ description = "Sniff out which async library your code is running under" optional = false python-versions = ">=3.7" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2"}, {file = "sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc"}, ] -markers = {main = "python_version <= \"3.13.0\" and (extra == \"vector-db-based\" or extra == \"manifest-server\")", dev = "python_version <= \"3.13.0\""} [[package]] name = "soupsieve" @@ -6288,7 +6288,7 @@ files = [ {file = "tomli-2.2.1-py3-none-any.whl", hash = "sha256:cb55c73c5f4408779d0cf3eef9f762b9c9f147a77de7b258bef0a5628adc85cc"}, {file = "tomli-2.2.1.tar.gz", hash = "sha256:cd45e1dc79c835ce60f7404ec8119f2eb06d38b1deba146f07ced3bbc44505ff"}, ] -markers = {main = "extra == \"dev\" and python_version == \"3.10\"", dev = "python_version <= \"3.13.0\""} +markers = {main = "python_version == \"3.10\"", dev = "python_version <= \"3.13.0\""} [[package]] name = "tqdm" @@ -6372,11 +6372,11 @@ description = "Typing stubs for requests" optional = false python-versions = ">=3.8" groups = ["main", "dev"] +markers = "python_version <= \"3.13.0\"" files = [ {file = "types-requests-2.32.0.20241016.tar.gz", hash = "sha256:0d9cad2f27515d0e3e3da7134a1b6f28fb97129d86b867f24d9c726452634d95"}, {file = "types_requests-2.32.0.20241016-py3-none-any.whl", hash = "sha256:4195d62d6d3e043a4eaaf08ff8a62184584d2e8684e9d2aa178c7915a7da3747"}, ] -markers = {main = "python_version <= \"3.13.0\" and extra == \"vector-db-based\"", dev = "python_version <= \"3.13.0\""} [package.dependencies] urllib3 = ">=2" @@ -7161,7 +7161,7 @@ files = [ ] [package.extras] -cffi = ["cffi (>=1.17,<2.0) ; platform_python_implementation != \"PyPy\" and python_version < \"3.14\"", "cffi (>=2.0.0b0) ; platform_python_implementation != \"PyPy\" and python_version >= \"3.14\""] +cffi = ["cffi (>=1.17,<2.0) ; platform_python_implementation != \"PyPy\" and python_version < \"3.14\"", "cffi (>=2.0.0b) ; platform_python_implementation != \"PyPy\" and python_version >= \"3.14\""] [extras] dev = ["pytest"] From a8e017cd2059f3c78a23a263d2f56d8a4f8cfe20 Mon Sep 17 00:00:00 2001 From: ZaneHyattAB Date: Thu, 8 Oct 2026 16:14:15 -0700 Subject: [PATCH 11/11] fix(http): warn when TCP keepalive is unavailable; note read timeouts use the retry window Address review nits on #1168: - Log a warning (once per process, via the cached options) when the probe socket can't be created or SO_KEEPALIVE is rejected, instead of silently falling back to plain sockets. - Note in the read_timeout_in_seconds description that each timed-out attempt counts against the error handler's retry window (600 s for manifest connectors), so a value near 300 allows only a few retries. Co-Authored-By: Claude Opus 5.5 --- .../declarative_component_schema.yaml | 2 +- .../models/declarative_component_schema.py | 2 +- .../sources/streams/http/tcp_keepalive.py | 16 +++++++- .../streams/http/test_tcp_keepalive.py | 39 +++++++++++++++++++ 4 files changed, 55 insertions(+), 4 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 5cf001b1ba..251bce895d 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2787,7 +2787,7 @@ definitions: - 30 read_timeout_in_seconds: title: Read Timeout (seconds) - description: The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes. + description: The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes. Each timed-out attempt counts against the error handler's retry window (600 s for manifest connectors), so a value near 300 allows only a few retries. type: number exclusiveMinimum: 0 examples: diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 2596ad8897..2c3d864cd3 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -3213,7 +3213,7 @@ class HttpRequester(BaseModelWithDeprecations): ) read_timeout_in_seconds: Optional[float] = Field( None, - description="The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes.", + description="The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes. Each timed-out attempt counts against the error handler's retry window (600 s for manifest connectors), so a value near 300 allows only a few retries.", examples=[300], gt=0, title="Read Timeout (seconds)", diff --git a/airbyte_cdk/sources/streams/http/tcp_keepalive.py b/airbyte_cdk/sources/streams/http/tcp_keepalive.py index 7693182861..35400fcd68 100644 --- a/airbyte_cdk/sources/streams/http/tcp_keepalive.py +++ b/airbyte_cdk/sources/streams/http/tcp_keepalive.py @@ -3,6 +3,7 @@ # import functools +import logging import socket from typing import List, Tuple, cast @@ -13,6 +14,8 @@ TCP_KEEPALIVE_INTERVAL_SECONDS = 10 TCP_KEEPALIVE_PROBE_COUNT = 6 +logger = logging.getLogger("airbyte") + @functools.lru_cache(maxsize=None) def tcp_keepalive_socket_options() -> Tuple[Tuple[int, int, int], ...]: @@ -45,16 +48,18 @@ def tcp_keepalive_socket_options() -> Tuple[Tuple[int, int, int], ...]: try: probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - except OSError: + except OSError as error: + _warn_keepalive_unavailable(error) return tuple(base_options) try: supported = base_options for level, optname, value in candidates: try: probe.setsockopt(level, optname, value) - except OSError: + except OSError as error: if (level, optname) == (socket.SOL_SOCKET, socket.SO_KEEPALIVE): # Without SO_KEEPALIVE the tuning options are useless. + _warn_keepalive_unavailable(error) return tuple(base_options) continue supported.append((level, optname, value)) @@ -63,6 +68,13 @@ def tcp_keepalive_socket_options() -> Tuple[Tuple[int, int, int], ...]: probe.close() +def _warn_keepalive_unavailable(error: OSError) -> None: + # Logged once per process because tcp_keepalive_socket_options() is cached. + logger.warning( + "TCP keepalive is unavailable on this platform, so connections use plain sockets: %s", error + ) + + class TcpKeepaliveHTTPAdapter(requests.adapters.HTTPAdapter): """HTTPAdapter that enables TCP keepalive on every connection it pools.""" diff --git a/unit_tests/sources/streams/http/test_tcp_keepalive.py b/unit_tests/sources/streams/http/test_tcp_keepalive.py index 87f2b1cda1..3f79772638 100644 --- a/unit_tests/sources/streams/http/test_tcp_keepalive.py +++ b/unit_tests/sources/streams/http/test_tcp_keepalive.py @@ -2,6 +2,7 @@ # Copyright (c) 2026 Airbyte, Inc., all rights reserved. # +import logging import socket import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -126,6 +127,44 @@ def close(self): assert isinstance(real_socket_cls(socket.AF_INET, socket.SOCK_STREAM), real_socket_cls) +def test_socket_options_warn_when_keepalive_rejected(monkeypatch, caplog): + class FakeSocket: + def setsockopt(self, level, optname, value): + if (level, optname) == (socket.SOL_SOCKET, socket.SO_KEEPALIVE): + raise OSError("unsupported") + + def close(self): + pass + + monkeypatch.setattr(socket, "socket", lambda *a, **kw: FakeSocket()) + + with caplog.at_level(logging.WARNING, logger="airbyte"): + options = tcp_keepalive_socket_options() + + assert options == tuple(urllib3.connection.HTTPConnection.default_socket_options) + assert "TCP keepalive is unavailable" in caplog.text + + +def test_socket_options_warn_when_probe_socket_fails(monkeypatch, caplog): + def failing_socket(*args, **kwargs): + raise OSError("no sockets") + + monkeypatch.setattr(socket, "socket", failing_socket) + + with caplog.at_level(logging.WARNING, logger="airbyte"): + options = tcp_keepalive_socket_options() + + assert options == tuple(urllib3.connection.HTTPConnection.default_socket_options) + assert "TCP keepalive is unavailable" in caplog.text + + +def test_socket_options_do_not_warn_when_supported(caplog): + with caplog.at_level(logging.WARNING, logger="airbyte"): + tcp_keepalive_socket_options() + + assert "TCP keepalive is unavailable" not in caplog.text + + def test_adapter_passes_options_to_pool_and_proxy_managers(): adapter = TcpKeepaliveHTTPAdapter()