replace httpx with impit

Author: Mantisus

docs/02_concepts/09_streaming.mdx

@@ -18,7 +18,7 @@ Supported streaming methods:
 - [`KeyValueStoreClient.stream_record`](/reference/class/KeyValueStoreClient#stream_record) - Stream key-value store records as raw data.
 - [`LogClient.stream`](/reference/class/LogClient#stream) - Stream logs in real time.
 
-These methods return a raw, context-managed `httpx.Response` object. The response must be consumed within a with block to ensure that the connection is closed automatically, preventing memory leaks or unclosed connections.
+These methods return a raw, context-managed `impit.Response` object. The response must be consumed within a with block to ensure that the connection is closed automatically, preventing memory leaks or unclosed connections.
 
 The following example demonstrates how to stream the logs of an Actor run incrementally:
 

docs/02_concepts/code/01_async_support.py

@@ -17,8 +17,8 @@ async def main() -> None:
     # Stream the logs
     async with log_client.stream() as async_log_stream:
         if async_log_stream:
-            async for line in async_log_stream.aiter_lines():
-                print(line)
+            async for bytes_chunk in async_log_stream.aiter_bytes():
+                print(bytes_chunk)
 
 
 if __name__ == '__main__':

docs/02_concepts/code/09_streaming_async.py

@@ -10,5 +10,5 @@ async def main() -> None:
 
     async with log_client.stream() as log_stream:
         if log_stream:
-            for line in log_stream.iter_lines():
-                print(line)
+            async for bytes_chunk in log_stream.aiter_bytes():
+                print(bytes_chunk)

docs/02_concepts/code/09_streaming_sync.py

@@ -10,5 +10,5 @@ def main() -> None:
 
     with log_client.stream() as log_stream:
         if log_stream:
-            for line in log_stream.iter_lines():
-                print(line)
+            for bytes_chunk in log_stream.iter_bytes():
+                print(bytes_chunk)

pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
     "apify-shared<2.0.0",
     "colorama>=0.4.0",
     "httpx>=0.25",
+    "impit>=0.5.1",
     "more_itertools>=10.0.0",
 ]
 

src/apify_client/_errors.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-import httpx
+import json as jsonlib
+
+import impit
 from apify_shared.utils import ignore_docs
 
 
@@ -17,20 +19,21 @@ class ApifyApiError(ApifyClientError):
     """
 
     @ignore_docs
-    def __init__(self, response: httpx.Response, attempt: int) -> None:
+    def __init__(self, response: impit.Response, attempt: int, method: str = 'GET') -> None:
         """Initialize a new instance.
 
         Args:
             response: The response to the failed API call.
             attempt: Which attempt was the request that failed.
+            method: The HTTP method used for the request.
         """
         self.message: str | None = None
         self.type: str | None = None
         self.data = dict[str, str]()
 
         self.message = f'Unexpected error: {response.text}'
         try:
-            response_data = response.json()
+            response_data = jsonlib.loads(response.text)
             if 'error' in response_data:
                 self.message = response_data['error']['message']
                 self.type = response_data['error']['type']
@@ -44,7 +47,7 @@ def __init__(self, response: httpx.Response, attempt: int) -> None:
         self.name = 'ApifyApiError'
         self.status_code = response.status_code
         self.attempt = attempt
-        self.http_method = response.request.method
+        self.http_method = method
 
         # TODO: self.client_method   # noqa: TD003
         # TODO: self.original_stack  # noqa: TD003
@@ -61,7 +64,7 @@ class InvalidResponseBodyError(ApifyClientError):
     """
 
     @ignore_docs
-    def __init__(self, response: httpx.Response) -> None:
+    def __init__(self, response: impit.Response) -> None:
         """Initialize a new instance.
 
         Args:
@@ -80,8 +83,8 @@ def is_retryable_error(exc: Exception) -> bool:
         exc,
         (
             InvalidResponseBodyError,
-            httpx.NetworkError,
-            httpx.TimeoutException,
-            httpx.RemoteProtocolError,
+            impit.NetworkError,
+            impit.TimeoutException,
+            impit.RemoteProtocolError,
         ),
     )

src/apify_client/_http_client.py

@@ -8,8 +8,9 @@
 from http import HTTPStatus
 from importlib import metadata
 from typing import TYPE_CHECKING, Any
+from urllib.parse import urlencode
 
-import httpx
+import impit
 from apify_shared.utils import ignore_docs, is_content_type_json, is_content_type_text, is_content_type_xml
 
 from apify_client._errors import ApifyApiError, InvalidResponseBodyError, is_retryable_error
@@ -59,13 +60,13 @@ def __init__(
         if token is not None:
             headers['Authorization'] = f'Bearer {token}'
 
-        self.httpx_client = httpx.Client(headers=headers, follow_redirects=True, timeout=timeout_secs)
-        self.httpx_async_client = httpx.AsyncClient(headers=headers, follow_redirects=True, timeout=timeout_secs)
+        self.impit_client = impit.Client(headers=headers, follow_redirects=True, timeout=timeout_secs)
+        self.impit_async_client = impit.AsyncClient(headers=headers, follow_redirects=True, timeout=timeout_secs)
 
         self.stats = stats or Statistics()
 
     @staticmethod
-    def _maybe_parse_response(response: httpx.Response) -> Any:
+    def _maybe_parse_response(response: impit.Response) -> Any:
         if response.status_code == HTTPStatus.NO_CONTENT:
             return None
 
@@ -75,7 +76,7 @@ def _maybe_parse_response(response: httpx.Response) -> Any:
 
         try:
             if is_content_type_json(content_type):
-                return response.json()
+                return jsonlib.loads(response.text)
             elif is_content_type_xml(content_type) or is_content_type_text(content_type):  # noqa: RET505
                 return response.text
             else:
@@ -131,6 +132,21 @@ def _prepare_request_call(
             data,
         )
 
+    def _build_url_with_params(self, url: str, params: dict | None = None) -> str:
+        if not params:
+            return url
+
+        param_pairs: list[tuple[str, str]] = []
+        for key, value in params.items():
+            if isinstance(value, list):
+                param_pairs.extend((key, str(v)) for v in value)
+            else:
+                param_pairs.append((key, str(value)))
+
+        query_string = urlencode(param_pairs)
+
+        return f'{url}?{query_string}'
+
 
 class HTTPClient(_BaseHTTPClient):
     def call(
@@ -145,7 +161,7 @@ def call(
         stream: bool | None = None,
         parse_response: bool | None = True,
         timeout_secs: int | None = None,
-    ) -> httpx.Response:
+    ) -> impit.Response:
         log_context.method.set(method)
         log_context.url.set(url)
 
@@ -156,41 +172,34 @@ def call(
 
         headers, params, content = self._prepare_request_call(headers, params, data, json)
 
-        httpx_client = self.httpx_client
+        impit_client = self.impit_client
 
-        def _make_request(stop_retrying: Callable, attempt: int) -> httpx.Response:
+        def _make_request(stop_retrying: Callable, attempt: int) -> impit.Response:
             log_context.attempt.set(attempt)
             logger.debug('Sending request')
 
             self.stats.requests += 1
 
             try:
-                request = httpx_client.build_request(
+                # Increase timeout with each attempt. Max timeout is bounded by the client timeout.
+                timeout = min(self.timeout_secs, (timeout_secs or self.timeout_secs) * 2 ** (attempt - 1))
+
+                url_with_params = self._build_url_with_params(url, params)
+
+                response = impit_client.request(
                     method=method,
-                    url=url,
+                    url=url_with_params,
                     headers=headers,
-                    params=params,
                     content=content,
-                )
-
-                # Increase timeout with each attempt. Max timeout is bounded by the client timeout.
-                timeout = min(self.timeout_secs, (timeout_secs or self.timeout_secs) * 2 ** (attempt - 1))
-                request.extensions['timeout'] = {
-                    'connect': timeout,
-                    'pool': timeout,
-                    'read': timeout,
-                    'write': timeout,
-                }
-
-                response = httpx_client.send(
-                    request=request,
+                    timeout=timeout,
                     stream=stream or False,
                 )
 
                 # If response status is < 300, the request was successful, and we can return the result
                 if response.status_code < 300:  # noqa: PLR2004
                     logger.debug('Request successful', extra={'status_code': response.status_code})
-                    if not stream:
+                    # TODODO Impit does not support setting custom attributes on the response object,
+                    if not stream and response.content == b'A unique condition for checking types. ABRACADABRA':
                         _maybe_parsed_body = (
                             self._maybe_parse_response(response) if parse_response else response.content
                         )
@@ -214,7 +223,7 @@ def _make_request(stop_retrying: Callable, attempt: int) -> httpx.Response:
             if response.status_code < 500 and response.status_code != HTTPStatus.TOO_MANY_REQUESTS:  # noqa: PLR2004
                 logger.debug('Status code is not retryable', extra={'status_code': response.status_code})
                 stop_retrying()
-            raise ApifyApiError(response, attempt)
+            raise ApifyApiError(response, attempt, method=method)
 
         return retry_with_exp_backoff(
             _make_request,
@@ -238,7 +247,7 @@ async def call(
         stream: bool | None = None,
         parse_response: bool | None = True,
         timeout_secs: int | None = None,
-    ) -> httpx.Response:
+    ) -> impit.Response:
         log_context.method.set(method)
         log_context.url.set(url)
 
@@ -249,38 +258,31 @@ async def call(
 
         headers, params, content = self._prepare_request_call(headers, params, data, json)
 
-        httpx_async_client = self.httpx_async_client
+        impit_async_client = self.impit_async_client
 
-        async def _make_request(stop_retrying: Callable, attempt: int) -> httpx.Response:
+        async def _make_request(stop_retrying: Callable, attempt: int) -> impit.Response:
             log_context.attempt.set(attempt)
             logger.debug('Sending request')
             try:
-                request = httpx_async_client.build_request(
+                # Increase timeout with each attempt. Max timeout is bounded by the client timeout.
+                timeout = min(self.timeout_secs, (timeout_secs or self.timeout_secs) * 2 ** (attempt - 1))
+
+                url_with_params = self._build_url_with_params(url, params)
+
+                response = await impit_async_client.request(
                     method=method,
-                    url=url,
+                    url=url_with_params,
                     headers=headers,
-                    params=params,
                     content=content,
-                )
-
-                # Increase timeout with each attempt. Max timeout is bounded by the client timeout.
-                timeout = min(self.timeout_secs, (timeout_secs or self.timeout_secs) * 2 ** (attempt - 1))
-                request.extensions['timeout'] = {
-                    'connect': timeout,
-                    'pool': timeout,
-                    'read': timeout,
-                    'write': timeout,
-                }
-
-                response = await httpx_async_client.send(
-                    request=request,
+                    timeout=timeout,
                     stream=stream or False,
                 )
 
                 # If response status is < 300, the request was successful, and we can return the result
                 if response.status_code < 300:  # noqa: PLR2004
                     logger.debug('Request successful', extra={'status_code': response.status_code})
-                    if not stream:
+                    # TODODO Impit does not support setting custom attributes on the response object,
+                    if not stream and response.content == b'A unique condition for checking types. ABRACADABRA':
                         _maybe_parsed_body = (
                             self._maybe_parse_response(response) if parse_response else response.content
                         )
@@ -304,7 +306,7 @@ async def _make_request(stop_retrying: Callable, attempt: int) -> httpx.Response
             if response.status_code < 500 and response.status_code != HTTPStatus.TOO_MANY_REQUESTS:  # noqa: PLR2004
                 logger.debug('Status code is not retryable', extra={'status_code': response.status_code})
                 stop_retrying()
-            raise ApifyApiError(response, attempt)
+            raise ApifyApiError(response, attempt, method=method)
 
         return await retry_with_exp_backoff_async(
             _make_request,

src/apify_client/clients/base/actor_job_base_client.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import json as jsonlib
 import math
 import time
 from datetime import datetime, timezone
@@ -39,7 +40,7 @@ def _wait_for_finish(self, wait_secs: int | None = None) -> dict | None:
                     method='GET',
                     params=self._params(waitForFinish=wait_for_finish),
                 )
-                job = parse_date_fields(pluck_data(response.json()))
+                job = parse_date_fields(pluck_data(jsonlib.loads(response.text)))
 
                 seconds_elapsed = math.floor((datetime.now(timezone.utc) - started_at).total_seconds())
                 if ActorJobStatus(job['status']).is_terminal or (
@@ -70,7 +71,7 @@ def _abort(self, *, gracefully: bool | None = None) -> dict:
             method='POST',
             params=self._params(gracefully=gracefully),
         )
-        return parse_date_fields(pluck_data(response.json()))
+        return parse_date_fields(pluck_data(jsonlib.loads(response.text)))
 
 
 @ignore_docs
@@ -94,7 +95,7 @@ async def _wait_for_finish(self, wait_secs: int | None = None) -> dict | None:
                     method='GET',
                     params=self._params(waitForFinish=wait_for_finish),
                 )
-                job = parse_date_fields(pluck_data(response.json()))
+                job = parse_date_fields(pluck_data(jsonlib.loads(response.text)))
 
                 seconds_elapsed = math.floor((datetime.now(timezone.utc) - started_at).total_seconds())
                 if ActorJobStatus(job['status']).is_terminal or (
@@ -125,4 +126,4 @@ async def _abort(self, *, gracefully: bool | None = None) -> dict:
             method='POST',
             params=self._params(gracefully=gracefully),
         )
-        return parse_date_fields(pluck_data(response.json()))
+        return parse_date_fields(pluck_data(jsonlib.loads(response.text)))

src/apify_client/clients/base/resource_client.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import json as jsonlib
+
 from apify_shared.utils import ignore_docs, parse_date_fields
 
 from apify_client._errors import ApifyApiError
@@ -20,7 +22,7 @@ def _get(self, timeout_secs: int | None = None) -> dict | None:
                 timeout_secs=timeout_secs,
             )
 
-            return parse_date_fields(pluck_data(response.json()))
+            return parse_date_fields(pluck_data(jsonlib.loads(response.text)))
 
         except ApifyApiError as exc:
             catch_not_found_or_throw(exc)
@@ -36,7 +38,7 @@ def _update(self, updated_fields: dict, timeout_secs: int | None = None) -> dict
             timeout_secs=timeout_secs,
         )
 
-        return parse_date_fields(pluck_data(response.json()))
+        return parse_date_fields(pluck_data(jsonlib.loads(response.text)))
 
     def _delete(self, timeout_secs: int | None = None) -> None:
         try:
@@ -64,7 +66,7 @@ async def _get(self, timeout_secs: int | None = None) -> dict | None:
                 timeout_secs=timeout_secs,
             )
 
-            return parse_date_fields(pluck_data(response.json()))
+            return parse_date_fields(pluck_data(jsonlib.loads(response.text)))
 
         except ApifyApiError as exc:
             catch_not_found_or_throw(exc)
@@ -80,7 +82,7 @@ async def _update(self, updated_fields: dict, timeout_secs: int | None = None) -
             timeout_secs=timeout_secs,
         )
 
-        return parse_date_fields(pluck_data(response.json()))
+        return parse_date_fields(pluck_data(jsonlib.loads(response.text)))
 
     async def _delete(self, timeout_secs: int | None = None) -> None:
         try: