feat: Unify the request pipeline across HTTP clients

Author: vdusek

.rules.md

@@ -60,7 +60,8 @@ Docstrings are written on sync clients and **automatically copied** to async cli
 
 ### HTTP Client Abstraction
 
-- `HttpClient`/`HttpClientAsync` — abstract base classes in `_http_clients/_base.py`
+- `HttpClient`/`HttpClientAsync` — base classes in `http_clients/_base.py` holding the shared request pipeline (retries,
+  timeouts, API errors); transports implement the `send_request`, error-classification, and lifecycle hooks
 - `ImpitHttpClient`/`ImpitHttpClientAsync` — default implementation (Rust-based Impit)
 - `HttpResponse` — Protocol (not a concrete class) for response objects
 - Users can plug in custom HTTP clients via `ApifyClient.with_custom_http_client()`

src/apify_client/_apify_client.py

@@ -226,19 +226,21 @@ def with_custom_http_client(
         """Create an `ApifyClient` instance with a custom HTTP client.
 
         Use this alternative constructor when you want to provide your own HTTP client implementation
-        instead of the default one. The custom client is responsible for its own configuration
-        (retries, timeouts, etc.); only the token is applied to it, as described below.
+        instead of the default one. The custom client controls its transport configuration; the shared
+        `HttpClient` pipeline handles request preparation, retries, timeouts, and API errors unless overridden.
 
         ### Usage
 
         ```python
         from apify_client import ApifyClient
         from apify_client.http_clients import HttpClient, HttpResponse
 
+
         class MyHttpClient(HttpClient):
-            def call(self, *, method, url, **kwargs) -> HttpResponse:
+            def send_request(self, *, method, url, headers, content, timeout, stream) -> HttpResponse:
                 ...
 
+
         client = ApifyClient.with_custom_http_client(
             token='MY-APIFY-TOKEN',
             http_client=MyHttpClient(),
@@ -588,22 +590,24 @@ def with_custom_http_client(
         """Create an `ApifyClientAsync` instance with a custom HTTP client.
 
         Use this alternative constructor when you want to provide your own HTTP client implementation
-        instead of the default one. The custom client is responsible for its own configuration
-        (retries, timeouts, etc.); only the token is applied to it, as described below.
+        instead of the default one. The custom client controls its transport configuration; the shared
+        `HttpClientAsync` pipeline handles request preparation, retries, timeouts, and API errors unless overridden.
 
         ### Usage
 
         ```python
         from apify_client import ApifyClientAsync
         from apify_client.http_clients import HttpClientAsync, HttpResponse
 
-        class MyHttpClient(HttpClientAsync):
-            async def call(self, *, method, url, **kwargs) -> HttpResponse:
+
+        class MyHttpClientAsync(HttpClientAsync):
+            async def send_request(self, *, method, url, headers, content, timeout, stream) -> HttpResponse:
                 ...
 
+
         client = ApifyClientAsync.with_custom_http_client(
             token='MY-APIFY-TOKEN',
-            http_client=MyHttpClient(),
+            http_client=MyHttpClientAsync(),
         )
         ```
 

src/apify_client/_streamed_log.py

@@ -9,8 +9,6 @@
 from threading import Thread
 from typing import TYPE_CHECKING, ClassVar, Self, cast
 
-import impit
-
 from apify_client._docs import docs_group
 
 if TYPE_CHECKING:
@@ -29,9 +27,8 @@ class StreamedLogBase:
     _stream_timeout: ClassVar[Timeout] = 'no_timeout'
     """Timeout for the log-stream long-poll request, which stays open for the whole Actor run.
 
-    impit applies its `timeout` to the whole request including the streamed body, so any bounded value truncates a
-    longer run mid-stream and raises `impit.TimeoutException` (#1040). `no_timeout` maps to impit's ~24h cap, which
-    is effectively unbounded for real runs and mirrors the JS client.
+    A bounded transport timeout can truncate a longer run mid-stream. `no_timeout` keeps the connection open for the
+    duration of the run (Impit currently maps it to an effective 24-hour cap) and mirrors the JS client.
     """
 
     def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> None:
@@ -162,13 +159,13 @@ def _stream_log(self) -> None:
                 finally:
                     # Flush the last buffered part even if the read timed out or was stopped.
                     self._log_buffer_content(include_last_part=True)
-        except impit.TimeoutException:
-            # With `no_timeout` this fires only if the run outlives impit's ~24h cap or the connection stalls.
-            # The stream cannot continue, so warn and let the thread end instead of leaking a traceback (#1040).
-            self._to_logger.warning('Log streaming stopped: the log stream request timed out.')
-        except Exception:
-            # Any other failure in log redirection must not escape the background thread; log it instead.
-            self._to_logger.exception('Log redirection stopped due to unexpected error:')
+        except Exception as exc:
+            if self._log_client._http_client.is_timeout_error(exc):  # noqa: SLF001
+                # The stream cannot continue, so warn and let the thread end instead of leaking a traceback.
+                self._to_logger.warning('Log streaming stopped: the log stream request timed out.')
+            else:
+                # Any other failure in log redirection must not escape the background thread; log it instead.
+                self._to_logger.exception('Log redirection stopped due to unexpected error:')
 
 
 @docs_group('Other')
@@ -241,10 +238,10 @@ async def _stream_log(self) -> None:
                 finally:
                     # Flush the last buffered part even if the task is cancelled by `stop()`.
                     self._log_buffer_content(include_last_part=True)
-        except impit.TimeoutException:
-            # As in `StreamedLog._stream_log`, impit's whole-request timeout on the long-lived stream is an
-            # expected terminal condition, not an error, so log a warning and end the task instead of a traceback.
-            self._to_logger.warning('Log streaming stopped: the log stream request timed out.')
-        except Exception:
-            # Exception in log redirection should not propagate further.
-            self._to_logger.exception('Log redirection stopped due to unexpected error:')
+        except Exception as exc:
+            if self._log_client._http_client.is_timeout_error(exc):  # noqa: SLF001
+                # A timeout on the long-lived stream is an expected terminal condition, not an error.
+                self._to_logger.warning('Log streaming stopped: the log stream request timed out.')
+            else:
+                # Exception in log redirection should not propagate further.
+                self._to_logger.exception('Log redirection stopped due to unexpected error:')

src/apify_client/_utils/errors.py

@@ -2,9 +2,7 @@
 
 from typing import TYPE_CHECKING
 
-import impit
-
-from apify_client.errors import InvalidResponseBodyError, NotFoundError
+from apify_client.errors import NotFoundError
 
 if TYPE_CHECKING:
     from apify_client.errors import ApifyApiError
@@ -33,19 +31,3 @@ def catch_not_found_for_resource_or_throw(exc: ApifyApiError, resource_id: str |
     if resource_id is None:
         raise exc
     catch_not_found_or_throw(exc)
-
-
-def is_retryable_error(exc: Exception) -> bool:
-    """Check if the given error is retryable.
-
-    All `impit.HTTPError` subclasses are considered retryable because they represent transport-level failures
-    (network issues, timeouts, protocol errors, body decoding errors) that are typically transient. HTTP status
-    code errors are handled separately in `_make_request` based on the response status code, not here.
-    """
-    return isinstance(
-        exc,
-        (
-            InvalidResponseBodyError,
-            impit.HTTPError,
-        ),
-    )

src/apify_client/http_clients/__init__.py

@@ -1,5 +1,5 @@
-from ._base import HttpClient, HttpClientAsync, HttpResponse
-from ._impit import ImpitHttpClient, ImpitHttpClientAsync
+from apify_client.http_clients._base import HttpClient, HttpClientAsync, HttpResponse
+from apify_client.http_clients._impit import ImpitHttpClient, ImpitHttpClientAsync
 
 __all__ = [
     'HttpClient',