diff --git a/.rules.md b/.rules.md index 27460217..cced5614 100644 --- a/.rules.md +++ b/.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()` diff --git a/README.md b/README.md index ba5b9084..7627e5fe 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,15 @@ uv add "apify-client[brotli]" ``` + [Impit](https://github.com/apify/impit) is the default HTTP client and is installed automatically. To use the + built-in [HTTPX](https://www.python-httpx.org/) client instead, install its optional extra: + + ```bash + pip install "apify-client[httpx]" + # or + uv add "apify-client[httpx]" + ``` + - From [conda-forge](https://anaconda.org/conda-forge/apify-client), it can be installed with [conda](https://docs.conda.io/en/latest/): ```bash @@ -124,7 +133,7 @@ For a guided walkthrough — authenticating, running an Actor, and reading its r - **Tiered timeouts** — short / medium / long tiers picked per endpoint, overridable per call ([Timeouts](https://docs.apify.com/api/client/python/docs/concepts/timeouts)). - **Pagination and streaming** — iterate datasets, key-value store keys, or live logs without manual paging or buffering ([Pagination](https://docs.apify.com/api/client/python/docs/concepts/pagination), [Streaming](https://docs.apify.com/api/client/python/docs/concepts/streaming-resources)). - **Convenience methods** — `call()`, `wait_for_finish()`, nested resource access, and other shortcuts that hide platform quirks ([Convenience methods](https://docs.apify.com/api/client/python/docs/concepts/convenience-methods)). -- **Pluggable HTTP layer** — swap the default [Impit](https://github.com/apify/impit)-based HTTP client for `httpx`, `requests`, `aiohttp`, or any custom implementation ([Custom HTTP clients](https://docs.apify.com/api/client/python/docs/concepts/custom-http-clients)). +- **Pluggable HTTP layer** — use the default [Impit](https://github.com/apify/impit)-based client, opt in to the built-in [HTTPX](https://www.python-httpx.org/) client, or plug in any custom implementation ([Custom HTTP clients](https://docs.apify.com/api/client/python/docs/concepts/custom-http-clients)). - **Structured errors** — every API error surfaces as an [`ApifyApiError`](https://docs.apify.com/api/client/python/reference/class/ApifyApiError) with HTTP-specific subclasses for precise handling ([Error handling](https://docs.apify.com/api/client/python/docs/concepts/error-handling)). - **Debug logging** — opt-in structured logging on the `apify_client` logger captures request URLs, status codes, retry attempts, and more ([Logging](https://docs.apify.com/api/client/python/docs/concepts/logging)). diff --git a/pyproject.toml b/pyproject.toml index 0019741f..dc23c5df 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,7 @@ dependencies = [ [project.optional-dependencies] brotli = ["brotli>=1.0.9"] +httpx = ["httpx>=0.27.0,<1.0.0"] [project.urls] "Apify Homepage" = "https://apify.com" diff --git a/src/apify_client/_apify_client.py b/src/apify_client/_apify_client.py index 6ca9413d..73beadef 100644 --- a/src/apify_client/_apify_client.py +++ b/src/apify_client/_apify_client.py @@ -226,8 +226,8 @@ 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 @@ -235,10 +235,12 @@ def with_custom_http_client( 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,8 +590,8 @@ 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 @@ -597,13 +599,15 @@ def with_custom_http_client( 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(), ) ``` diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index bdc89d8a..428caa12 100644 --- a/src/apify_client/_streamed_log.py +++ b/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:') diff --git a/src/apify_client/http_clients/__init__.py b/src/apify_client/http_clients/__init__.py index 1c821677..d1e06c90 100644 --- a/src/apify_client/http_clients/__init__.py +++ b/src/apify_client/http_clients/__init__.py @@ -1,10 +1,35 @@ -from ._base import HttpClient, HttpClientAsync, HttpResponse -from ._impit import ImpitHttpClient, ImpitHttpClientAsync +from apify_client._utils.try_import import install_import_hook as _install_import_hook +from apify_client._utils.try_import import try_import as _try_import +from apify_client.http_clients._base import HttpClient, HttpClientAsync, HttpResponse +from apify_client.http_clients._impit import ImpitHttpClient, ImpitHttpClientAsync -__all__ = [ - 'HttpClient', - 'HttpClientAsync', - 'HttpResponse', - 'ImpitHttpClient', - 'ImpitHttpClientAsync', -] +_install_import_hook(__name__) + +# `httpx` is an optional extra, so it's wrapped in try_import. Accessing the HTTPX clients +# without the extra installed raises a clear ImportError instead of failing at package import time. +with _try_import( + __name__, + 'HttpxHttpClient', + 'HttpxHttpClientAsync', + dependency_name='httpx', +) as _httpx_import: + from apify_client.http_clients._httpx import HttpxHttpClient, HttpxHttpClientAsync + +if _httpx_import.available: + __all__ = [ + 'HttpClient', + 'HttpClientAsync', + 'HttpResponse', + 'HttpxHttpClient', + 'HttpxHttpClientAsync', + 'ImpitHttpClient', + 'ImpitHttpClientAsync', + ] +else: + __all__ = [ + 'HttpClient', + 'HttpClientAsync', + 'HttpResponse', + 'ImpitHttpClient', + 'ImpitHttpClientAsync', + ] diff --git a/src/apify_client/http_clients/_base.py b/src/apify_client/http_clients/_base.py index 5a4491f8..625fe4ec 100644 --- a/src/apify_client/http_clients/_base.py +++ b/src/apify_client/http_clients/_base.py @@ -1,13 +1,17 @@ from __future__ import annotations +import asyncio import json as jsonlib import logging import os +import random import sys -from abc import ABC, abstractmethod +import time +from contextlib import suppress from datetime import UTC, datetime, timedelta +from http import HTTPStatus from importlib import metadata -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, TypeVar from urllib.parse import urlencode # `Protocol` comes from `typing_extensions`, not `typing`, because its runtime `isinstance` check looks attributes @@ -25,18 +29,23 @@ MIN_COMPRESSION_SIZE, ) from apify_client._docs import docs_group -from apify_client._logging import LoggerOnce, logger_name +from apify_client._logging import LoggerOnce, log_context, logger_name from apify_client._statistics import ClientStatistics from apify_client._utils.http import is_compressible_content_type from apify_client._utils.time import to_seconds +from apify_client.errors import ApifyApiError from apify_client.http_compressors._gzip import GzipHttpCompressor if TYPE_CHECKING: - from collections.abc import AsyncIterator, Iterator, Mapping + from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping + from types import TracebackType + from typing import Self from apify_client.http_compressors._base import HttpCompressor from apify_client.types import JsonSerializable, Timeout +T = TypeVar('T') + logger = logging.getLogger(logger_name) logger_once = LoggerOnce(logger) @@ -93,12 +102,11 @@ def aiter_bytes(self) -> AsyncIterator[bytes]: class HttpClientBase: """Shared configuration and utilities for HTTP clients. - Provides common functionality for both sync and async HTTP clients including: - header construction, parameter parsing, request body preparation, URL building, - and timeout calculation. + Provides common functionality for both sync and async HTTP clients including: header construction, parameter + parsing, request body preparation, URL building, timeout calculation, and error classification. - Subclasses should call `super().__init__()` to initialize shared configuration. - The helper methods are then available for use in the `call()` implementation. + Subclasses should call `super().__init__()` to initialize shared configuration. The helpers are then used by the + inherited `call`, and stay available to a client that replaces it. """ def __init__( @@ -166,6 +174,23 @@ def set_default_authorization(self, token: str) -> None: if self._get_header(self._headers, 'authorization') is None: self._headers['Authorization'] = f'Bearer {token}' + def is_timeout_error(self, exc: Exception) -> bool: + """Return whether an exception represents a transport timeout. + + Recognizes Python's own `TimeoutError`. Transport adapters extend it with the timeout types their HTTP + library defines. + """ + return isinstance(exc, TimeoutError) + + def is_retryable_transport_error(self, exc: Exception) -> bool: + """Return whether an underlying HTTP-library exception is retryable. + + The default classifies nothing as retryable, so a transport that doesn't override it gives up on the first + connection failure. Every transport adapter should map its own transient error types here. + """ + _ = exc + return False + @staticmethod def _merge_headers(base: dict[str, str] | None, override: dict[str, str] | None) -> dict[str, str]: """Merge two header dicts, treating header names case-insensitively. @@ -333,21 +358,70 @@ def _build_url_with_params(self, url: str, *, params: dict[str, Any] | None = No return f'{url}?{query_string}' + def _handle_request_exception(self, exc: Exception, *, stop_retrying: Callable[[], None]) -> None: + """Stop retrying when an exception is not a retryable transport failure.""" + logger.debug('Request threw exception', exc_info=exc) + if not self.is_retryable_transport_error(exc): + logger.debug('Exception is not retryable', exc_info=exc) + stop_retrying() -@docs_group('HTTP clients') -class HttpClient(HttpClientBase, ABC): - """Abstract base class for synchronous HTTP clients used by `ApifyClient`. + def _handle_response_status( + self, + response: HttpResponse, + *, + attempt: int, + stop_retrying: Callable[[], None], + ) -> bool: + """Record the response status and stop retrying unless it is a server error or a rate limit. - Extend this class to create a custom synchronous HTTP client. Override the `call` method - with your implementation. Helper methods from the base class are available for request - preparation, URL building, and parameter parsing. + Returns whether the response is a success, so the caller can hand it back instead of raising. + """ + if response.status_code < HTTPStatus.MULTIPLE_CHOICES: + logger.debug('Request successful', extra={'status_code': response.status_code}) + return True - Implementations must send the client's default headers from `self._headers` with every request, - otherwise the `Authorization` header never reaches the API. The `_prepare_request_call` helper - merges them into the per-request headers automatically. + if response.status_code == HTTPStatus.TOO_MANY_REQUESTS: + self._statistics.add_rate_limit_error(attempt) + + logger.debug('Request unsuccessful', extra={'status_code': response.status_code}) + if ( + response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR + and response.status_code != HTTPStatus.TOO_MANY_REQUESTS + ): + logger.debug('Status code is not retryable', extra={'status_code': response.status_code}) + stop_retrying() + + return False + + +@docs_group('HTTP clients') +class HttpClient(HttpClientBase): + """Base class for synchronous HTTP clients used by `ApifyClient`. + + Concrete clients inherit its request preparation, retry, and error handling, and implement the transport, + error-classification, and lifecycle hooks. Only `send_request` has to be implemented, the other hooks have + defaults. Replacing `call` itself also remains supported, and then the transport and retry-classification hooks + are bypassed. Helper methods from `HttpClientBase` remain available for request preparation, URL building, and + parameter parsing. + + The client's default headers from `self._headers` have to go out with every request, otherwise the + `Authorization` header never reaches the API. The inherited `call` merges them into the per-request headers + through `_prepare_request_call`, so only a client that replaces `call` has to do it itself. """ - @abstractmethod + def __enter__(self) -> Self: + """Return this client and close it when the context exits.""" + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Close resources owned by the HTTP client.""" + self.close() + def call( self, *, @@ -356,16 +430,16 @@ def call( headers: dict[str, str] | None = None, params: dict[str, Any] | None = None, data: str | bytes | bytearray | None = None, - json: Any = None, + json: JsonSerializable | None = None, stream: bool | None = None, timeout: Timeout = 'medium', ) -> HttpResponse: - """Make an HTTP request. + """Make an HTTP request with automatic retry and exponential backoff. Args: method: HTTP method (GET, POST, PUT, DELETE, etc.). url: Full URL to make the request to. - headers: Additional headers to include in this request. + headers: Additional headers to include. params: Query parameters to append to the URL. data: Raw request body data. Cannot be used together with json. json: JSON-serializable data for the request body. Cannot be used together with data. @@ -381,17 +455,176 @@ def call( ApifyApiError: If the request fails after all retries or returns a non-retryable error status. ValueError: If both json and data are provided. """ + log_context.method.set(method) + log_context.url.set(url) + + self._statistics.calls += 1 + + prepared_headers, prepared_params, content = self._prepare_request_call( + headers=headers, + params=params, + data=data, + json=json, + ) + + return self._retry_with_exp_backoff( + lambda stop_retrying, attempt: self._make_request( + stop_retrying=stop_retrying, + attempt=attempt, + method=method, + url=url, + headers=prepared_headers, + params=prepared_params, + content=content, + stream=stream, + timeout=timeout, + ), + max_retries=self._max_retries, + backoff_base=self._min_delay_between_retries, + ) + + def close(self) -> None: + """Close resources owned by the HTTP client. + + Transports that own a connection pool or a session override it. The default does nothing. + """ + + def send_request( + self, + *, + method: str, + url: str, + headers: dict[str, str], + content: bytes | None, + timeout: float | None, + stream: bool, + ) -> HttpResponse: + """Send one request through the underlying HTTP library. + + Required by the inherited `call`, so a transport adapter must implement it. Overriding `call` itself + bypasses it entirely. Let the HTTP library's exceptions propagate unwrapped: `call` classifies them + through `is_retryable_transport_error` and `is_timeout_error`. + + Args: + method: HTTP method (GET, POST, PUT, DELETE, etc.). + url: Full request URL, with the query parameters already encoded into it. + headers: Final request headers, with the client's default headers already merged in. + content: Request body, already serialized and compressed, or None for a request without a body. + timeout: Timeout for this attempt in seconds, or None for no timeout at all. + stream: Whether to return the response with the body unread, so the caller can stream it. + + Returns: + The HTTP response object. + """ + raise NotImplementedError('Implement `send_request` to provide a transport, or override `call` entirely.') + + @staticmethod + def _retry_with_exp_backoff( + func: Callable[[Callable[[], None], int], T], + *, + max_retries: int = 8, + backoff_base: timedelta = timedelta(milliseconds=500), + backoff_factor: float = 2, + random_factor: float = 1, + ) -> T: + """Retry a function with exponential backoff and jitter.""" + if max_retries < 1: + raise ValueError(f'max_retries must be at least 1, got {max_retries}') + + random_factor = min(max(0, random_factor), 1) + backoff_factor = min(max(1, backoff_factor), 10) + swallow = True + + def stop_retrying() -> None: + nonlocal swallow + swallow = False + + for attempt in range(1, max_retries + 1): + try: + return func(stop_retrying, attempt) + except Exception: + if not swallow: + raise + + random_sleep_factor = random.uniform(1, 1 + random_factor) + backoff_base_secs = to_seconds(backoff_base) + backoff_exp_factor = backoff_factor ** (attempt - 1) + time.sleep(random_sleep_factor * backoff_base_secs * backoff_exp_factor) + + return func(stop_retrying, max_retries + 1) + + def _make_request( + self, + *, + stop_retrying: Callable[[], None], + attempt: int, + method: str, + url: str, + headers: dict[str, str], + params: dict[str, Any] | None, + content: bytes | None, + stream: bool | None, + timeout: Timeout, + ) -> HttpResponse: + """Execute one request attempt through the transport adapter.""" + log_context.attempt.set(attempt) + logger.debug('Sending request') + + self._statistics.requests += 1 + + try: + response = self.send_request( + method=method, + url=self._build_url_with_params(url, params=params), + headers=headers, + content=content, + timeout=self._compute_timeout(timeout, attempt=attempt), + stream=stream or False, + ) + except Exception as exc: + self._handle_request_exception(exc, stop_retrying=stop_retrying) + raise + + if self._handle_response_status(response, attempt=attempt, stop_retrying=stop_retrying): + return response + + # Read the response in case it is a stream, so the error can be raised properly. A failed read goes through + # the same classification as a failed send. + try: + response.read() + except Exception as exc: + logger.debug('Reading the error response failed', exc_info=exc) + with suppress(Exception): + response.close() + if not self.is_retryable_transport_error(exc): + logger.debug('Exception is not retryable', exc_info=exc) + stop_retrying() + raise + + raise ApifyApiError(response, attempt, method=method) @docs_group('HTTP clients') -class HttpClientAsync(HttpClientBase, ABC): - """Abstract base class for asynchronous HTTP clients used by `ApifyClientAsync`. +class HttpClientAsync(HttpClientBase): + """Base class for asynchronous HTTP clients used by `ApifyClientAsync`. Extend this class to create a custom asynchronous HTTP client. See `HttpClient` for details on the expected behavior. """ - @abstractmethod + async def __aenter__(self) -> Self: + """Return this client and close it when the async context exits.""" + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Close resources owned by the asynchronous HTTP client.""" + await self.aclose() + async def call( self, *, @@ -400,16 +633,16 @@ async def call( headers: dict[str, str] | None = None, params: dict[str, Any] | None = None, data: str | bytes | bytearray | None = None, - json: Any = None, + json: JsonSerializable | None = None, stream: bool | None = None, timeout: Timeout = 'medium', ) -> HttpResponse: - """Make an HTTP request. + """Make an HTTP request with automatic retry and exponential backoff. Args: method: HTTP method (GET, POST, PUT, DELETE, etc.). url: Full URL to make the request to. - headers: Additional headers to include in this request. + headers: Additional headers to include. params: Query parameters to append to the URL. data: Raw request body data. Cannot be used together with json. json: JSON-serializable data for the request body. Cannot be used together with data. @@ -425,3 +658,163 @@ async def call( ApifyApiError: If the request fails after all retries or returns a non-retryable error status. ValueError: If both json and data are provided. """ + log_context.method.set(method) + log_context.url.set(url) + + self._statistics.calls += 1 + + # Serializing and compressing a request body is CPU-bound and would block the event loop, so + # offload preparation to a worker thread whenever there is something to compress. A body the + # client sends as it is costs less to prepare inline than the hop itself. A `json` body always + # hops, as its size is only known once serialized. + if json is not None or self._is_body_worth_compressing(data): + prepared_headers, prepared_params, content = await asyncio.to_thread( + self._prepare_request_call, + headers=headers, + params=params, + data=data, + json=json, + ) + else: + prepared_headers, prepared_params, content = self._prepare_request_call( + headers=headers, + params=params, + data=data, + json=json, + ) + + return await self._retry_with_exp_backoff( + lambda stop_retrying, attempt: self._make_request( + stop_retrying=stop_retrying, + attempt=attempt, + method=method, + url=url, + headers=prepared_headers, + params=prepared_params, + content=content, + stream=stream, + timeout=timeout, + ), + max_retries=self._max_retries, + backoff_base=self._min_delay_between_retries, + ) + + async def aclose(self) -> None: + """Close resources owned by the asynchronous HTTP client. + + Transports that own a connection pool or a session override it. The default does nothing. + """ + + async def send_request( + self, + *, + method: str, + url: str, + headers: dict[str, str], + content: bytes | None, + timeout: float | None, + stream: bool, + ) -> HttpResponse: + """Send one request through the underlying HTTP library. + + Required by the inherited `call`, so a transport adapter must implement it. Overriding `call` itself + bypasses it entirely. Let the HTTP library's exceptions propagate unwrapped: `call` classifies them + through `is_retryable_transport_error` and `is_timeout_error`. + + Args: + method: HTTP method (GET, POST, PUT, DELETE, etc.). + url: Full request URL, with the query parameters already encoded into it. + headers: Final request headers, with the client's default headers already merged in. + content: Request body, already serialized and compressed, or None for a request without a body. + timeout: Timeout for this attempt in seconds, or None for no timeout at all. + stream: Whether to return the response with the body unread, so the caller can stream it. + + Returns: + The HTTP response object. + """ + raise NotImplementedError('Implement `send_request` to provide a transport, or override `call` entirely.') + + @staticmethod + async def _retry_with_exp_backoff( + func: Callable[[Callable[[], None], int], Awaitable[T]], + *, + max_retries: int = 8, + backoff_base: timedelta = timedelta(milliseconds=500), + backoff_factor: float = 2, + random_factor: float = 1, + ) -> T: + """Retry an async function with exponential backoff and jitter.""" + if max_retries < 1: + raise ValueError(f'max_retries must be at least 1, got {max_retries}') + + random_factor = min(max(0, random_factor), 1) + backoff_factor = min(max(1, backoff_factor), 10) + swallow = True + + def stop_retrying() -> None: + nonlocal swallow + swallow = False + + for attempt in range(1, max_retries + 1): + try: + return await func(stop_retrying, attempt) + except Exception: + if not swallow: + raise + + random_sleep_factor = random.uniform(1, 1 + random_factor) + backoff_base_secs = to_seconds(backoff_base) + backoff_exp_factor = backoff_factor ** (attempt - 1) + await asyncio.sleep(random_sleep_factor * backoff_base_secs * backoff_exp_factor) + + return await func(stop_retrying, max_retries + 1) + + async def _make_request( + self, + *, + stop_retrying: Callable[[], None], + attempt: int, + method: str, + url: str, + headers: dict[str, str], + params: dict[str, Any] | None, + content: bytes | None, + stream: bool | None, + timeout: Timeout, + ) -> HttpResponse: + """Execute one request attempt through the transport adapter.""" + log_context.attempt.set(attempt) + logger.debug('Sending request') + + self._statistics.requests += 1 + + try: + response = await self.send_request( + method=method, + url=self._build_url_with_params(url, params=params), + headers=headers, + content=content, + timeout=self._compute_timeout(timeout, attempt=attempt), + stream=stream or False, + ) + except Exception as exc: + self._handle_request_exception(exc, stop_retrying=stop_retrying) + raise + + if self._handle_response_status(response, attempt=attempt, stop_retrying=stop_retrying): + return response + + # Read the response in case it is a stream, so the error can be raised properly. A failed read goes through + # the same classification as a failed send. + try: + await response.aread() + except Exception as exc: + logger.debug('Reading the error response failed', exc_info=exc) + with suppress(Exception): + await response.aclose() + if not self.is_retryable_transport_error(exc): + logger.debug('Exception is not retryable', exc_info=exc) + stop_retrying() + raise + + raise ApifyApiError(response, attempt, method=method) diff --git a/src/apify_client/http_clients/_httpx.py b/src/apify_client/http_clients/_httpx.py new file mode 100644 index 00000000..8e256492 --- /dev/null +++ b/src/apify_client/http_clients/_httpx.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import httpx +from typing_extensions import override + +from apify_client._consts import ( + DEFAULT_MAX_RETRIES, + DEFAULT_MIN_DELAY_BETWEEN_RETRIES, + DEFAULT_TIMEOUT_LONG, + DEFAULT_TIMEOUT_MAX, + DEFAULT_TIMEOUT_MEDIUM, + DEFAULT_TIMEOUT_SHORT, +) +from apify_client._docs import docs_group +from apify_client.http_clients._base import HttpClient, HttpClientAsync + +if TYPE_CHECKING: + from datetime import timedelta + + from apify_client._statistics import ClientStatistics + from apify_client.http_compressors._base import HttpCompressor + + +_PERMANENT_ERRORS = ( + # A request HTTPX rejects before sending it, e.g. one carrying an invalid header value. + httpx.LocalProtocolError, + # A URL scheme HTTPX refuses to speak, which repeating the request cannot change. + httpx.UnsupportedProtocol, + # An over-long redirect chain is a routing loop, which repeating the request cannot break. + httpx.TooManyRedirects, + # Only `Response.raise_for_status()` raises this, and the client never calls it - the shared pipeline decides on + # status codes from the response itself. + httpx.HTTPStatusError, +) +"""HTTPX errors that a retry cannot fix. Everything else in the `httpx.HTTPError` tree counts as transient.""" + + +@docs_group('HTTP clients') +class HttpxHttpClient(HttpClient): + """Synchronous HTTP client for the Apify API built on top of [HTTPX](https://www.python-httpx.org/). + + This client wraps `httpx.Client` and adds automatic retries with exponential backoff for rate-limited + (HTTP 429) and server error (HTTP 5xx) responses. + + Requires the `httpx` extra: `pip install "apify-client[httpx]"`. + """ + + def __init__( + self, + *, + token: str | None = None, + timeout_short: timedelta = DEFAULT_TIMEOUT_SHORT, + timeout_medium: timedelta = DEFAULT_TIMEOUT_MEDIUM, + timeout_long: timedelta = DEFAULT_TIMEOUT_LONG, + timeout_max: timedelta = DEFAULT_TIMEOUT_MAX, + max_retries: int = DEFAULT_MAX_RETRIES, + min_delay_between_retries: timedelta = DEFAULT_MIN_DELAY_BETWEEN_RETRIES, + statistics: ClientStatistics | None = None, + headers: dict[str, str] | None = None, + http_compressor: HttpCompressor | None = None, + ) -> None: + """Initialize the HTTPX-based synchronous HTTP client. + + Args: + token: Apify API token for authentication. + timeout_short: Default timeout for short-duration API operations (simple CRUD operations, ...). + timeout_medium: Default timeout for medium-duration API operations (batch operations, listing, ...). + timeout_long: Default timeout for long-duration API operations (long-polling, streaming, ...). + timeout_max: Maximum timeout cap for any single request attempt, including tier and per-call timeouts. + max_retries: Maximum number of retry attempts for failed requests. + min_delay_between_retries: Minimum delay between retries (increases exponentially with each attempt). + statistics: Statistics tracker for API calls. Created automatically if not provided. + headers: Additional HTTP headers to include in all requests. + http_compressor: Compressor used to compress request bodies. Defaults to `GzipHttpCompressor`. + """ + super().__init__( + token=token, + timeout_short=timeout_short, + timeout_medium=timeout_medium, + timeout_long=timeout_long, + timeout_max=timeout_max, + max_retries=max_retries, + min_delay_between_retries=min_delay_between_retries, + statistics=statistics, + headers=headers, + http_compressor=http_compressor, + ) + + self._httpx_client = httpx.Client( + follow_redirects=True, + event_hooks={'response': [self._clear_response_cookies]}, + ) + + @override + def is_timeout_error(self, exc: Exception) -> bool: + return super().is_timeout_error(exc) or isinstance(exc, httpx.TimeoutException) + + @override + def is_retryable_transport_error(self, exc: Exception) -> bool: + # Every error from HTTPX's own hierarchy counts as transient except the permanently-failing types listed in + # `_PERMANENT_ERRORS`. Retrying is the default because HTTPX also reports genuinely transient failures + # through its generic base class. HTTP status code errors are handled by the shared pipeline based on the + # response status code, not here. + return isinstance(exc, httpx.HTTPError) and not isinstance(exc, _PERMANENT_ERRORS) + + @override + def close(self) -> None: + """Close the underlying HTTPX connection pool.""" + self._httpx_client.close() + + def _clear_response_cookies(self, _response: httpx.Response) -> None: + """Prevent HTTPX's shared cookie jar from leaking server cookies into later API requests.""" + self._httpx_client.cookies.clear() + + @override + def send_request( + self, + *, + method: str, + url: str, + headers: dict[str, str], + content: bytes | None, + timeout: float | None, + stream: bool, + ) -> httpx.Response: + request = self._httpx_client.build_request( + method=method, + url=url, + headers=headers, + content=content, + timeout=timeout, + ) + _restore_explicit_cookie_header(request, headers) + return self._httpx_client.send(request, stream=stream) + + +@docs_group('HTTP clients') +class HttpxHttpClientAsync(HttpClientAsync): + """Asynchronous HTTP client for the Apify API built on top of [HTTPX](https://www.python-httpx.org/). + + This client wraps `httpx.AsyncClient` and adds automatic retries with exponential backoff for rate-limited + (HTTP 429) and server error (HTTP 5xx) responses. + + Requires the `httpx` extra: `pip install "apify-client[httpx]"`. + """ + + def __init__( + self, + *, + token: str | None = None, + timeout_short: timedelta = DEFAULT_TIMEOUT_SHORT, + timeout_medium: timedelta = DEFAULT_TIMEOUT_MEDIUM, + timeout_long: timedelta = DEFAULT_TIMEOUT_LONG, + timeout_max: timedelta = DEFAULT_TIMEOUT_MAX, + max_retries: int = DEFAULT_MAX_RETRIES, + min_delay_between_retries: timedelta = DEFAULT_MIN_DELAY_BETWEEN_RETRIES, + statistics: ClientStatistics | None = None, + headers: dict[str, str] | None = None, + http_compressor: HttpCompressor | None = None, + ) -> None: + """Initialize the HTTPX-based asynchronous HTTP client. + + Args: + token: Apify API token for authentication. + timeout_short: Default timeout for short-duration API operations (simple CRUD operations, ...). + timeout_medium: Default timeout for medium-duration API operations (batch operations, listing, ...). + timeout_long: Default timeout for long-duration API operations (long-polling, streaming, ...). + timeout_max: Maximum timeout cap for any single request attempt, including tier and per-call timeouts. + max_retries: Maximum number of retry attempts for failed requests. + min_delay_between_retries: Minimum delay between retries (increases exponentially with each attempt). + statistics: Statistics tracker for API calls. Created automatically if not provided. + headers: Additional HTTP headers to include in all requests. + http_compressor: Compressor used to compress request bodies. Defaults to `GzipHttpCompressor`. + """ + super().__init__( + token=token, + timeout_short=timeout_short, + timeout_medium=timeout_medium, + timeout_long=timeout_long, + timeout_max=timeout_max, + max_retries=max_retries, + min_delay_between_retries=min_delay_between_retries, + statistics=statistics, + headers=headers, + http_compressor=http_compressor, + ) + + self._httpx_async_client = httpx.AsyncClient( + follow_redirects=True, + event_hooks={'response': [self._clear_response_cookies]}, + ) + + @override + def is_timeout_error(self, exc: Exception) -> bool: + return super().is_timeout_error(exc) or isinstance(exc, httpx.TimeoutException) + + @override + def is_retryable_transport_error(self, exc: Exception) -> bool: + # Every error from HTTPX's own hierarchy counts as transient except the permanently-failing types listed in + # `_PERMANENT_ERRORS`. Retrying is the default because HTTPX also reports genuinely transient failures + # through its generic base class. HTTP status code errors are handled by the shared pipeline based on the + # response status code, not here. + return isinstance(exc, httpx.HTTPError) and not isinstance(exc, _PERMANENT_ERRORS) + + @override + async def aclose(self) -> None: + """Close the underlying asynchronous HTTPX connection pool.""" + await self._httpx_async_client.aclose() + + async def _clear_response_cookies(self, _response: httpx.Response) -> None: + """Prevent HTTPX's shared cookie jar from leaking server cookies into later API requests.""" + self._httpx_async_client.cookies.clear() + + @override + async def send_request( + self, + *, + method: str, + url: str, + headers: dict[str, str], + content: bytes | None, + timeout: float | None, + stream: bool, + ) -> httpx.Response: + request = self._httpx_async_client.build_request( + method=method, + url=url, + headers=headers, + content=content, + timeout=timeout, + ) + _restore_explicit_cookie_header(request, headers) + return await self._httpx_async_client.send(request, stream=stream) + + +def _restore_explicit_cookie_header(request: httpx.Request, headers: dict[str, str]) -> None: + """Keep only cookies explicitly supplied for this request, never cookies from HTTPX's shared jar.""" + explicit_cookie = next((value for key, value in headers.items() if key.lower() == 'cookie'), None) + if explicit_cookie is None: + request.headers.pop('cookie', None) + else: + request.headers['cookie'] = explicit_cookie diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py index 8c45dfda..e18a420b 100644 --- a/src/apify_client/http_clients/_impit.py +++ b/src/apify_client/http_clients/_impit.py @@ -1,15 +1,9 @@ from __future__ import annotations -import asyncio -import logging -import random -import time -from contextlib import suppress -from datetime import timedelta -from http import HTTPStatus -from typing import TYPE_CHECKING, Any, TypeVar +from typing import TYPE_CHECKING import impit +from typing_extensions import override from apify_client._consts import ( DEFAULT_MAX_RETRIES, @@ -20,22 +14,13 @@ DEFAULT_TIMEOUT_SHORT, ) from apify_client._docs import docs_group -from apify_client._logging import log_context, logger_name -from apify_client._utils.time import to_seconds -from apify_client.errors import ApifyApiError from apify_client.http_clients._base import HttpClient, HttpClientAsync if TYPE_CHECKING: - from collections.abc import Awaitable, Callable + from datetime import timedelta from apify_client._statistics import ClientStatistics - from apify_client.http_clients._base import HttpResponse from apify_client.http_compressors._base import HttpCompressor - from apify_client.types import JsonSerializable, Timeout - -T = TypeVar('T') - -logger = logging.getLogger(logger_name) _PERMANENT_ERRORS = ( @@ -47,22 +32,11 @@ impit.UnsupportedProtocol, # An over-long redirect chain is a routing loop, which repeating the request cannot break. impit.TooManyRedirects, - # Only `Response.raise_for_status()` raises this, and the client never calls it - `_make_request` decides on + # Only `Response.raise_for_status()` raises this, and the client never calls it - the shared pipeline decides on # status codes from the response itself. impit.HTTPStatusError, ) - - -def _is_retryable_error(exc: Exception) -> bool: - """Check if an exception represents a transient transport failure that should be retried. - - Every error from Impit's own hierarchy counts as transient except the permanently-failing types listed in - `_PERMANENT_ERRORS`. Retrying is the default because Impit also reports genuinely transient failures through - its generic base class, e.g. a bare `impit.HTTPError` wrapping a failure its internal HTTP library did not - classify. HTTP status code errors are handled separately in `_make_request` based on the response status code, - not here. - """ - return isinstance(exc, impit.HTTPError) and not isinstance(exc, _PERMANENT_ERRORS) +"""Impit errors that a retry cannot fix. Everything else in the `impit.HTTPError` tree counts as transient.""" @docs_group('HTTP clients') @@ -115,214 +89,54 @@ def __init__( http_compressor=http_compressor, ) - self._impit_client = impit.Client( - follow_redirects=True, - ) + self._impit_client = impit.Client(follow_redirects=True) - def call( - self, - *, - method: str, - url: str, - headers: dict[str, str] | None = None, - params: dict[str, Any] | None = None, - data: str | bytes | bytearray | None = None, - json: JsonSerializable | None = None, - stream: bool | None = None, - timeout: Timeout = 'medium', - ) -> HttpResponse: - """Make an HTTP request with automatic retry and exponential backoff. + @override + def is_timeout_error(self, exc: Exception) -> bool: + return super().is_timeout_error(exc) or isinstance(exc, impit.TimeoutException) - Args: - method: HTTP method (GET, POST, PUT, DELETE, etc.). - url: Full URL to make the request to. - headers: Additional headers to include. - params: Query parameters to append to the URL. - data: Raw request body data. Cannot be used together with json. - json: JSON-serializable data for the request body. Cannot be used together with data. - stream: Whether to stream the response body. - timeout: Timeout for the API HTTP request. Use `short`, `medium`, or `long` tier literals for - preconfigured timeouts. A `timedelta` overrides it for this call (capped at `timeout_max`), and - `no_timeout` disables the timeout entirely. + @override + def is_retryable_transport_error(self, exc: Exception) -> bool: + # Every error from Impit's own hierarchy counts as transient except the permanently-failing types listed in + # `_PERMANENT_ERRORS`. Retrying is the default because Impit also reports genuinely transient failures through + # its generic base class, e.g. a bare `impit.HTTPError` wrapping a failure its internal HTTP library did not + # classify. HTTP status code errors are handled by the shared pipeline based on the response status code, + # not here. + return isinstance(exc, impit.HTTPError) and not isinstance(exc, _PERMANENT_ERRORS) - Returns: - The HTTP response object. + @override + def close(self) -> None: + """Release resources owned by this client. - Raises: - ApifyApiError: If the request fails after all retries or returns a non-retryable error status. - ValueError: If both json and data are provided. + Delegates to Impit's own teardown, which releases nothing and leaves the client usable, because Impit + doesn't expose a way to close its connection pool. Routing through it keeps this client correct once + Impit does. """ - log_context.method.set(method) - log_context.url.set(url) - - self._statistics.calls += 1 - - prepared_headers, prepared_params, content = self._prepare_request_call( - headers=headers, - params=params, - data=data, - json=json, - ) + self._impit_client.__exit__(None, None, None) - return self._retry_with_exp_backoff( - lambda stop_retrying, attempt: self._make_request( - stop_retrying=stop_retrying, - attempt=attempt, - method=method, - url=url, - headers=prepared_headers, - params=prepared_params, - content=content, - stream=stream, - timeout=timeout, - ), - max_retries=self._max_retries, - backoff_base=self._min_delay_between_retries, - ) - - def _make_request( + @override + def send_request( self, *, - stop_retrying: Callable[[], None], - attempt: int, method: str, url: str, headers: dict[str, str], - params: dict[str, Any] | None, content: bytes | None, - stream: bool | None, - timeout: Timeout, + timeout: float | None, + stream: bool, ) -> impit.Response: - """Execute a single HTTP request attempt. - - Args: - stop_retrying: Callback to signal that retries should stop. - attempt: Current attempt number (1-indexed). - method: HTTP method. - url: Request URL. - headers: Request headers. - params: Query parameters. - content: Request body content. - stream: Whether to stream the response. - timeout: Timeout for this request. - - Returns: - The HTTP response object. - - Raises: - ApifyApiError: If the request fails with an error status. - """ - log_context.attempt.set(attempt) - logger.debug('Sending request') - - self._statistics.requests += 1 - - try: - url_with_params = self._build_url_with_params(url, params=params) - - # Impit treats timeout=None as "use client default (30s)", not "no timeout". - # Use a large value (24 hours) to effectively disable the timeout. - # This can be removed once impit updates its behaviour: https://github.com/apify/impit/issues/401 - computed_timeout = self._compute_timeout(timeout, attempt=attempt) - impit_timeout = 86_400 if computed_timeout is None else computed_timeout - - response = self._impit_client.request( - method=method, - url=url_with_params, - headers=headers, - content=content, - timeout=impit_timeout, - stream=stream or False, - ) - - if response.status_code < HTTPStatus.MULTIPLE_CHOICES: - logger.debug('Request successful', extra={'status_code': response.status_code}) - return response - - if response.status_code == HTTPStatus.TOO_MANY_REQUESTS: - self._statistics.add_rate_limit_error(attempt) - - except Exception as exc: - logger.debug('Request threw exception', exc_info=exc) - if not _is_retryable_error(exc): - logger.debug('Exception is not retryable', exc_info=exc) - stop_retrying() - raise - - # Retry only server errors (5xx) and rate limits (429). - logger.debug('Request unsuccessful', extra={'status_code': response.status_code}) - if ( - response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR - and response.status_code != HTTPStatus.TOO_MANY_REQUESTS - ): - logger.debug('Status code is not retryable', extra={'status_code': response.status_code}) - stop_retrying() - - # Read the response in case it is a stream, so we can raise the error properly. A failed read goes through - # the same classification as a failed send. - try: - response.read() - except Exception as exc: - logger.debug('Reading the error response failed', exc_info=exc) - with suppress(Exception): - response.close() - if not _is_retryable_error(exc): - logger.debug('Exception is not retryable', exc_info=exc) - stop_retrying() - raise - - raise ApifyApiError(response, attempt, method=method) - - @staticmethod - def _retry_with_exp_backoff( - func: Callable[[Callable[[], None], int], T], - *, - max_retries: int = 8, - backoff_base: timedelta = timedelta(milliseconds=500), - backoff_factor: float = 2, - random_factor: float = 1, - ) -> T: - """Retry a function with exponential backoff and jitter. - - Args: - func: Function to retry. Receives (stop_retrying callback, attempt number). - max_retries: Maximum retry attempts. - backoff_base: Base delay. - backoff_factor: Exponential multiplier (clamped to 1-10). - random_factor: Jitter factor (clamped to 0-1). - - Returns: - The function's return value on success. - - Raises: - Exception: Re-raises the last exception if all retries fail or stop_retrying is called. - """ - if max_retries < 1: - raise ValueError(f'max_retries must be at least 1, got {max_retries}') - - random_factor = min(max(0, random_factor), 1) - backoff_factor = min(max(1, backoff_factor), 10) - swallow = True - - def stop_retrying() -> None: - nonlocal swallow - swallow = False - - for attempt in range(1, max_retries + 1): - try: - return func(stop_retrying, attempt) - except Exception: - if not swallow: - raise - - random_sleep_factor = random.uniform(1, 1 + random_factor) - backoff_base_secs = to_seconds(backoff_base) - backoff_exp_factor = backoff_factor ** (attempt - 1) - - sleep_time_secs = random_sleep_factor * backoff_base_secs * backoff_exp_factor - time.sleep(sleep_time_secs) - - return func(stop_retrying, max_retries + 1) + # Impit treats timeout=None as "use client default (30s)", not "no timeout". + # Use a large value (24 hours) to effectively disable the timeout. + # This can be removed once impit updates its behaviour: https://github.com/apify/impit/issues/401 + impit_timeout = 86_400 if timeout is None else timeout + return self._impit_client.request( + method=method, + url=url, + headers=headers, + content=content, + timeout=impit_timeout, + stream=stream, + ) @docs_group('HTTP clients') @@ -375,224 +189,47 @@ def __init__( http_compressor=http_compressor, ) - self._impit_async_client = impit.AsyncClient( - follow_redirects=True, - ) + self._impit_async_client = impit.AsyncClient(follow_redirects=True) - async def call( - self, - *, - method: str, - url: str, - headers: dict[str, str] | None = None, - params: dict[str, Any] | None = None, - data: str | bytes | bytearray | None = None, - json: JsonSerializable | None = None, - stream: bool | None = None, - timeout: Timeout = 'medium', - ) -> HttpResponse: - """Make an HTTP request with automatic retry and exponential backoff. + @override + def is_timeout_error(self, exc: Exception) -> bool: + return super().is_timeout_error(exc) or isinstance(exc, impit.TimeoutException) - Args: - method: HTTP method (GET, POST, PUT, DELETE, etc.). - url: Full URL to make the request to. - headers: Additional headers to include. - params: Query parameters to append to the URL. - data: Raw request body data. Cannot be used together with json. - json: JSON-serializable data for the request body. Cannot be used together with data. - stream: Whether to stream the response body. - timeout: Timeout for the API HTTP request. Use `short`, `medium`, or `long` tier literals for - preconfigured timeouts. A `timedelta` overrides it for this call (capped at `timeout_max`), and - `no_timeout` disables the timeout entirely. + @override + def is_retryable_transport_error(self, exc: Exception) -> bool: + # Every error from Impit's own hierarchy counts as transient except the permanently-failing types listed in + # `_PERMANENT_ERRORS`. Retrying is the default because Impit also reports genuinely transient failures through + # its generic base class, e.g. a bare `impit.HTTPError` wrapping a failure its internal HTTP library did not + # classify. HTTP status code errors are handled by the shared pipeline based on the response status code, + # not here. + return isinstance(exc, impit.HTTPError) and not isinstance(exc, _PERMANENT_ERRORS) - Returns: - The HTTP response object. + @override + async def aclose(self) -> None: + """Release resources owned by this client. - Raises: - ApifyApiError: If the request fails after all retries or returns a non-retryable error status. - ValueError: If both json and data are provided. + See `ImpitHttpClient.close` for what Impit's teardown does. """ - log_context.method.set(method) - log_context.url.set(url) - - self._statistics.calls += 1 + await self._impit_async_client.__aexit__(None, None, None) - # Serializing and compressing a request body is CPU-bound and would block the event loop, so - # offload preparation to a worker thread whenever there is something to compress. A body the - # client sends as it is costs less to prepare inline than the hop itself. A `json` body always - # hops, as its size is only known once serialized. - if json is not None or self._is_body_worth_compressing(data): - prepared_headers, prepared_params, content = await asyncio.to_thread( - self._prepare_request_call, - headers=headers, - params=params, - data=data, - json=json, - ) - else: - prepared_headers, prepared_params, content = self._prepare_request_call( - headers=headers, - params=params, - data=data, - json=json, - ) - - return await self._retry_with_exp_backoff( - lambda stop_retrying, attempt: self._make_request( - stop_retrying=stop_retrying, - attempt=attempt, - method=method, - url=url, - headers=prepared_headers, - params=prepared_params, - content=content, - stream=stream, - timeout=timeout, - ), - max_retries=self._max_retries, - backoff_base=self._min_delay_between_retries, - ) - - async def _make_request( + @override + async def send_request( self, *, - stop_retrying: Callable[[], None], - attempt: int, method: str, url: str, headers: dict[str, str], - params: dict[str, Any] | None, content: bytes | None, - stream: bool | None, - timeout: Timeout, + timeout: float | None, + stream: bool, ) -> impit.Response: - """Execute a single HTTP request attempt. - - Args: - stop_retrying: Callback to signal that retries should stop. - attempt: Current attempt number (1-indexed). - method: HTTP method. - url: Request URL. - headers: Request headers. - params: Query parameters. - content: Request body content. - stream: Whether to stream the response. - timeout: Timeout for this request. - - Returns: - The HTTP response object. - - Raises: - ApifyApiError: If the request fails with an error status. - """ - log_context.attempt.set(attempt) - logger.debug('Sending request') - - self._statistics.requests += 1 - - try: - url_with_params = self._build_url_with_params(url, params=params) - - # Impit treats timeout=None as "use client default (30s)", not "no timeout". - # Use a large value (24 hours) to effectively disable the timeout. - # This can be removed once impit updates its behaviour: https://github.com/apify/impit/issues/401 - computed_timeout = self._compute_timeout(timeout, attempt=attempt) - impit_timeout = 86_400 if computed_timeout is None else computed_timeout - - response = await self._impit_async_client.request( - method=method, - url=url_with_params, - headers=headers, - content=content, - timeout=impit_timeout, - stream=stream or False, - ) - - if response.status_code < HTTPStatus.MULTIPLE_CHOICES: - logger.debug('Request successful', extra={'status_code': response.status_code}) - return response - - if response.status_code == HTTPStatus.TOO_MANY_REQUESTS: - self._statistics.add_rate_limit_error(attempt) - - except Exception as exc: - logger.debug('Request threw exception', exc_info=exc) - if not _is_retryable_error(exc): - logger.debug('Exception is not retryable', exc_info=exc) - stop_retrying() - raise - - # Retry only server errors (5xx) and rate limits (429). - logger.debug('Request unsuccessful', extra={'status_code': response.status_code}) - if ( - response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR - and response.status_code != HTTPStatus.TOO_MANY_REQUESTS - ): - logger.debug('Status code is not retryable', extra={'status_code': response.status_code}) - stop_retrying() - - # Read the response in case it is a stream, so we can raise the error properly. A failed read goes through - # the same classification as a failed send. - try: - await response.aread() - except Exception as exc: - logger.debug('Reading the error response failed', exc_info=exc) - with suppress(Exception): - await response.aclose() - if not _is_retryable_error(exc): - logger.debug('Exception is not retryable', exc_info=exc) - stop_retrying() - raise - - raise ApifyApiError(response, attempt, method=method) - - @staticmethod - async def _retry_with_exp_backoff( - func: Callable[[Callable[[], None], int], Awaitable[T]], - *, - max_retries: int = 8, - backoff_base: timedelta = timedelta(milliseconds=500), - backoff_factor: float = 2, - random_factor: float = 1, - ) -> T: - """Retry an async function with exponential backoff and jitter. - - Args: - func: Async function to retry. Receives (stop_retrying callback, attempt number). - max_retries: Maximum retry attempts. - backoff_base: Base delay. - backoff_factor: Exponential multiplier (clamped to 1-10). - random_factor: Jitter factor (clamped to 0-1). - - Returns: - The function's return value on success. - - Raises: - Exception: Re-raises the last exception if all retries fail or stop_retrying is called. - """ - if max_retries < 1: - raise ValueError(f'max_retries must be at least 1, got {max_retries}') - - random_factor = min(max(0, random_factor), 1) - backoff_factor = min(max(1, backoff_factor), 10) - swallow = True - - def stop_retrying() -> None: - nonlocal swallow - swallow = False - - for attempt in range(1, max_retries + 1): - try: - return await func(stop_retrying, attempt) - except Exception: - if not swallow: - raise - - random_sleep_factor = random.uniform(1, 1 + random_factor) - backoff_base_secs = to_seconds(backoff_base) - backoff_exp_factor = backoff_factor ** (attempt - 1) - - sleep_time_secs = random_sleep_factor * backoff_base_secs * backoff_exp_factor - await asyncio.sleep(sleep_time_secs) - - return await func(stop_retrying, max_retries + 1) + # See the synchronous implementation for why None maps to 24 hours. + impit_timeout = 86_400 if timeout is None else timeout + return await self._impit_async_client.request( + method=method, + url=url, + headers=headers, + content=content, + timeout=impit_timeout, + stream=stream, + ) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 1db53b71..e0c9bed6 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -2,6 +2,7 @@ import json import os +from dataclasses import dataclass from typing import TYPE_CHECKING import pytest @@ -17,9 +18,35 @@ from apify_client import ApifyClient, ApifyClientAsync from apify_client._consts import DEFAULT_API_URL from apify_client._utils.crypto import create_hmac_signature, create_storage_content_signature +from apify_client.http_clients import ( + HttpClient, + HttpClientAsync, + HttpxHttpClient, + HttpxHttpClientAsync, + ImpitHttpClient, + ImpitHttpClientAsync, +) if TYPE_CHECKING: - from collections.abc import Generator + from collections.abc import AsyncGenerator, Generator + + +@dataclass(frozen=True) +class HttpClientClasses: + """Synchronous and asynchronous variants of a built-in HTTP client.""" + + sync: type[HttpClient] + async_: type[HttpClientAsync] + + +DEFAULT_HTTP_CLIENT_CLASSES = HttpClientClasses(sync=ImpitHttpClient, async_=ImpitHttpClientAsync) +"""HTTP clients the live-API suite runs with unless a test asks for another transport.""" + +ALL_HTTP_CLIENT_CLASSES = [ + pytest.param(DEFAULT_HTTP_CLIENT_CLASSES, id='impit'), + pytest.param(HttpClientClasses(sync=HttpxHttpClient, async_=HttpxHttpClientAsync), id='httpx'), +] +"""Every built-in HTTP client, for tests that exercise transport behavior rather than an API resource.""" # ============================================================================ @@ -110,17 +137,17 @@ def test_kvs_of_another_user(api_token_2: str) -> Generator[KvsFixture]: @pytest.fixture -def apify_client(api_token: str) -> ApifyClient: - """Sync Apify client instance.""" - api_url = os.getenv(API_URL_ENV_VAR) or DEFAULT_API_URL - return ApifyClient(api_token, api_url=api_url) +def http_client_classes(request: pytest.FixtureRequest) -> HttpClientClasses: + """Return the sync and async classes of the HTTP client the test runs with. + Defaults to Impit so the live-API suite isn't multiplied by every transport. A transport-level test opts into + the full matrix with `@pytest.mark.parametrize('http_client_classes', ALL_HTTP_CLIENT_CLASSES, indirect=True)`. + """ + if not hasattr(request, 'param'): + return DEFAULT_HTTP_CLIENT_CLASSES -@pytest.fixture -def apify_client_async(api_token: str) -> ApifyClientAsync: - """Async Apify client instance.""" - api_url = os.getenv(API_URL_ENV_VAR) or DEFAULT_API_URL - return ApifyClientAsync(api_token, api_url=api_url) + assert isinstance(request.param, HttpClientClasses) + return request.param @pytest.fixture(params=['sync', 'async']) @@ -130,13 +157,30 @@ def client_type(request: pytest.FixtureRequest) -> str: @pytest.fixture -def client( +async def client( client_type: str, - apify_client: ApifyClient, - apify_client_async: ApifyClientAsync, -) -> ApifyClient | ApifyClientAsync: - """Return sync or async client based on parametrization.""" - return apify_client if client_type == 'sync' else apify_client_async + api_token: str, + http_client_classes: HttpClientClasses, +) -> AsyncGenerator[ApifyClient | ApifyClientAsync]: + """Return each sync/async and HTTP client implementation combination.""" + api_url = os.getenv(API_URL_ENV_VAR) or DEFAULT_API_URL + if client_type == 'sync': + http_client = http_client_classes.sync() + yield ApifyClient.with_custom_http_client( + api_token, + api_url=api_url, + http_client=http_client, + ) + http_client.close() + return + + http_client_async = http_client_classes.async_() + yield ApifyClientAsync.with_custom_http_client( + api_token, + api_url=api_url, + http_client=http_client_async, + ) + await http_client_async.aclose() @pytest.fixture diff --git a/tests/integration/test_apify_client.py b/tests/integration/test_apify_client.py index 126f40b3..4c15eab8 100644 --- a/tests/integration/test_apify_client.py +++ b/tests/integration/test_apify_client.py @@ -4,13 +4,17 @@ from typing import TYPE_CHECKING +import pytest + from .._utils import maybe_await +from .conftest import ALL_HTTP_CLIENT_CLASSES from apify_client._models import UserPrivateInfo, UserPublicInfo if TYPE_CHECKING: from apify_client import ApifyClient, ApifyClientAsync +@pytest.mark.parametrize('http_client_classes', ALL_HTTP_CLIENT_CLASSES, indirect=True) async def test_apify_client(client: ApifyClient | ApifyClientAsync) -> None: """Test basic apify client functionality.""" user_client = client.user('me') diff --git a/tests/integration/test_dataset.py b/tests/integration/test_dataset.py index 333c7229..b7acab4c 100644 --- a/tests/integration/test_dataset.py +++ b/tests/integration/test_dataset.py @@ -18,6 +18,7 @@ maybe_await, poll_until_condition, ) +from .conftest import ALL_HTTP_CLIENT_CLASSES from apify_client._models import Dataset, DatasetListItem, DatasetStatistics, ListOfDatasets from apify_client._resource_clients.dataset import DatasetItemsPage from apify_client.errors import ApifyApiError @@ -698,6 +699,7 @@ async def get_items() -> DatasetItemsPage: await maybe_await(dataset_client.delete()) +@pytest.mark.parametrize('http_client_classes', ALL_HTTP_CLIENT_CLASSES, indirect=True) async def test_dataset_stream_items(client: ApifyClient | ApifyClientAsync, *, is_async: bool) -> None: """Test streaming dataset items.""" dataset_name = get_random_resource_name('dataset') diff --git a/tests/integration/test_key_value_store.py b/tests/integration/test_key_value_store.py index 5d1d9238..fee82954 100644 --- a/tests/integration/test_key_value_store.py +++ b/tests/integration/test_key_value_store.py @@ -19,6 +19,7 @@ maybe_sleep, poll_until_condition, ) +from .conftest import ALL_HTTP_CLIENT_CLASSES from apify_client._models import KeyValueStore, KeyValueStoreKey, ListOfKeys, ListOfKeyValueStores from apify_client.errors import ApifyApiError from apify_client.http_clients import HttpResponse @@ -706,6 +707,7 @@ async def get_keys() -> ListOfKeys: await maybe_await(store_client.delete()) +@pytest.mark.parametrize('http_client_classes', ALL_HTTP_CLIENT_CLASSES, indirect=True) async def test_key_value_store_stream_record_own(client: ApifyClient | ApifyClientAsync, *, is_async: bool) -> None: """Test streaming a record from one's own key-value store (no signature).""" store_name = get_random_resource_name('kvs') diff --git a/tests/integration/test_log.py b/tests/integration/test_log.py index df687e2f..13db91f8 100644 --- a/tests/integration/test_log.py +++ b/tests/integration/test_log.py @@ -5,7 +5,10 @@ from contextlib import AbstractAsyncContextManager, AbstractContextManager from typing import TYPE_CHECKING +import pytest + from .._utils import maybe_await +from .conftest import ALL_HTTP_CLIENT_CLASSES from apify_client._models import ListOfBuilds, Run from apify_client.http_clients import HttpResponse @@ -72,6 +75,7 @@ async def test_log_get_as_bytes(client: ApifyClient | ApifyClientAsync) -> None: await maybe_await(run_client.delete()) +@pytest.mark.parametrize('http_client_classes', ALL_HTTP_CLIENT_CLASSES, indirect=True) async def test_log_stream_from_run(client: ApifyClient | ApifyClientAsync, *, is_async: bool) -> None: """Test streaming a run's log via the stream() context manager.""" actor = client.actor(HELLO_WORLD_ACTOR) diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index b732cc1c..d1391a3a 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -6,7 +6,14 @@ import pytest from pytest_httpserver import HTTPServer -from apify_client.http_clients import HttpClient, HttpClientAsync, ImpitHttpClient, ImpitHttpClientAsync +from apify_client.http_clients import ( + HttpClient, + HttpClientAsync, + HttpxHttpClient, + HttpxHttpClientAsync, + ImpitHttpClient, + ImpitHttpClientAsync, +) if TYPE_CHECKING: from collections.abc import Iterable @@ -32,13 +39,23 @@ def httpserver(make_httpserver: HTTPServer) -> Iterable[HTTPServer]: server.clear() -@pytest.fixture(params=[pytest.param(ImpitHttpClient, id='impit')]) +@pytest.fixture( + params=[ + pytest.param(ImpitHttpClient, id='impit'), + pytest.param(HttpxHttpClient, id='httpx'), + ] +) def http_client_class(request: pytest.FixtureRequest) -> type[HttpClient]: """Return each built-in synchronous HTTP client class.""" return request.param -@pytest.fixture(params=[pytest.param(ImpitHttpClientAsync, id='impit')]) +@pytest.fixture( + params=[ + pytest.param(ImpitHttpClientAsync, id='impit'), + pytest.param(HttpxHttpClientAsync, id='httpx'), + ] +) def http_client_async_class(request: pytest.FixtureRequest) -> type[HttpClientAsync]: """Return each built-in asynchronous HTTP client class.""" return request.param diff --git a/tests/unit/test_client_errors.py b/tests/unit/test_client_errors.py index a0c1c0b9..61f3998a 100644 --- a/tests/unit/test_client_errors.py +++ b/tests/unit/test_client_errors.py @@ -18,7 +18,6 @@ ServerError, UnauthorizedError, ) -from apify_client.http_clients import ImpitHttpClient, ImpitHttpClientAsync if TYPE_CHECKING: from collections.abc import Awaitable, Callable @@ -26,6 +25,8 @@ from pytest_httpserver import HTTPServer from werkzeug import Request + from apify_client.http_clients import HttpClient, HttpClientAsync + _TEST_PATH = '/errors' _EXPECTED_MESSAGE = 'some_message' _EXPECTED_TYPE = 'some_type' @@ -84,13 +85,21 @@ def streaming_handler(_request: Request) -> Response: @pytest.fixture -def sync_client(httpserver: HTTPServer) -> ApifyClient: - return ApifyClient(token='test', api_url=httpserver.url_for('/').removesuffix('/')) +def sync_client(httpserver: HTTPServer, http_client_class: type[HttpClient]) -> ApifyClient: + return ApifyClient.with_custom_http_client( + token='test', + api_url=httpserver.url_for('/').removesuffix('/'), + http_client=http_client_class(), + ) @pytest.fixture -def async_client(httpserver: HTTPServer) -> ApifyClientAsync: - return ApifyClientAsync(token='test', api_url=httpserver.url_for('/').removesuffix('/')) +def async_client(httpserver: HTTPServer, http_client_async_class: type[HttpClientAsync]) -> ApifyClientAsync: + return ApifyClientAsync.with_custom_http_client( + token='test', + api_url=httpserver.url_for('/').removesuffix('/'), + http_client=http_client_async_class(), + ) @pytest.fixture @@ -101,9 +110,9 @@ def test_endpoint(httpserver: HTTPServer) -> str: return str(httpserver.url_for(_TEST_PATH)) -def test_client_apify_api_error_with_data(test_endpoint: str) -> None: +def test_client_apify_api_error_with_data(test_endpoint: str, http_client_class: type[HttpClient]) -> None: """Test that client correctly throws ApifyApiError with error data from response.""" - client = ImpitHttpClient() + client = http_client_class() with pytest.raises(ApifyApiError) as exc: client.call(method='GET', url=test_endpoint) @@ -113,9 +122,11 @@ def test_client_apify_api_error_with_data(test_endpoint: str) -> None: assert exc.value.data == _EXPECTED_DATA -async def test_async_client_apify_api_error_with_data(test_endpoint: str) -> None: +async def test_async_client_apify_api_error_with_data( + test_endpoint: str, http_client_async_class: type[HttpClientAsync] +) -> None: """Test that async client correctly throws ApifyApiError with error data from response.""" - client = ImpitHttpClientAsync() + client = http_client_async_class() with pytest.raises(ApifyApiError) as exc: await client.call(method='GET', url=test_endpoint) @@ -125,12 +136,12 @@ async def test_async_client_apify_api_error_with_data(test_endpoint: str) -> Non assert exc.value.data == _EXPECTED_DATA -def test_client_apify_api_error_streamed(httpserver: HTTPServer) -> None: +def test_client_apify_api_error_streamed(httpserver: HTTPServer, http_client_class: type[HttpClient]) -> None: """Test that client correctly throws ApifyApiError when the response has stream.""" error = json.loads(RAW_ERROR.decode()) - client = ImpitHttpClient() + client = http_client_class() httpserver.expect_request('/stream_error').respond_with_handler(streaming_handler) @@ -141,12 +152,14 @@ def test_client_apify_api_error_streamed(httpserver: HTTPServer) -> None: assert exc.value.type == error['error']['type'] -async def test_async_client_apify_api_error_streamed(httpserver: HTTPServer) -> None: +async def test_async_client_apify_api_error_streamed( + httpserver: HTTPServer, http_client_async_class: type[HttpClientAsync] +) -> None: """Test that async client correctly throws ApifyApiError when the response has stream.""" error = json.loads(RAW_ERROR.decode()) - client = ImpitHttpClientAsync() + client = http_client_async_class() httpserver.expect_request('/stream_error').respond_with_handler(streaming_handler) @@ -157,12 +170,14 @@ async def test_async_client_apify_api_error_streamed(httpserver: HTTPServer) -> assert exc.value.type == error['error']['type'] -def test_apify_api_error_dispatches_to_subclass_for_known_status(httpserver: HTTPServer) -> None: +def test_apify_api_error_dispatches_to_subclass_for_known_status( + httpserver: HTTPServer, http_client_class: type[HttpClient] +) -> None: """Mapped HTTP status codes dispatch to their matching subclass.""" httpserver.expect_request('/dispatch').respond_with_json( {'error': {'type': 'record-not-found', 'message': 'nope'}}, status=404 ) - client = ImpitHttpClient() + client = http_client_class() with pytest.raises(NotFoundError) as exc: client.call(method='GET', url=str(httpserver.url_for('/dispatch'))) @@ -173,10 +188,12 @@ def test_apify_api_error_dispatches_to_subclass_for_known_status(httpserver: HTT assert exc.value.type == 'record-not-found' -def test_apify_api_error_dispatches_streamed_response(httpserver: HTTPServer) -> None: +def test_apify_api_error_dispatches_streamed_response( + httpserver: HTTPServer, http_client_class: type[HttpClient] +) -> None: """Dispatch works even when the response body comes in as a stream (403 → ForbiddenError).""" httpserver.expect_request('/stream_dispatch').respond_with_handler(streaming_handler) - client = ImpitHttpClient() + client = http_client_class() with pytest.raises(ForbiddenError) as exc: client.call(method='GET', url=httpserver.url_for('/stream_dispatch'), stream=True) @@ -186,12 +203,14 @@ def test_apify_api_error_dispatches_streamed_response(httpserver: HTTPServer) -> assert exc.value.type == 'insufficient-permissions' -def test_apify_api_error_dispatches_5xx_to_server_error(httpserver: HTTPServer) -> None: +def test_apify_api_error_dispatches_5xx_to_server_error( + httpserver: HTTPServer, http_client_class: type[HttpClient] +) -> None: """Any 5xx status falls under the ServerError subclass.""" httpserver.expect_request('/server_error').respond_with_json( {'error': {'type': 'internal-error', 'message': 'boom'}}, status=503 ) - client = ImpitHttpClient(max_retries=1) + client = http_client_class(max_retries=1) with pytest.raises(ServerError) as exc: client.call(method='GET', url=str(httpserver.url_for('/server_error'))) @@ -200,12 +219,14 @@ def test_apify_api_error_dispatches_5xx_to_server_error(httpserver: HTTPServer) assert exc.value.status_code == 503 -def test_apify_api_error_falls_back_for_unmapped_status(httpserver: HTTPServer) -> None: +def test_apify_api_error_falls_back_for_unmapped_status( + httpserver: HTTPServer, http_client_class: type[HttpClient] +) -> None: """Statuses without a dedicated subclass fall back to the base ApifyApiError.""" httpserver.expect_request('/unmapped').respond_with_json( {'error': {'type': 'whatever', 'message': 'nope'}}, status=418 ) - client = ImpitHttpClient() + client = http_client_class() with pytest.raises(ApifyApiError) as exc: client.call(method='GET', url=str(httpserver.url_for('/unmapped'))) @@ -227,14 +248,17 @@ def test_apify_api_error_falls_back_for_unmapped_status(httpserver: HTTPServer) ], ) def test_apify_api_error_dispatches_all_mapped_statuses( - httpserver: HTTPServer, status_code: int, expected_cls: type[ApifyApiError] + httpserver: HTTPServer, + http_client_class: type[HttpClient], + status_code: int, + expected_cls: type[ApifyApiError], ) -> None: """Every status in `_STATUS_TO_CLASS` dispatches to its matching subclass.""" httpserver.expect_request('/dispatch_all').respond_with_json( {'error': {'type': 'some-type', 'message': 'msg'}}, status=status_code ) # Use max_retries=1 so retryable statuses (429) don't loop during the test. - client = ImpitHttpClient(max_retries=1) + client = http_client_class(max_retries=1) with pytest.raises(expected_cls) as exc: client.call(method='GET', url=str(httpserver.url_for('/dispatch_all'))) @@ -258,10 +282,12 @@ def test_apify_api_error_subclass_constructed_directly_keeps_its_class() -> None assert error.type == 'record-not-found' -def test_apify_api_error_falls_back_for_unparsable_body(httpserver: HTTPServer) -> None: +def test_apify_api_error_falls_back_for_unparsable_body( + httpserver: HTTPServer, http_client_class: type[HttpClient] +) -> None: """When the body can't be parsed, status-based dispatch still applies and `.type` is None.""" httpserver.expect_request('/unparsable').respond_with_data('', status=418, content_type='text/html') - client = ImpitHttpClient(max_retries=1) + client = http_client_class(max_retries=1) with pytest.raises(ApifyApiError) as exc: client.call(method='GET', url=str(httpserver.url_for('/unparsable'))) diff --git a/tests/unit/test_client_headers.py b/tests/unit/test_client_headers.py index d0a6ed78..b8e0b259 100644 --- a/tests/unit/test_client_headers.py +++ b/tests/unit/test_client_headers.py @@ -6,19 +6,34 @@ from importlib import metadata from typing import TYPE_CHECKING +import httpx from werkzeug import Request, Response -from apify_client.http_clients import ImpitHttpClient, ImpitHttpClientAsync +from apify_client.http_clients import HttpxHttpClient, HttpxHttpClientAsync, ImpitHttpClient, ImpitHttpClientAsync if TYPE_CHECKING: from pytest_httpserver import HTTPServer + from apify_client.http_clients import HttpClient, HttpClientAsync + def _parse_accept_encoding(header: str) -> set[str]: """Parse Accept-Encoding header into a set of encoding names, ignoring order and whitespace.""" return {enc.strip() for enc in header.split(',')} +def _transport_wire_headers( + client_class: type[HttpClient | HttpClientAsync], +) -> tuple[dict[str, str], set[str]]: + """Return the headers the transport adds on its own and the content encodings it advertises.""" + if issubclass(client_class, (ImpitHttpClient, ImpitHttpClientAsync)): + return {}, {'zstd', 'gzip', 'deflate', 'br'} + # HTTPX advertises whichever decoders happen to be installed alongside it, so read the set off the client + # itself rather than hard-coding it and breaking whenever the environment gains or loses a codec. + with httpx.Client() as probe: + return {'Connection': 'keep-alive'}, _parse_accept_encoding(probe.headers['accept-encoding']) + + def _header_handler(request: Request) -> Response: return Response( status=200, @@ -34,49 +49,53 @@ def _get_user_agent() -> str: return f'ApifyClient/{client_version} ({sys.platform}; Python/{python_version}); isAtHome/{is_at_home}' -async def test_default_headers_async(httpserver: HTTPServer) -> None: +async def test_default_headers_async(httpserver: HTTPServer, http_client_async_class: type[HttpClientAsync]) -> None: """Test that default headers are sent with each request.""" - client = ImpitHttpClientAsync(token='placeholder_token') + client = http_client_async_class(token='placeholder_token') httpserver.expect_request('/').respond_with_handler(_header_handler) api_url = httpserver.url_for('/').removesuffix('/') response = await client.call(method='GET', url=f'{api_url}/') request_headers = json.loads(response.text)['received_headers'] + transport_headers, expected_encodings = _transport_wire_headers(http_client_async_class) expected_headers = { 'User-Agent': _get_user_agent(), 'Accept': 'application/json, */*', 'Authorization': 'Bearer placeholder_token', 'Host': f'{httpserver.host}:{httpserver.port}', + **transport_headers, } assert {k: v for k, v in request_headers.items() if k != 'Accept-Encoding'} == expected_headers - assert _parse_accept_encoding(request_headers['Accept-Encoding']) == {'gzip', 'br', 'zstd', 'deflate'} + assert _parse_accept_encoding(request_headers['Accept-Encoding']) == expected_encodings -def test_default_headers_sync(httpserver: HTTPServer) -> None: +def test_default_headers_sync(httpserver: HTTPServer, http_client_class: type[HttpClient]) -> None: """Test that default headers are sent with each request.""" - client = ImpitHttpClient(token='placeholder_token') + client = http_client_class(token='placeholder_token') httpserver.expect_request('/').respond_with_handler(_header_handler) api_url = httpserver.url_for('/').removesuffix('/') response = client.call(method='GET', url=f'{api_url}/') request_headers = json.loads(response.text)['received_headers'] + transport_headers, expected_encodings = _transport_wire_headers(http_client_class) expected_headers = { 'User-Agent': _get_user_agent(), 'Accept': 'application/json, */*', 'Authorization': 'Bearer placeholder_token', 'Host': f'{httpserver.host}:{httpserver.port}', + **transport_headers, } assert {k: v for k, v in request_headers.items() if k != 'Accept-Encoding'} == expected_headers - assert _parse_accept_encoding(request_headers['Accept-Encoding']) == {'gzip', 'br', 'zstd', 'deflate'} + assert _parse_accept_encoding(request_headers['Accept-Encoding']) == expected_encodings -async def test_headers_async(httpserver: HTTPServer) -> None: +async def test_headers_async(httpserver: HTTPServer, http_client_async_class: type[HttpClientAsync]) -> None: """Test that custom headers are sent with each request.""" - client = ImpitHttpClientAsync( + client = http_client_async_class( token='placeholder_token', headers={'Test-Header': 'blah', 'User-Agent': 'CustomUserAgent/1.0', 'Authorization': 'strange_value'}, ) @@ -86,6 +105,7 @@ async def test_headers_async(httpserver: HTTPServer) -> None: response = await client.call(method='GET', url=f'{api_url}/') request_headers = json.loads(response.text)['received_headers'] + transport_headers, expected_encodings = _transport_wire_headers(http_client_async_class) expected_headers = { 'Test-Header': 'blah', @@ -93,14 +113,15 @@ async def test_headers_async(httpserver: HTTPServer) -> None: 'Accept': 'application/json, */*', 'Authorization': 'strange_value', 'Host': f'{httpserver.host}:{httpserver.port}', + **transport_headers, } assert {k: v for k, v in request_headers.items() if k != 'Accept-Encoding'} == expected_headers - assert _parse_accept_encoding(request_headers['Accept-Encoding']) == {'gzip', 'br', 'zstd', 'deflate'} + assert _parse_accept_encoding(request_headers['Accept-Encoding']) == expected_encodings -def test_headers_sync(httpserver: HTTPServer) -> None: +def test_headers_sync(httpserver: HTTPServer, http_client_class: type[HttpClient]) -> None: """Test that custom headers are sent with each request.""" - client = ImpitHttpClient( + client = http_client_class( token='placeholder_token', headers={ 'Test-Header': 'blah', @@ -114,6 +135,7 @@ def test_headers_sync(httpserver: HTTPServer) -> None: response = client.call(method='GET', url=f'{api_url}/') request_headers = json.loads(response.text)['received_headers'] + transport_headers, expected_encodings = _transport_wire_headers(http_client_class) expected_headers = { 'Test-Header': 'blah', @@ -121,14 +143,17 @@ def test_headers_sync(httpserver: HTTPServer) -> None: 'Accept': 'application/json, */*', 'Authorization': 'strange_value', 'Host': f'{httpserver.host}:{httpserver.port}', + **transport_headers, } assert {k: v for k, v in request_headers.items() if k != 'Accept-Encoding'} == expected_headers - assert _parse_accept_encoding(request_headers['Accept-Encoding']) == {'gzip', 'br', 'zstd', 'deflate'} + assert _parse_accept_encoding(request_headers['Accept-Encoding']) == expected_encodings -async def test_per_request_headers_override_defaults_async(httpserver: HTTPServer) -> None: +async def test_per_request_headers_override_defaults_async( + httpserver: HTTPServer, http_client_async_class: type[HttpClientAsync] +) -> None: """Test that a per-request header overrides a same-named default header on the wire, without duplication.""" - client = ImpitHttpClientAsync(token='placeholder_token') + client = http_client_async_class(token='placeholder_token') httpserver.expect_request('/').respond_with_handler(_header_handler) api_url = httpserver.url_for('/').removesuffix('/') @@ -141,9 +166,11 @@ async def test_per_request_headers_override_defaults_async(httpserver: HTTPServe assert request_headers['Authorization'] == 'Bearer per-request' -def test_per_request_headers_override_defaults_sync(httpserver: HTTPServer) -> None: +def test_per_request_headers_override_defaults_sync( + httpserver: HTTPServer, http_client_class: type[HttpClient] +) -> None: """Test that a per-request header overrides a same-named default header on the wire, without duplication.""" - client = ImpitHttpClient(token='placeholder_token') + client = http_client_class(token='placeholder_token') httpserver.expect_request('/').respond_with_handler(_header_handler) api_url = httpserver.url_for('/').removesuffix('/') @@ -154,3 +181,45 @@ def test_per_request_headers_override_defaults_sync(httpserver: HTTPServer) -> N # WSGI joins duplicate headers into one comma-separated value, so exact equality # also proves the authorization header was sent only once. assert request_headers['Authorization'] == 'Bearer per-request' + + +def _echo_cookie_handler(request: Request) -> Response: + return Response(json.dumps({'cookie': request.headers.get('Cookie')}), content_type='application/json') + + +def test_httpx_does_not_reuse_server_cookies(httpserver: HTTPServer) -> None: + """A Set-Cookie response must not silently leak into a later API request through HTTPX's shared cookie jar.""" + httpserver.expect_request('/set-cookie').respond_with_data('ok', headers={'Set-Cookie': 'session=secret'}) + httpserver.expect_request('/echo-cookie').respond_with_handler(_echo_cookie_handler) + + with HttpxHttpClient() as client: + client.call(method='GET', url=httpserver.url_for('/set-cookie')) + response = client.call(method='GET', url=httpserver.url_for('/echo-cookie')) + + assert response.json() == {'cookie': None} + + +async def test_httpx_async_does_not_reuse_server_cookies(httpserver: HTTPServer) -> None: + """The asynchronous HTTPX pool also remains stateless between API calls.""" + httpserver.expect_request('/set-cookie').respond_with_data('ok', headers={'Set-Cookie': 'session=secret'}) + httpserver.expect_request('/echo-cookie').respond_with_handler(_echo_cookie_handler) + + async with HttpxHttpClientAsync() as client: + await client.call(method='GET', url=httpserver.url_for('/set-cookie')) + response = await client.call(method='GET', url=httpserver.url_for('/echo-cookie')) + + assert response.json() == {'cookie': None} + + +def test_httpx_keeps_explicit_cookie_header(httpserver: HTTPServer) -> None: + """Disabling the shared cookie jar must not remove a Cookie header explicitly supplied by the caller.""" + httpserver.expect_request('/echo-explicit-cookie').respond_with_handler(_echo_cookie_handler) + + with HttpxHttpClient() as client: + response = client.call( + method='GET', + url=httpserver.url_for('/echo-explicit-cookie'), + headers={'Cookie': 'explicit=value'}, + ) + + assert response.json() == {'cookie': 'explicit=value'} diff --git a/tests/unit/test_client_streaming.py b/tests/unit/test_client_streaming.py index d369455e..9e5782e7 100644 --- a/tests/unit/test_client_streaming.py +++ b/tests/unit/test_client_streaming.py @@ -105,7 +105,7 @@ def test_protocol_check_leaves_stream_unread_sync( with client.dataset(DATASET_ID).stream_items(item_format='json') as response: assert isinstance(response, HttpResponse) - # `is_stream_consumed` is transport state, not part of the protocol, but the built-in client exposes it. + # `is_stream_consumed` is transport state, not part of the protocol, but both built-in clients expose it. raw: Any = response assert raw.is_stream_consumed is False @@ -124,6 +124,6 @@ async def test_protocol_check_leaves_stream_unread_async( async with client.dataset(DATASET_ID).stream_items(item_format='json') as response: assert isinstance(response, HttpResponse) - # `is_stream_consumed` is transport state, not part of the protocol, but the built-in client exposes it. + # `is_stream_consumed` is transport state, not part of the protocol, but both built-in clients expose it. raw: Any = response assert raw.is_stream_consumed is False diff --git a/tests/unit/test_client_timeouts.py b/tests/unit/test_client_timeouts.py index 03ddf36b..58fd03b2 100644 --- a/tests/unit/test_client_timeouts.py +++ b/tests/unit/test_client_timeouts.py @@ -3,40 +3,28 @@ import logging from datetime import timedelta from typing import TYPE_CHECKING, Any -from unittest.mock import Mock +from unittest.mock import AsyncMock, Mock +import httpx +import impit import pytest -from impit import HTTPError, Response, TimeoutException from apify_client._logging import LoggerOnce, logger_name -from apify_client.http_clients import ImpitHttpClient, ImpitHttpClientAsync +from apify_client.http_clients import ( + HttpClient, + HttpClientAsync, + HttpxHttpClient, + HttpxHttpClientAsync, + ImpitHttpClient, + ImpitHttpClientAsync, +) from apify_client.http_clients import _base as http_client_base if TYPE_CHECKING: - from collections.abc import Iterator - from _pytest.logging import LogCaptureFixture - -class EndOfTestError(Exception): - """Custom exception that is raised after the relevant part of the code is executed to stop the test.""" - - -@pytest.fixture -def patch_request(monkeypatch: pytest.MonkeyPatch) -> Iterator[list]: - timeouts = [] - - def mock_request(*_args: Any, **kwargs: Any) -> None: - timeouts.append(kwargs.get('timeout')) - raise EndOfTestError - - async def mock_request_async(*args: Any, **kwargs: Any) -> None: - return mock_request(*args, **kwargs) - - monkeypatch.setattr('impit.Client.request', mock_request) - monkeypatch.setattr('impit.AsyncClient.request', mock_request_async) - yield timeouts - monkeypatch.undo() +UNSET_HTTPX_TIMEOUT = {'connect': None, 'read': None, 'write': None, 'pool': None} +"""What HTTPX stores on a request built with `timeout=None`: every sub-timeout unset, not the client default.""" @pytest.fixture @@ -45,125 +33,99 @@ def fresh_logger_once(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(http_client_base, 'logger_once', LoggerOnce(http_client_base.logger)) -def test_no_timeout_passes_large_value_to_impit_sync(patch_request: list) -> None: - """Test that `no_timeout` passes a large timeout to impit to effectively disable the timeout.""" - client = ImpitHttpClient(timeout_short=timedelta(seconds=10)) - - with pytest.raises(EndOfTestError): - client.call(method='GET', url='http://placeholder.url/no_timeout', timeout='no_timeout') - - assert patch_request == [86_400] - - -async def test_no_timeout_passes_large_value_to_impit_async(patch_request: list) -> None: - """Test that `no_timeout` passes a large timeout to impit to effectively disable the timeout.""" - client = ImpitHttpClientAsync(timeout_short=timedelta(seconds=10)) - - with pytest.raises(EndOfTestError): - await client.call(method='GET', url='http://placeholder.url/no_timeout', timeout='no_timeout') - - assert patch_request == [86_400] - - -def test_default_timeout_uses_medium_tier_sync(patch_request: list) -> None: - """Test that omitting timeout uses the 'medium' tier (sync client).""" - client = ImpitHttpClient(timeout_medium=timedelta(seconds=30)) - - with pytest.raises(EndOfTestError): - client.call(method='GET', url='http://placeholder.url/default_timeout') - - assert patch_request == [30.0] - - -async def test_default_timeout_uses_medium_tier_async(patch_request: list) -> None: - """Test that omitting timeout uses the 'medium' tier (async client).""" - client = ImpitHttpClientAsync(timeout_medium=timedelta(seconds=30)) - - with pytest.raises(EndOfTestError): - await client.call(method='GET', url='http://placeholder.url/default_timeout') - - assert patch_request == [30.0] - - -def test_short_tier_resolves_correctly_sync(patch_request: list) -> None: - """Test that `'short'` tier resolves to timeout_short value.""" - client = ImpitHttpClient(timeout_short=timedelta(seconds=5)) - - with pytest.raises(EndOfTestError): - client.call(method='GET', url='http://placeholder.url/short_tier', timeout='short') - - assert patch_request == [5.0] - - -def test_medium_tier_resolves_correctly_sync(patch_request: list) -> None: - """Test that `'medium'` tier resolves to the configured timeout value.""" - client = ImpitHttpClient(timeout_medium=timedelta(seconds=30)) - - with pytest.raises(EndOfTestError): - client.call(method='GET', url='http://placeholder.url/timeout_tier', timeout='medium') - - assert patch_request == [30.0] - - -def test_long_tier_resolves_correctly_sync(patch_request: list) -> None: - """Test that `'long'` tier resolves to timeout_long value.""" - client = ImpitHttpClient(timeout_long=timedelta(seconds=300)) - - with pytest.raises(EndOfTestError): - client.call(method='GET', url='http://placeholder.url/long_tier', timeout='long') - - assert patch_request == [300.0] - - -async def test_medium_tier_resolves_correctly_async(patch_request: list) -> None: - """Test that `'medium'` tier resolves to the configured timeout value (async).""" - client = ImpitHttpClientAsync(timeout_medium=timedelta(seconds=30)) - - with pytest.raises(EndOfTestError): - await client.call(method='GET', url='http://placeholder.url/timeout_tier', timeout='medium') - - assert patch_request == [30.0] - - -async def test_long_tier_resolves_correctly_async(patch_request: list) -> None: - """Test that `'long'` tier resolves to timeout_long value (async).""" - client = ImpitHttpClientAsync(timeout_long=timedelta(seconds=300)) +def successful_response() -> Mock: + return Mock(status_code=200) + + +def retryable_error(client: HttpClient | HttpClientAsync) -> Exception: + if isinstance(client, (ImpitHttpClient, ImpitHttpClientAsync)): + return impit.TimeoutException('timeout') + return httpx.ReadTimeout('timeout', request=httpx.Request('GET', 'https://example.com')) + + +@pytest.mark.parametrize( + ('timeout', 'expected'), + [ + pytest.param('no_timeout', None, id='no-timeout'), + pytest.param(None, 30.0, id='default-medium'), + pytest.param('short', 5.0, id='short'), + pytest.param('medium', 30.0, id='medium'), + pytest.param('long', 300.0, id='long'), + ], +) +def test_timeout_resolves_for_sync_clients( + http_client_class: type[HttpClient], + monkeypatch: pytest.MonkeyPatch, + timeout: str | None, + expected: float | None, +) -> None: + """Every synchronous client resolves timeout tiers at the shared transport boundary.""" + client = http_client_class( + timeout_short=timedelta(seconds=5), + timeout_medium=timedelta(seconds=30), + timeout_long=timedelta(seconds=300), + ) + send_request = Mock(return_value=successful_response()) + monkeypatch.setattr(client, 'send_request', send_request) + + kwargs: dict[str, Any] = {'method': 'GET', 'url': 'https://example.com'} + if timeout is not None: + kwargs['timeout'] = timeout + client.call(**kwargs) + + assert send_request.call_args.kwargs['timeout'] == expected + + +@pytest.mark.parametrize( + ('timeout', 'expected'), + [ + pytest.param('no_timeout', None, id='no-timeout'), + pytest.param(None, 30.0, id='default-medium'), + pytest.param('short', 5.0, id='short'), + pytest.param('medium', 30.0, id='medium'), + pytest.param('long', 300.0, id='long'), + ], +) +async def test_timeout_resolves_for_async_clients( + http_client_async_class: type[HttpClientAsync], + monkeypatch: pytest.MonkeyPatch, + timeout: str | None, + expected: float | None, +) -> None: + """Every asynchronous client resolves timeout tiers at the shared transport boundary.""" + client = http_client_async_class( + timeout_short=timedelta(seconds=5), + timeout_medium=timedelta(seconds=30), + timeout_long=timedelta(seconds=300), + ) + send_request = AsyncMock(return_value=successful_response()) + monkeypatch.setattr(client, 'send_request', send_request) - with pytest.raises(EndOfTestError): - await client.call(method='GET', url='http://placeholder.url/long_tier', timeout='long') + kwargs: dict[str, Any] = {'method': 'GET', 'url': 'https://example.com'} + if timeout is not None: + kwargs['timeout'] = timeout + await client.call(**kwargs) - assert patch_request == [300.0] + assert send_request.call_args.kwargs['timeout'] == expected -def test_compute_timeout_with_timedelta() -> None: - """Test _compute_timeout with a concrete timedelta doubles per attempt, capped at max.""" - client = ImpitHttpClient(timeout_max=timedelta(seconds=600)) +def test_compute_timeout_with_timedelta(http_client_class: type[HttpClient]) -> None: + """Concrete timedeltas double per attempt and are capped at the configured maximum.""" + client = http_client_class(timeout_max=timedelta(seconds=20)) assert client._compute_timeout(timedelta(seconds=5), attempt=1) == 5.0 assert client._compute_timeout(timedelta(seconds=5), attempt=2) == 10.0 assert client._compute_timeout(timedelta(seconds=5), attempt=3) == 20.0 - assert client._compute_timeout(timedelta(seconds=5), attempt=4) == 40.0 - - -def test_compute_timeout_caps_at_max() -> None: - """Test _compute_timeout caps at timeout_max.""" - client = ImpitHttpClient(timeout_max=timedelta(seconds=10)) - - assert client._compute_timeout(timedelta(seconds=5), attempt=1) == 5.0 - assert client._compute_timeout(timedelta(seconds=5), attempt=2) == 10.0 - assert client._compute_timeout(timedelta(seconds=5), attempt=3) == 10.0 # capped - - -def test_compute_timeout_no_timeout_returns_none() -> None: - """Test _compute_timeout with 'no_timeout' returns None.""" - client = ImpitHttpClient() + assert client._compute_timeout(timedelta(seconds=5), attempt=4) == 20.0 assert client._compute_timeout('no_timeout', attempt=1) is None @pytest.mark.usefixtures('fresh_logger_once') -def test_compute_timeout_explicit_timedelta_above_max_warns(caplog: LogCaptureFixture) -> None: - """Test an explicit timedelta larger than timeout_max is capped, and the cut-off is logged once.""" - client = ImpitHttpClient(timeout_max=timedelta(seconds=360)) +def test_compute_timeout_explicit_timedelta_above_max_warns( + http_client_class: type[HttpClient], caplog: LogCaptureFixture +) -> None: + """An explicit timedelta larger than timeout_max is capped, and the cut-off is logged once.""" + client = http_client_class(timeout_max=timedelta(seconds=360)) with caplog.at_level(logging.WARNING, logger=logger_name): assert client._compute_timeout(timedelta(minutes=30), attempt=1) == 360.0 @@ -176,9 +138,9 @@ def test_compute_timeout_explicit_timedelta_above_max_warns(caplog: LogCaptureFi @pytest.mark.usefixtures('fresh_logger_once') -def test_compute_timeout_tier_above_max_warns(caplog: LogCaptureFixture) -> None: - """Test a tier configured larger than timeout_max is capped, and the cut-off is logged too.""" - client = ImpitHttpClient(timeout_long=timedelta(seconds=600), timeout_max=timedelta(seconds=360)) +def test_compute_timeout_tier_above_max_warns(http_client_class: type[HttpClient], caplog: LogCaptureFixture) -> None: + """A tier configured larger than timeout_max is capped, and the cut-off is logged too.""" + client = http_client_class(timeout_long=timedelta(seconds=600), timeout_max=timedelta(seconds=360)) with caplog.at_level(logging.WARNING, logger=logger_name): assert client._compute_timeout('long', attempt=1) == 360.0 @@ -188,9 +150,11 @@ def test_compute_timeout_tier_above_max_warns(caplog: LogCaptureFixture) -> None @pytest.mark.usefixtures('fresh_logger_once') -def test_compute_timeout_within_max_does_not_warn(caplog: LogCaptureFixture) -> None: - """Test a base timeout within timeout_max is used as-is, without a warning.""" - client = ImpitHttpClient(timeout_long=timedelta(seconds=300), timeout_max=timedelta(seconds=360)) +def test_compute_timeout_within_max_does_not_warn( + http_client_class: type[HttpClient], caplog: LogCaptureFixture +) -> None: + """A base timeout within timeout_max is used as-is, without a warning.""" + client = http_client_class(timeout_long=timedelta(seconds=300), timeout_max=timedelta(seconds=360)) with caplog.at_level(logging.WARNING, logger=logger_name): assert client._compute_timeout(timedelta(seconds=120), attempt=1) == 120.0 @@ -201,103 +165,87 @@ def test_compute_timeout_within_max_does_not_warn(caplog: LogCaptureFixture) -> assert caplog.records == [] -async def test_dynamic_timeout_async_client(monkeypatch: pytest.MonkeyPatch) -> None: - """Tests timeout values for request with retriable errors. - - Values should increase with each attempt, starting from initial call value and bounded by timeout_max. - """ - should_raise_error = iter((True, True, True, False)) - call_timeout = 1 - timeout_max = 5 - expected_timeouts = [call_timeout, 2, 4, timeout_max] - retry_counter_mock = Mock() - - timeouts = [] - - async def mock_request(*_args: Any, **kwargs: Any) -> Response: - timeouts.append(kwargs.get('timeout')) - retry_counter_mock() - should_raise = next(should_raise_error) - if should_raise: - raise TimeoutException +def test_dynamic_timeout_sync_client(http_client_class: type[HttpClient], monkeypatch: pytest.MonkeyPatch) -> None: + """Synchronous clients increase timeout values after retryable transport errors.""" + client = http_client_class( + timeout_short=timedelta(seconds=1), + timeout_max=timedelta(seconds=5), + min_delay_between_retries=timedelta(0), + ) + timeouts: list[float | None] = [] - return Response(status_code=200) + def send_request(*_args: Any, **kwargs: Any) -> Mock: + timeouts.append(kwargs['timeout']) + if len(timeouts) < 4: + raise retryable_error(client) + return successful_response() - monkeypatch.setattr('impit.AsyncClient.request', mock_request) + monkeypatch.setattr(client, 'send_request', send_request) - response = await ImpitHttpClientAsync( - timeout_short=timedelta(seconds=call_timeout), - timeout_max=timedelta(seconds=timeout_max), - ).call(method='GET', url='http://placeholder.url/async_timeout', timeout=timedelta(seconds=call_timeout)) + response = client.call(method='GET', url='https://example.com', timeout=timedelta(seconds=1)) - # Check that the retry counter was called the expected number of times - # (4 times: 3 retries + 1 final successful call) - assert retry_counter_mock.call_count == 4 - assert timeouts == expected_timeouts - # Check that the response is successful + assert timeouts == [1.0, 2.0, 4.0, 5.0] assert response.status_code == 200 -async def test_retry_on_http_error_async_client(monkeypatch: pytest.MonkeyPatch) -> None: - """Tests that bare impit.HTTPError (e.g. body decode errors) are retried. - - This reproduces the scenario where the HTTP response body is truncated mid-stream - (e.g. "unexpected EOF during chunk size line"), which impit raises as a generic HTTPError. - """ - should_raise_error = iter((True, True, False)) - retry_counter_mock = Mock() - - async def mock_request(*_args: Any, **_kwargs: Any) -> Response: - retry_counter_mock() - should_raise = next(should_raise_error) - if should_raise: - raise HTTPError('The internal HTTP library has thrown an error: unexpected EOF during chunk size line') - - return Response(status_code=200) - - monkeypatch.setattr('impit.AsyncClient.request', mock_request) - - response = await ImpitHttpClientAsync(timeout_short=timedelta(seconds=5)).call( - method='GET', url='http://placeholder.url/http_error' +async def test_dynamic_timeout_async_client( + http_client_async_class: type[HttpClientAsync], monkeypatch: pytest.MonkeyPatch +) -> None: + """Asynchronous clients increase timeout values after retryable transport errors.""" + client = http_client_async_class( + timeout_short=timedelta(seconds=1), + timeout_max=timedelta(seconds=5), + min_delay_between_retries=timedelta(0), ) + timeouts: list[float | None] = [] - # 3 attempts: 2 failures + 1 success - assert retry_counter_mock.call_count == 3 - assert response.status_code == 200 - - -def test_dynamic_timeout_sync_client(monkeypatch: pytest.MonkeyPatch) -> None: - """Tests timeout values for request with retriable errors. - - Values should increase with each attempt, starting from initial call value and bounded by timeout_max. - """ - should_raise_error = iter((True, True, True, False)) - call_timeout = 1 - timeout_max = 5 - expected_timeouts = [call_timeout, 2, 4, timeout_max] - retry_counter_mock = Mock() - - timeouts = [] + async def send_request(*_args: Any, **kwargs: Any) -> Mock: + timeouts.append(kwargs['timeout']) + if len(timeouts) < 4: + raise retryable_error(client) + return successful_response() - def mock_request(*_args: Any, **kwargs: Any) -> Response: - timeouts.append(kwargs.get('timeout')) - retry_counter_mock() - should_raise = next(should_raise_error) - if should_raise: - raise TimeoutException + monkeypatch.setattr(client, 'send_request', send_request) - return Response(status_code=200) + response = await client.call(method='GET', url='https://example.com', timeout=timedelta(seconds=1)) - monkeypatch.setattr('impit.Client.request', mock_request) + assert timeouts == [1.0, 2.0, 4.0, 5.0] + assert response.status_code == 200 - response = ImpitHttpClient( - timeout_short=timedelta(seconds=call_timeout), - timeout_max=timedelta(seconds=timeout_max), - ).call(method='GET', url='http://placeholder.url/sync_timeout', timeout=timedelta(seconds=call_timeout)) - # Check that the retry counter was called the expected number of times - # (4 times: 3 retries + 1 final successful call) - assert retry_counter_mock.call_count == 4 - assert timeouts == expected_timeouts - # Check that the response is successful - assert response.status_code == 200 +def test_no_timeout_mapping_for_sync_adapters(monkeypatch: pytest.MonkeyPatch) -> None: + """Each synchronous adapter maps no-timeout to its underlying library semantics.""" + impit_client = ImpitHttpClient() + impit_client._impit_client = Mock(request=Mock(return_value=successful_response())) + impit_client.send_request( + method='GET', url='https://example.com', headers={}, content=None, timeout=None, stream=False + ) + assert impit_client._impit_client.request.call_args.kwargs['timeout'] == 86_400 + + # Only the transport call is stubbed, so the real `build_request` decides what `None` means to HTTPX. + with HttpxHttpClient() as httpx_client: + send = Mock(return_value=successful_response()) + monkeypatch.setattr(httpx_client._httpx_client, 'send', send) + httpx_client.send_request( + method='GET', url='https://example.com', headers={}, content=None, timeout=None, stream=False + ) + assert send.call_args.args[0].extensions['timeout'] == UNSET_HTTPX_TIMEOUT + + +async def test_no_timeout_mapping_for_async_adapters(monkeypatch: pytest.MonkeyPatch) -> None: + """Each asynchronous adapter maps no-timeout to its underlying library semantics.""" + impit_client = ImpitHttpClientAsync() + impit_client._impit_async_client = Mock(request=AsyncMock(return_value=successful_response())) + await impit_client.send_request( + method='GET', url='https://example.com', headers={}, content=None, timeout=None, stream=False + ) + assert impit_client._impit_async_client.request.call_args.kwargs['timeout'] == 86_400 + + # Only the transport call is stubbed, so the real `build_request` decides what `None` means to HTTPX. + async with HttpxHttpClientAsync() as httpx_client: + send = AsyncMock(return_value=successful_response()) + monkeypatch.setattr(httpx_client._httpx_async_client, 'send', send) + await httpx_client.send_request( + method='GET', url='https://example.com', headers={}, content=None, timeout=None, stream=False + ) + assert send.call_args.args[0].extensions['timeout'] == UNSET_HTTPX_TIMEOUT diff --git a/tests/unit/test_http_clients.py b/tests/unit/test_http_clients.py index ef5f1ac0..8a3aedd8 100644 --- a/tests/unit/test_http_clients.py +++ b/tests/unit/test_http_clients.py @@ -11,14 +11,22 @@ from unittest.mock import AsyncMock, Mock import brotli +import httpx import impit import pytest from apify_client._consts import MIN_COMPRESSION_SIZE from apify_client._statistics import ClientStatistics from apify_client.errors import InvalidResponseBodyError -from apify_client.http_clients import HttpClient, HttpClientAsync, HttpResponse, ImpitHttpClient, ImpitHttpClientAsync -from apify_client.http_clients._impit import _is_retryable_error +from apify_client.http_clients import ( + HttpClient, + HttpClientAsync, + HttpResponse, + HttpxHttpClient, + HttpxHttpClientAsync, + ImpitHttpClient, + ImpitHttpClientAsync, +) from apify_client.http_compressors._base import HttpCompressor from apify_client.http_compressors._brotli import BrotliHttpCompressor from apify_client.http_compressors._gzip import GzipHttpCompressor @@ -30,18 +38,41 @@ from apify_client.types import JsonSerializable -class _ConcreteHttpClient(HttpClient): - """Minimal concrete HttpClient for testing base class helpers.""" +class ConcreteHttpClient(HttpClient): + """Minimal concrete HttpClient for testing base class helpers, relying on the hook defaults.""" + + +class CallOnlyHttpClient(HttpClient): + """Custom client written against the pre-hook contract, overriding `call` and nothing else.""" + + def call(self, **kwargs: Any) -> HttpResponse: + _ = kwargs + return Mock(status_code=200) + + +class CallOnlyHttpClientAsync(HttpClientAsync): + """Asynchronous counterpart of `CallOnlyHttpClient`.""" + + async def call(self, **kwargs: Any) -> HttpResponse: + _ = kwargs + return Mock(status_code=200) - def call(self, *, method: str, url: str, **kwargs: Any) -> HttpResponse: - raise NotImplementedError +def test_call_only_http_client_keeps_working() -> None: + """A client that only overrides `call` still instantiates and inherits the hook defaults.""" + with CallOnlyHttpClient() as client: + assert client.call(method='GET', url='https://example.com').status_code == 200 + assert client.is_timeout_error(TimeoutError('test')) + assert not client.is_retryable_transport_error(TimeoutError('test')) -class _ConcreteHttpClientAsync(HttpClientAsync): - """Minimal concrete HttpClientAsync for testing base class helpers.""" - async def call(self, *, method: str, url: str, **kwargs: Any) -> HttpResponse: - raise NotImplementedError +async def test_call_only_http_client_async_keeps_working() -> None: + """The asynchronous base is equally tolerant of a client that only overrides `call`.""" + async with CallOnlyHttpClientAsync() as client: + response = await client.call(method='GET', url='https://example.com') + assert response.status_code == 200 + assert client.is_timeout_error(TimeoutError('test')) + assert not client.is_retryable_transport_error(TimeoutError('test')) def test_retry_with_exp_backoff() -> None: @@ -73,7 +104,7 @@ def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any: # Returns the correct result after the correct time (should take 100 + 200 + 400 + 800 = 1500 ms) start = time.time() - result = ImpitHttpClient._retry_with_exp_backoff( + result = HttpClient._retry_with_exp_backoff( returns_on_fifth_attempt, backoff_base=timedelta(milliseconds=100), backoff_factor=2, random_factor=0 ) elapsed_time_seconds = time.time() - start @@ -85,7 +116,7 @@ def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any: # Stops retrying when failed for max_retries times attempt_counter = 0 with pytest.raises(RetryableError): - ImpitHttpClient._retry_with_exp_backoff( + HttpClient._retry_with_exp_backoff( returns_on_fifth_attempt, max_retries=3, backoff_base=timedelta(milliseconds=1) ) assert attempt_counter == 4 @@ -93,7 +124,7 @@ def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any: # Bails when the bail function is called attempt_counter = 0 with pytest.raises(NonRetryableError): - ImpitHttpClient._retry_with_exp_backoff(bails_on_third_attempt, backoff_base=timedelta(milliseconds=1)) + HttpClient._retry_with_exp_backoff(bails_on_third_attempt, backoff_base=timedelta(milliseconds=1)) assert attempt_counter == 3 @@ -126,7 +157,7 @@ async def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any: # Returns the correct result after the correct time (should take 100 + 200 + 400 + 800 = 1500 ms) start = time.time() - result = await ImpitHttpClientAsync._retry_with_exp_backoff( + result = await HttpClientAsync._retry_with_exp_backoff( returns_on_fifth_attempt, backoff_base=timedelta(milliseconds=100), backoff_factor=2, random_factor=0 ) elapsed_time_seconds = time.time() - start @@ -138,7 +169,7 @@ async def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any: # Stops retrying when failed for max_retries times attempt_counter = 0 with pytest.raises(RetryableError): - await ImpitHttpClientAsync._retry_with_exp_backoff( + await HttpClientAsync._retry_with_exp_backoff( returns_on_fifth_attempt, max_retries=3, backoff_base=timedelta(milliseconds=1) ) assert attempt_counter == 4 @@ -146,9 +177,7 @@ async def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any: # Bails when the bail function is called attempt_counter = 0 with pytest.raises(NonRetryableError): - await ImpitHttpClientAsync._retry_with_exp_backoff( - bails_on_third_attempt, backoff_base=timedelta(milliseconds=1) - ) + await HttpClientAsync._retry_with_exp_backoff(bails_on_third_attempt, backoff_base=timedelta(milliseconds=1)) assert attempt_counter == 3 @@ -156,7 +185,7 @@ def test_base_http_client_initialization() -> None: """Test HttpClient initialization with various configurations.""" statistics = ClientStatistics() - client = _ConcreteHttpClient( + client = ConcreteHttpClient( token='test_token', timeout_short=timedelta(seconds=30), max_retries=5, @@ -171,13 +200,13 @@ def test_base_http_client_initialization() -> None: assert client._headers['Authorization'] == 'Bearer test_token' # Test without statistics (should create default) - client2 = _ConcreteHttpClient(token='test_token') + client2 = ConcreteHttpClient(token='test_token') assert isinstance(client2._statistics, ClientStatistics) def test_http_client_init_headers_override_defaults_case_insensitively() -> None: """Constructor headers replace same-named default headers even when their casings differ.""" - client = _ConcreteHttpClient(token='default_token', headers={'authorization': 'Bearer custom'}) + client = ConcreteHttpClient(token='default_token', headers={'authorization': 'Bearer custom'}) auth_headers = {key: value for key, value in client._headers.items() if key.lower() == 'authorization'} assert auth_headers == {'authorization': 'Bearer custom'} @@ -187,14 +216,14 @@ def test_http_client_init_workflow_key_header(monkeypatch: pytest.MonkeyPatch) - """The X-Apify-Workflow-Key default header is set from the APIFY_WORKFLOW_KEY env var.""" monkeypatch.setenv('APIFY_WORKFLOW_KEY', 'workflow_key_123') - client = _ConcreteHttpClient() + client = ConcreteHttpClient() assert client._headers['X-Apify-Workflow-Key'] == 'workflow_key_123' def test_set_default_authorization_sets_token_when_missing() -> None: """set_default_authorization sets the Bearer token when no authorization header is configured.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() client.set_default_authorization('test_token') @@ -223,21 +252,37 @@ def test_merge_headers( def test_http_client_creates_sync_impit_client() -> None: - """Test that ImpitHttpClient creates sync impit client correctly.""" + """The synchronous adapter creates the underlying Impit client and closes it through the close hook.""" client = ImpitHttpClient(token='test_token_123') - # Check that sync impit client is created - assert client._impit_client is not None assert isinstance(client._impit_client, impit.Client) + client.close() -def test_http_client_async_creates_async_impit_client() -> None: - """Test that ImpitHttpClientAsync creates async impit client correctly.""" +async def test_http_client_async_creates_async_impit_client() -> None: + """The asynchronous adapter creates the underlying Impit client and closes it through the close hook.""" client = ImpitHttpClientAsync(token='test_token_123') - # Check that async impit client is created - assert client._impit_async_client is not None assert isinstance(client._impit_async_client, impit.AsyncClient) + await client.aclose() + + +def test_http_client_creates_sync_httpx_client() -> None: + """The synchronous HTTPX adapter creates the underlying HTTPX client, and the close hook closes its pool.""" + client = HttpxHttpClient(token='test_token_123') + + assert isinstance(client._httpx_client, httpx.Client) + client.close() + assert client._httpx_client.is_closed + + +async def test_http_client_async_creates_async_httpx_client() -> None: + """The asynchronous HTTPX adapter creates the underlying HTTPX client, and the close hook closes its pool.""" + client = HttpxHttpClientAsync(token='test_token_123') + + assert isinstance(client._httpx_async_client, httpx.AsyncClient) + await client.aclose() + assert client._httpx_async_client.is_closed def test_parse_params_none() -> None: @@ -324,9 +369,10 @@ def test_parse_params_mixed() -> None: pytest.param(impit.ProxyError('proxy error'), id='ProxyError'), ], ) -def test_is_retryable_error(exc: Exception) -> None: - """A transient transport failure is retried.""" - assert _is_retryable_error(exc) +def test_is_retryable_transport_error(exc: Exception) -> None: + """A transient transport failure is classified as retryable.""" + with ImpitHttpClient() as client: + assert client.is_retryable_transport_error(exc) @pytest.mark.parametrize( @@ -345,9 +391,90 @@ def test_is_retryable_error(exc: Exception) -> None: pytest.param(Exception('generic exception'), id='Exception'), ], ) -def test_is_not_retryable_error(exc: Exception) -> None: +def test_is_not_retryable_transport_error(exc: Exception) -> None: """A transport failure a retry cannot fix, and anything outside Impit's hierarchy, is not retried.""" - assert not _is_retryable_error(exc) + with ImpitHttpClient() as client: + assert not client.is_retryable_transport_error(exc) + + +def test_sync_http_client_classifies_timeout_errors() -> None: + """The built-in synchronous client exposes transport-neutral timeout classification.""" + with ImpitHttpClient() as client: + assert client.is_timeout_error(TimeoutError('test')) + assert client.is_timeout_error(impit.TimeoutException('test')) + assert not client.is_timeout_error(ValueError('test')) + + +async def test_async_http_client_classifies_timeout_errors() -> None: + """The built-in asynchronous client exposes transport-neutral timeout classification.""" + async with ImpitHttpClientAsync() as client: + assert client.is_timeout_error(TimeoutError('test')) + assert client.is_timeout_error(impit.TimeoutException('test')) + assert not client.is_timeout_error(ValueError('test')) + + +@pytest.mark.parametrize( + 'exc', + [ + # Even the generic base class is transient: HTTPX subclasses it for every failure mode, so an + # unclassified failure is safer to retry. + pytest.param(httpx.HTTPError('unclassified failure'), id='bare HTTPError'), + pytest.param(httpx.TimeoutException('timeout'), id='TimeoutException'), + pytest.param(httpx.NetworkError('network error'), id='NetworkError'), + pytest.param(httpx.RemoteProtocolError('remote protocol error'), id='RemoteProtocolError'), + pytest.param(httpx.DecodingError('decoding error'), id='DecodingError'), + # One `ProxyError` covers both a proxy rejecting the CONNECT tunnel and a 407, so a transient case cannot + # be told from a permanent one - retrying is the safer default. + pytest.param(httpx.ProxyError('proxy error'), id='ProxyError'), + ], +) +def test_httpx_is_retryable_transport_error(exc: Exception) -> None: + """A transient HTTPX transport failure is classified as retryable.""" + with HttpxHttpClient() as client: + assert client.is_retryable_transport_error(exc) + + +@pytest.mark.parametrize( + 'exc', + [ + pytest.param(httpx.LocalProtocolError('invalid header value'), id='LocalProtocolError'), + pytest.param(httpx.UnsupportedProtocol('unsupported scheme'), id='UnsupportedProtocol'), + pytest.param(httpx.TooManyRedirects('too many redirects'), id='TooManyRedirects'), + pytest.param( + httpx.HTTPStatusError( + 'status error', + request=httpx.Request('GET', 'https://example.com'), + response=httpx.Response(500), + ), + id='HTTPStatusError', + ), + # HTTPX reports a bad URL outside the `httpx.HTTPError` tree entirely. + pytest.param(httpx.InvalidURL('unsupported scheme'), id='InvalidURL'), + pytest.param(ValueError('value error'), id='ValueError'), + pytest.param(RuntimeError('runtime error'), id='RuntimeError'), + pytest.param(Exception('generic exception'), id='Exception'), + ], +) +def test_httpx_is_not_retryable_transport_error(exc: Exception) -> None: + """A transport failure a retry cannot fix, and anything outside HTTPX's hierarchy, is not retried.""" + with HttpxHttpClient() as client: + assert not client.is_retryable_transport_error(exc) + + +def test_sync_httpx_client_classifies_timeout_errors() -> None: + """The built-in synchronous HTTPX client exposes transport-neutral timeout classification.""" + with HttpxHttpClient() as client: + assert client.is_timeout_error(TimeoutError('test')) + assert client.is_timeout_error(httpx.TimeoutException('test')) + assert not client.is_timeout_error(ValueError('test')) + + +async def test_async_httpx_client_classifies_timeout_errors() -> None: + """The built-in asynchronous HTTPX client exposes transport-neutral timeout classification.""" + async with HttpxHttpClientAsync() as client: + assert client.is_timeout_error(TimeoutError('test')) + assert client.is_timeout_error(httpx.TimeoutException('test')) + assert not client.is_timeout_error(ValueError('test')) def test_permanent_transport_error_is_not_retried() -> None: @@ -387,6 +514,31 @@ def test_transient_transport_error_is_retried() -> None: assert request.call_count == 3 +def test_httpx_permanent_transport_error_is_not_retried(monkeypatch: pytest.MonkeyPatch) -> None: + """The HTTPX adapter feeds the same fail-fast classification into the shared pipeline.""" + with HttpxHttpClient(token='test_token', min_delay_between_retries=timedelta(0)) as client: + send_request = Mock(side_effect=httpx.UnsupportedProtocol('unsupported scheme')) + monkeypatch.setattr(client, 'send_request', send_request) + + with pytest.raises(httpx.UnsupportedProtocol): + client.call(method='GET', url='https://api.test.com/endpoint') + + send_request.assert_called_once() + + +def test_httpx_transient_transport_error_is_retried(monkeypatch: pytest.MonkeyPatch) -> None: + """The HTTPX adapter keeps a transient transport failure inside the shared retry loop.""" + with HttpxHttpClient(token='test_token', max_retries=2, min_delay_between_retries=timedelta(0)) as client: + send_request = Mock(side_effect=httpx.TimeoutException('timeout')) + monkeypatch.setattr(client, 'send_request', send_request) + + with pytest.raises(httpx.TimeoutException): + client.call(method='GET', url='https://api.test.com/endpoint') + + # `max_retries` attempts inside the backoff loop, plus the final one it makes after the last delay. + assert send_request.call_count == 3 + + def test_error_response_read_failure_is_retried_and_closed() -> None: """A failure while buffering a streamed error body is retried like a failed send, and the response is closed.""" client = ImpitHttpClient(token='test_token', max_retries=1, min_delay_between_retries=timedelta(0)) @@ -476,7 +628,7 @@ def compressor_case(request: pytest.FixtureRequest) -> tuple: def test_prepare_request_call_basic() -> None: """Test _prepare_request_call returns the client default headers when no per-request values are given.""" - client = _ConcreteHttpClient(token='test_token') + client = ConcreteHttpClient(token='test_token') headers, params, data = client._prepare_request_call() assert headers == client._headers @@ -488,7 +640,7 @@ def test_prepare_request_call_basic() -> None: def test_prepare_request_call_with_json() -> None: """A small JSON body is serialized and typed, but sent uncompressed and without a `Content-Encoding`.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() json_data = {'key': 'value', 'number': 42} headers, _params, data = client._prepare_request_call(json=json_data) @@ -510,7 +662,7 @@ def test_prepare_request_call_with_json() -> None: ) def test_prepare_request_call_with_falsy_json(json_body: JsonSerializable, expected: bytes) -> None: """A falsy but valid JSON body is still serialized and sent, rather than treated as no body at all.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() headers, _params, data = client._prepare_request_call(json=json_body) @@ -528,7 +680,7 @@ def test_prepare_request_call_with_falsy_json(json_body: JsonSerializable, expec ) def test_prepare_request_call_with_data(data: str | bytes | bytearray) -> None: """A raw body of any accepted type is normalized to bytes.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() _headers, _params, prepared = client._prepare_request_call(data=data) @@ -550,7 +702,7 @@ def test_prepare_request_call_compresses_body_at_or_above_threshold( ) -> None: """A raw body of at least `MIN_COMPRESSION_SIZE` bytes is compressed and labeled with its encoding.""" compressor, content_encoding, decompress = compressor_case - client = _ConcreteHttpClient(http_compressor=compressor) + client = ConcreteHttpClient(http_compressor=compressor) body = b'x' * body_size headers, _params, data = client._prepare_request_call(data=body) @@ -570,7 +722,7 @@ def test_prepare_request_call_compresses_body_at_or_above_threshold( def test_prepare_request_call_skips_compression_below_threshold(compressor_case: tuple, body_size: int) -> None: """A raw body under `MIN_COMPRESSION_SIZE` is sent verbatim with no `Content-Encoding`, whichever compressor.""" compressor, _content_encoding, _decompress = compressor_case - client = _ConcreteHttpClient(http_compressor=compressor) + client = ConcreteHttpClient(http_compressor=compressor) body = b'x' * body_size headers, _params, data = client._prepare_request_call(data=body) @@ -582,7 +734,7 @@ def test_prepare_request_call_skips_compression_below_threshold(compressor_case: def test_prepare_request_call_compresses_bytearray_data(compressor_case: tuple) -> None: """A `bytearray` body above the threshold is compressed without error (regression: needs bytes conversion).""" compressor, content_encoding, decompress = compressor_case - client = _ConcreteHttpClient(http_compressor=compressor) + client = ConcreteHttpClient(http_compressor=compressor) body = bytearray(b'test bytearray' * 128) headers, _params, data = client._prepare_request_call(data=body) @@ -594,7 +746,7 @@ def test_prepare_request_call_compresses_bytearray_data(compressor_case: tuple) def test_prepare_request_call_compresses_json_above_threshold(compressor_case: tuple) -> None: """A JSON body that serializes to at least `MIN_COMPRESSION_SIZE` bytes is compressed.""" compressor, content_encoding, decompress = compressor_case - client = _ConcreteHttpClient(http_compressor=compressor) + client = ConcreteHttpClient(http_compressor=compressor) json_data = {'key': 'x' * MIN_COMPRESSION_SIZE} headers, _params, data = client._prepare_request_call(json=json_data) @@ -607,7 +759,7 @@ def test_prepare_request_call_compresses_json_above_threshold(compressor_case: t def test_prepare_request_call_measures_threshold_in_bytes_not_characters(compressor_case: tuple) -> None: """A `str` body under the threshold in characters but over it in UTF-8 bytes is still compressed.""" compressor, content_encoding, decompress = compressor_case - client = _ConcreteHttpClient(http_compressor=compressor) + client = ConcreteHttpClient(http_compressor=compressor) # U+00E9 encodes to 2 bytes, so this body is under the threshold in characters but over it in bytes. body = '\u00e9' * (MIN_COMPRESSION_SIZE // 2 + 1) @@ -628,7 +780,7 @@ def test_prepare_request_call_measures_threshold_in_bytes_not_characters(compres ) def test_is_body_worth_compressing(data: Any) -> None: """The gate reports a body `_prepare_request_call` would compress, judging a `str` by its encoded bytes.""" - assert _ConcreteHttpClient._is_body_worth_compressing(data) + assert ConcreteHttpClient._is_body_worth_compressing(data) @pytest.mark.parametrize( @@ -643,7 +795,7 @@ def test_is_body_worth_compressing(data: Any) -> None: ) def test_is_body_not_worth_compressing(data: Any) -> None: """A body below the threshold, or of a type the client sends as it is, needs no worker-thread hop.""" - assert not _ConcreteHttpClient._is_body_worth_compressing(data) + assert not ConcreteHttpClient._is_body_worth_compressing(data) @pytest.mark.parametrize( @@ -656,7 +808,7 @@ def test_is_body_not_worth_compressing(data: Any) -> None: ) def test_prepare_request_call_skips_compression_for_already_compressed_content(content_type: str) -> None: """An already-compressed body is sent verbatim, carries no `Content-Encoding`, and keeps every other header.""" - client = _ConcreteHttpClient(token='test_token', http_compressor=GzipHttpCompressor()) + client = ConcreteHttpClient(token='test_token', http_compressor=GzipHttpCompressor()) # Above the size threshold, so the content type is what skips compression here. payload = b'\x89PNG' + b'\xff' * MIN_COMPRESSION_SIZE @@ -674,7 +826,7 @@ def test_prepare_request_call_skips_compression_for_already_compressed_content(c def test_prepare_request_call_keeps_caller_content_encoding_for_a_file_like_body() -> None: """A file-like body skips compression entirely, and its `Content-Encoding` reaches the transport untouched.""" - client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor()) + client = ConcreteHttpClient(http_compressor=GzipHttpCompressor()) stream = BytesIO(gzip.compress(b'raw payload')) headers, _params, data = client._prepare_request_call( @@ -700,7 +852,7 @@ def test_prepare_request_call_compresses_exceptions_to_compressed_prefixes( ) -> None: """Types that are text or raw are compressed even when they sit under an already-compressed prefix.""" compressor, content_encoding, decompress = compressor_case - client = _ConcreteHttpClient(http_compressor=compressor) + client = ConcreteHttpClient(http_compressor=compressor) payload = b'x' * MIN_COMPRESSION_SIZE headers, _params, data = client._prepare_request_call( @@ -715,7 +867,7 @@ def test_prepare_request_call_compresses_exceptions_to_compressed_prefixes( def test_prepare_request_call_json_and_data_error() -> None: """Test _prepare_request_call raises error when both json and data are provided.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() with pytest.raises(ValueError, match='Cannot pass both "json" and "data" parameters'): client._prepare_request_call(json={'key': 'value'}, data='string') @@ -723,7 +875,7 @@ def test_prepare_request_call_json_and_data_error() -> None: def test_prepare_request_call_with_params() -> None: """Test _prepare_request_call parses params correctly.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() _headers, params, _data = client._prepare_request_call(params={'limit': 10, 'flag': True}) @@ -736,7 +888,7 @@ def test_prepare_request_call_does_not_mutate_caller_headers() -> None: A caller that reuses a shared headers dict across calls must not see stale `Content-Type`/`Content-Encoding` headers leak in from a prior JSON/body call. """ - client = _ConcreteHttpClient() + client = ConcreteHttpClient() caller_headers = {'x-trace-id': 'abc-123'} original = dict(caller_headers) @@ -750,7 +902,7 @@ def test_prepare_request_call_does_not_mutate_caller_headers() -> None: def test_prepare_request_call_per_request_headers_override_defaults_case_insensitively() -> None: """A per-request header replaces a same-named default header even when their casings differ.""" - client = _ConcreteHttpClient(token='default_token') + client = ConcreteHttpClient(token='default_token') headers, _params, _data = client._prepare_request_call(headers={'authorization': 'Bearer per-request'}) @@ -760,7 +912,7 @@ def test_prepare_request_call_per_request_headers_override_defaults_case_insensi def test_prepare_request_call_json_keeps_caller_content_type() -> None: """A caller-supplied content type is not overwritten by the JSON default, regardless of casing.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() headers, _params, _data = client._prepare_request_call( headers={'content-type': 'application/json; charset=utf-8'}, @@ -791,7 +943,7 @@ def test_prepare_request_call_json_keeps_caller_content_type() -> None: ) def test_prepare_request_call_keeps_caller_content_encoding(caller_headers: dict[str, str], body: bytes) -> None: """A caller-supplied `Content-Encoding` marks the body as pre-encoded, so it goes out untouched and labeled.""" - client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor()) + client = ConcreteHttpClient(http_compressor=GzipHttpCompressor()) headers, _params, data = client._prepare_request_call(headers=caller_headers, data=body) @@ -802,7 +954,7 @@ def test_prepare_request_call_keeps_caller_content_encoding(caller_headers: dict def test_prepare_request_call_keeps_client_wide_content_encoding() -> None: """A `Content-Encoding` configured on the client counts as caller-supplied on every request it sends.""" - client = _ConcreteHttpClient(headers={'Content-Encoding': 'identity'}, http_compressor=GzipHttpCompressor()) + client = ConcreteHttpClient(headers={'Content-Encoding': 'identity'}, http_compressor=GzipHttpCompressor()) body = b'x' * MIN_COMPRESSION_SIZE headers, _params, data = client._prepare_request_call(data=body) @@ -813,7 +965,7 @@ def test_prepare_request_call_keeps_client_wide_content_encoding() -> None: def test_build_url_with_params_none() -> None: """Test _build_url_with_params with None params.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() url = client._build_url_with_params('https://api.test.com/endpoint') assert url == 'https://api.test.com/endpoint' @@ -821,7 +973,7 @@ def test_build_url_with_params_none() -> None: def test_build_url_with_params_simple() -> None: """Test _build_url_with_params with simple params.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() url = client._build_url_with_params('https://api.test.com/endpoint', params={'key': 'value', 'limit': 10}) assert 'key=value' in url @@ -831,7 +983,7 @@ def test_build_url_with_params_simple() -> None: def test_build_url_with_params_list() -> None: """Test _build_url_with_params with list values.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() url = client._build_url_with_params('https://api.test.com/endpoint', params={'tags': ['tag1', 'tag2', 'tag3']}) assert 'tags=tag1' in url @@ -841,7 +993,7 @@ def test_build_url_with_params_list() -> None: def test_build_url_with_params_mixed() -> None: """Test _build_url_with_params with mixed param types.""" - client = _ConcreteHttpClient() + client = ConcreteHttpClient() url = client._build_url_with_params( 'https://api.test.com/endpoint', params={'limit': 10, 'tags': ['a', 'b'], 'name': 'test'} @@ -865,11 +1017,13 @@ def compress(self, data: bytes) -> bytes: return gzip.compress(data) -async def test_async_call_compresses_request_body_off_the_event_loop() -> None: +async def test_async_call_compresses_request_body_off_the_event_loop( + http_client_async_class: type[HttpClientAsync], monkeypatch: pytest.MonkeyPatch +) -> None: """Body serialization and compression must run in a worker thread, not block the event loop.""" compressor = _ThreadRecordingCompressor() - client = ImpitHttpClientAsync(token='test_token', http_compressor=compressor) - client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200))) + client = http_client_async_class(token='test_token', http_compressor=compressor) + monkeypatch.setattr(client, 'send_request', AsyncMock(return_value=Mock(status_code=200))) await client.call( method='POST', @@ -881,11 +1035,13 @@ async def test_async_call_compresses_request_body_off_the_event_loop() -> None: assert compressor.compress_thread_id != threading.get_ident() -async def test_async_call_compresses_a_multibyte_str_body_off_the_event_loop() -> None: +async def test_async_call_compresses_a_multibyte_str_body_off_the_event_loop( + http_client_async_class: type[HttpClientAsync], monkeypatch: pytest.MonkeyPatch +) -> None: """A `str` body over the threshold only once encoded is still compressed, so it must be offloaded.""" compressor = _ThreadRecordingCompressor() - client = ImpitHttpClientAsync(token='test_token', http_compressor=compressor) - client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200))) + client = http_client_async_class(token='test_token', http_compressor=compressor) + monkeypatch.setattr(client, 'send_request', AsyncMock(return_value=Mock(status_code=200))) await client.call( method='PUT', @@ -904,10 +1060,12 @@ def _to_thread_spy(monkeypatch: pytest.MonkeyPatch) -> Mock: return spy -async def test_async_call_skips_thread_offload_without_a_body(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_async_call_skips_thread_offload_without_a_body( + http_client_async_class: type[HttpClientAsync], monkeypatch: pytest.MonkeyPatch +) -> None: """A bodyless request has nothing to compress, so it must not pay the worker-thread hop.""" - client = ImpitHttpClientAsync(token='test_token') - client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200))) + client = http_client_async_class(token='test_token') + monkeypatch.setattr(client, 'send_request', AsyncMock(return_value=Mock(status_code=200))) spy = _to_thread_spy(monkeypatch) await client.call(method='GET', url='https://api.test.com/endpoint') @@ -916,11 +1074,12 @@ async def test_async_call_skips_thread_offload_without_a_body(monkeypatch: pytes async def test_async_call_skips_thread_offload_for_a_body_below_the_threshold( + http_client_async_class: type[HttpClientAsync], monkeypatch: pytest.MonkeyPatch, ) -> None: """A raw body too small to be compressed must not pay the worker-thread hop either.""" - client = ImpitHttpClientAsync(token='test_token') - client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200))) + client = http_client_async_class(token='test_token') + monkeypatch.setattr(client, 'send_request', AsyncMock(return_value=Mock(status_code=200))) spy = _to_thread_spy(monkeypatch) await client.call(method='PUT', url='https://api.test.com/endpoint', data=b'x' * (MIN_COMPRESSION_SIZE - 1)) @@ -928,10 +1087,12 @@ async def test_async_call_skips_thread_offload_for_a_body_below_the_threshold( spy.assert_not_called() -async def test_async_call_offloads_a_body_at_the_threshold(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_async_call_offloads_a_body_at_the_threshold( + http_client_async_class: type[HttpClientAsync], monkeypatch: pytest.MonkeyPatch +) -> None: """A raw body large enough to be compressed is prepared in a worker thread.""" - client = ImpitHttpClientAsync(token='test_token') - client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200))) + client = http_client_async_class(token='test_token') + monkeypatch.setattr(client, 'send_request', AsyncMock(return_value=Mock(status_code=200))) spy = _to_thread_spy(monkeypatch) await client.call(method='PUT', url='https://api.test.com/endpoint', data=b'x' * MIN_COMPRESSION_SIZE) @@ -940,11 +1101,12 @@ async def test_async_call_offloads_a_body_at_the_threshold(monkeypatch: pytest.M async def test_async_call_skips_thread_offload_for_a_body_it_cannot_compress( + http_client_async_class: type[HttpClientAsync], monkeypatch: pytest.MonkeyPatch, ) -> None: """A body of a type the client passes through needs no hop, and deciding that must not need its length.""" - client = ImpitHttpClientAsync(token='test_token') - client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200))) + client = http_client_async_class(token='test_token') + monkeypatch.setattr(client, 'send_request', AsyncMock(return_value=Mock(status_code=200))) spy = _to_thread_spy(monkeypatch) # `encode_key_value_store_record_value` passes file-like bodies through, so the gate cannot assume a length. diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 76db2240..4c2f2271 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -24,6 +24,7 @@ from pytest_httpserver import HTTPServer from apify_client._literals import ActorJobStatus + from apify_client.http_clients import HttpClient, HttpClientAsync _MOCKED_RUN_ID = 'mocked_run_id' _MOCKED_ACTOR_NAME = 'mocked_actor_name' @@ -873,11 +874,10 @@ def generate_logs() -> Iterator[bytes]: def test_streamed_log_sync_does_not_leak_exception_on_stream_timeout( caplog: LogCaptureFixture, httpserver: HTTPServer, + http_client_class: type[HttpClient], monkeypatch: pytest.MonkeyPatch, ) -> None: - """The streaming thread ends quietly when the log-stream request hits its total timeout (regression #1040).""" - # impit enforces a whole-request timeout, so a still-running Actor whose run outlives the timeout makes - # `iter_bytes()` raise `impit.TimeoutException`. Shorten the timeout to trigger this quickly. + """The streaming thread ends quietly when either transport times out while reading the log stream.""" monkeypatch.setattr(StreamedLog, '_stream_timeout', timedelta(seconds=1)) release_server = threading.Event() @@ -897,7 +897,10 @@ def generate_logs() -> Iterator[bytes]: _register_run_and_actor_endpoints(httpserver) api_url = httpserver.url_for('/').removesuffix('/') - run_client = ApifyClient(token='mocked_token', api_url=api_url).run(run_id=_MOCKED_RUN_ID) + http_client = http_client_class() + run_client = ApifyClient.with_custom_http_client( + token='mocked_token', api_url=api_url, http_client=http_client + ).run(run_id=_MOCKED_RUN_ID) streamed_log = run_client.get_streamed_log() logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}' @@ -913,6 +916,7 @@ def generate_logs() -> Iterator[bytes]: finally: release_server.set() streamed_log.stop() + http_client.close() leaked = [args.exc_type.__name__ for args in thread_exceptions] assert not leaked, f'streaming thread leaked an uncaught exception: {leaked}' @@ -963,9 +967,10 @@ def _recording_call(**kwargs: object) -> object: async def test_streamed_log_async_does_not_error_on_stream_timeout( caplog: LogCaptureFixture, httpserver: HTTPServer, + http_client_async_class: type[HttpClientAsync], monkeypatch: pytest.MonkeyPatch, ) -> None: - """The async streaming task ends quietly on a stream-request timeout, matching the sync regression for #1040.""" + """The async streaming task treats either transport's stream timeout as an expected terminal condition.""" monkeypatch.setattr(StreamedLogAsync, '_stream_timeout', timedelta(seconds=1)) release_server = threading.Event() @@ -984,7 +989,10 @@ def generate_logs() -> Iterator[bytes]: _register_run_and_actor_endpoints(httpserver) api_url = httpserver.url_for('/').removesuffix('/') - run_client = ApifyClientAsync(token='mocked_token', api_url=api_url).run(run_id=_MOCKED_RUN_ID) + http_client = http_client_async_class() + run_client = ApifyClientAsync.with_custom_http_client( + token='mocked_token', api_url=api_url, http_client=http_client + ).run(run_id=_MOCKED_RUN_ID) streamed_log = await run_client.get_streamed_log() logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}' @@ -997,6 +1005,7 @@ def generate_logs() -> Iterator[bytes]: finally: release_server.set() await streamed_log.stop() + await http_client.aclose() assert not task.cancelled() assert task.exception() is None, f'async streaming task raised on stream timeout: {task.exception()!r}' diff --git a/tests/unit/test_pluggable_http_client.py b/tests/unit/test_pluggable_http_client.py index eb24183c..8ec5a164 100644 --- a/tests/unit/test_pluggable_http_client.py +++ b/tests/unit/test_pluggable_http_client.py @@ -1,8 +1,16 @@ from __future__ import annotations +import asyncio import json as jsonlib +import subprocess +import sys from dataclasses import dataclass, field +from datetime import timedelta +from http.client import HTTPConnection +from textwrap import dedent from typing import TYPE_CHECKING, Any +from unittest.mock import AsyncMock, Mock +from urllib.parse import urlsplit import impit import pytest @@ -92,7 +100,7 @@ def call( json: Any = None, stream: bool | None = None, timeout: Timeout = 'medium', - ) -> FakeResponse: + ) -> HttpResponse: self.calls.append( { 'method': method, @@ -126,7 +134,7 @@ async def call( json: Any = None, stream: bool | None = None, timeout: Timeout = 'medium', - ) -> FakeResponse: + ) -> HttpResponse: self.calls.append( { 'method': method, @@ -142,7 +150,74 @@ async def call( return _make_fake_response() -# -- Protocol / ABC conformance tests -- +def _stdlib_fetch( + *, + method: str, + url: str, + headers: dict[str, str], + content: bytes | None, + timeout: float | None, +) -> FakeResponse: + """Send one request over `http.client` and adapt the result to the `HttpResponse` protocol.""" + parts = urlsplit(url) + connection = HTTPConnection(parts.hostname or 'localhost', parts.port, timeout=timeout) + try: + path = f'{parts.path}?{parts.query}' if parts.query else parts.path + connection.request(method, path, body=content, headers=headers) + raw = connection.getresponse() + body = raw.read() + try: + parsed = jsonlib.loads(body) + except ValueError: + parsed = None + return FakeResponse( + status_code=raw.status, + text=body.decode(), + content=body, + headers={key.lower(): value for key, value in raw.getheaders()}, + _json=parsed, + ) + finally: + connection.close() + + +class StdlibHttpClient(HttpClient): + """A hooks-only custom sync client: implements `send_request` and inherits the shared pipeline.""" + + def send_request( + self, + *, + method: str, + url: str, + headers: dict[str, str], + content: bytes | None, + timeout: float | None, + stream: bool, + ) -> HttpResponse: + _ = stream + return _stdlib_fetch(method=method, url=url, headers=headers, content=content, timeout=timeout) + + +class StdlibHttpClientAsync(HttpClientAsync): + """A hooks-only custom async client: implements `send_request` and inherits the shared pipeline.""" + + async def send_request( + self, + *, + method: str, + url: str, + headers: dict[str, str], + content: bytes | None, + timeout: float | None, + stream: bool, + ) -> HttpResponse: + _ = stream + return await asyncio.to_thread( + _stdlib_fetch, method=method, url=url, headers=headers, content=content, timeout=timeout + ) + + +# -- Protocol conformance tests -- def test_fake_response_satisfies_http_response_protocol() -> None: @@ -163,15 +238,15 @@ def test_fake_http_client_async_is_http_client_async() -> None: assert isinstance(client, HttpClientAsync) -def test_apify_http_client_is_http_client() -> None: - """Test that ImpitHttpClient is an instance of HttpClient.""" - client = ImpitHttpClient() +def test_apify_http_client_is_http_client(http_client_class: type[HttpClient]) -> None: + """Test that each built-in synchronous client is an HttpClient.""" + client = http_client_class() assert isinstance(client, HttpClient) -def test_apify_http_client_async_is_http_client_async() -> None: - """Test that ImpitHttpClientAsync is an instance of HttpClientAsync.""" - client = ImpitHttpClientAsync() +def test_apify_http_client_async_is_http_client_async(http_client_async_class: type[HttpClientAsync]) -> None: + """Test that each built-in asynchronous client is an HttpClientAsync.""" + client = http_client_async_class() assert isinstance(client, HttpClientAsync) @@ -184,16 +259,16 @@ async def test_fake_response_async_methods() -> None: assert chunks == [b'hello'] -def test_http_client_abc_not_instantiable() -> None: - """Test that HttpClient cannot be instantiated directly (it's abstract).""" - with pytest.raises(TypeError, match='abstract method'): - HttpClient() +def test_http_client_without_transport_fails_loudly() -> None: + """The sync base carries hook defaults, but its inherited `call` still needs a transport implementation.""" + with pytest.raises(NotImplementedError, match='Implement `send_request`'): + HttpClient().call(method='GET', url='https://example.com') -def test_http_client_async_abc_not_instantiable() -> None: - """Test that HttpClientAsync cannot be instantiated directly (it's abstract).""" - with pytest.raises(TypeError, match='abstract method'): - HttpClientAsync() +async def test_http_client_async_without_transport_fails_loudly() -> None: + """The async base carries hook defaults, but its inherited `call` still needs a transport implementation.""" + with pytest.raises(NotImplementedError, match='Implement `send_request`'): + await HttpClientAsync().call(method='GET', url='https://example.com') # -- ApifyClient with custom http_client via classmethod -- @@ -292,10 +367,61 @@ async def test_apify_client_async_with_custom_http_client_accepts_url_params() - def test_public_exports() -> None: """HTTP client types are exposed from `apify_client.http_clients`, not the root namespace.""" - for name in ('HttpClient', 'HttpClientAsync', 'HttpResponse', 'ImpitHttpClient', 'ImpitHttpClientAsync'): + for name in ( + 'HttpClient', + 'HttpClientAsync', + 'HttpResponse', + 'HttpxHttpClient', + 'HttpxHttpClientAsync', + 'ImpitHttpClient', + 'ImpitHttpClientAsync', + ): assert hasattr(http_clients_module, name) assert not hasattr(apify_client_module, name) + assert not hasattr(http_clients_module, 'HttpClientBase') + + +def test_httpx_clients_raise_clear_error_when_extra_missing() -> None: + """Missing HTTPX keeps normal and star imports usable while explicit HTTPX access raises a clear error.""" + script = dedent( + """ + import sys + + class BlockHttpx: + def find_spec(self, name, *_args): + if name == 'httpx' or name.startswith('httpx.'): + raise ModuleNotFoundError(f"No module named '{name}'", name='httpx') + return None + + sys.meta_path.insert(0, BlockHttpx()) + + import apify_client.http_clients as module + assert module.HttpClient is not None + assert module.ImpitHttpClient is not None + + namespace = {} + exec('from apify_client.http_clients import *', namespace) + assert namespace['HttpClient'] is module.HttpClient + assert 'HttpxHttpClient' not in namespace + + for name in ('HttpxHttpClient', 'HttpxHttpClientAsync'): + try: + getattr(module, name) + except ImportError as exc: + assert "No module named 'httpx'" in str(exc) + else: + raise AssertionError(f'{name} did not raise ImportError') + """ + ) + result = subprocess.run( # noqa: S603 + [sys.executable, '-c', script], + check=False, + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stderr + # -- http_client property -- @@ -374,14 +500,14 @@ async def test_custom_http_client_async_error_handling() -> None: # -- Integration with real HTTP server -- -def test_custom_http_client_with_real_server(httpserver: HTTPServer) -> None: - """Test that a custom HTTP client wrapping ImpitHttpClient works with a real server.""" +def test_custom_http_client_with_real_server(httpserver: HTTPServer, http_client_class: type[HttpClient]) -> None: + """Test that a custom HTTP client wrapping a built-in client works with a real server.""" httpserver.expect_request('/v2/datasets/test-dataset').respond_with_json( {'data': {'id': 'test-dataset', 'name': 'My Dataset'}}, ) # Create a wrapping client that adds custom headers - inner_client = ImpitHttpClient(token='test_token') + inner_client = http_client_class(token='test_token') class WrappingHttpClient(HttpClient): def call(self, *, method: str, url: str, **kwargs: Any) -> HttpResponse: @@ -400,14 +526,16 @@ def call(self, *, method: str, url: str, **kwargs: Any) -> HttpResponse: assert result['data']['id'] == 'test-dataset' -async def test_custom_http_client_async_with_real_server(httpserver: HTTPServer) -> None: - """Test that a custom async HTTP client wrapping ImpitHttpClientAsync works with a real server.""" +async def test_custom_http_client_async_with_real_server( + httpserver: HTTPServer, http_client_async_class: type[HttpClientAsync] +) -> None: + """Test that a custom async HTTP client wrapping a built-in client works with a real server.""" httpserver.expect_request('/v2/datasets/test-dataset').respond_with_json( {'data': {'id': 'test-dataset', 'name': 'My Dataset'}}, ) # Create a wrapping client that adds custom headers - inner_client = ImpitHttpClientAsync(token='test_token') + inner_client = http_client_async_class(token='test_token') class WrappingHttpClientAsync(HttpClientAsync): async def call(self, *, method: str, url: str, **kwargs: Any) -> HttpResponse: @@ -517,12 +645,14 @@ async def test_custom_http_client_async_sends_token_from_classmethod(httpserver: assert result['received_headers']['Authorization'] == 'Bearer test_token' -def test_custom_http_client_impit_instance_sends_token(httpserver: HTTPServer) -> None: - """Token from with_custom_http_client reaches the wire even for a pre-built tokenless ImpitHttpClient.""" +def test_custom_builtin_http_client_instance_sends_token( + httpserver: HTTPServer, http_client_class: type[HttpClient] +) -> None: + """Token from with_custom_http_client reaches the wire for each pre-built tokenless client.""" httpserver.expect_request('/v2/datasets/test-dataset').respond_with_handler(_echo_headers) api_url = httpserver.url_for('/').removesuffix('/') - client = ApifyClient.with_custom_http_client(token='test_token', api_url=api_url, http_client=ImpitHttpClient()) + client = ApifyClient.with_custom_http_client(token='test_token', api_url=api_url, http_client=http_client_class()) result = client.dataset('test-dataset')._get(timeout='short') @@ -548,3 +678,123 @@ def test_custom_http_client_keeps_differently_cased_authorization() -> None: assert 'Authorization' not in http_client._headers assert http_client._headers['authorization'] == 'Bearer client_token' + + +def test_hooks_only_client_retries_and_sends_default_headers(httpserver: HTTPServer) -> None: + """A client implementing only `send_request` inherits header merging and 5xx retries from the shared `call`.""" + auth_headers: list[str | None] = [] + + def handler(request: Request) -> Response: + auth_headers.append(request.headers.get('Authorization')) + if len(auth_headers) < 3: + return Response( + response='{"error": {"type": "internal-error", "message": "Server exploded."}}', + status=500, + content_type='application/json', + ) + return Response(response='{"data": {"id": "abc"}}', status=200, content_type='application/json') + + httpserver.expect_request('/v2/things/abc').respond_with_handler(handler) + + client = StdlibHttpClient(token='hook_token', min_delay_between_retries=timedelta(milliseconds=1)) + response = client.call(method='GET', url=httpserver.url_for('/v2/things/abc')) + + assert response.status_code == 200 + assert response.json() == {'data': {'id': 'abc'}} + assert auth_headers == ['Bearer hook_token'] * 3 + + +def test_hooks_only_client_raises_api_error_without_retry(httpserver: HTTPServer) -> None: + """A non-retryable error status reaches the caller as `ApifyApiError` after a single attempt.""" + request_count = 0 + + def handler(_request: Request) -> Response: + nonlocal request_count + request_count += 1 + return Response( + response='{"error": {"type": "record-not-found", "message": "Not there."}}', + status=404, + content_type='application/json', + ) + + httpserver.expect_request('/v2/things/missing').respond_with_handler(handler) + + client = StdlibHttpClient(token='hook_token') + with pytest.raises(ApifyApiError) as exc_info: + client.call(method='GET', url=httpserver.url_for('/v2/things/missing')) + + assert exc_info.value.status_code == 404 + assert request_count == 1 + + +async def test_hooks_only_client_async_retries_and_sends_default_headers(httpserver: HTTPServer) -> None: + """The async shared `call` gives a `send_request`-only client header merging and 5xx retries too.""" + auth_headers: list[str | None] = [] + + def handler(request: Request) -> Response: + auth_headers.append(request.headers.get('Authorization')) + if len(auth_headers) < 3: + return Response( + response='{"error": {"type": "internal-error", "message": "Server exploded."}}', + status=500, + content_type='application/json', + ) + return Response(response='{"data": {"id": "abc"}}', status=200, content_type='application/json') + + httpserver.expect_request('/v2/things/abc').respond_with_handler(handler) + + client = StdlibHttpClientAsync(token='hook_token', min_delay_between_retries=timedelta(milliseconds=1)) + response = await client.call(method='GET', url=httpserver.url_for('/v2/things/abc')) + + assert response.status_code == 200 + assert response.json() == {'data': {'id': 'abc'}} + assert auth_headers == ['Bearer hook_token'] * 3 + + +async def test_hooks_only_client_async_raises_api_error_without_retry(httpserver: HTTPServer) -> None: + """The async pipeline raises `ApifyApiError` without retrying a non-retryable status.""" + request_count = 0 + + def handler(_request: Request) -> Response: + nonlocal request_count + request_count += 1 + return Response( + response='{"error": {"type": "record-not-found", "message": "Not there."}}', + status=404, + content_type='application/json', + ) + + httpserver.expect_request('/v2/things/missing').respond_with_handler(handler) + + client = StdlibHttpClientAsync(token='hook_token') + with pytest.raises(ApifyApiError) as exc_info: + await client.call(method='GET', url=httpserver.url_for('/v2/things/missing')) + + assert exc_info.value.status_code == 404 + assert request_count == 1 + + +def test_hooks_only_client_does_not_retry_an_unclassified_transport_error(monkeypatch: pytest.MonkeyPatch) -> None: + """The default `is_retryable_transport_error` classifies nothing, so a transport failure ends the call at once.""" + client = StdlibHttpClient(token='hook_token', min_delay_between_retries=timedelta(milliseconds=1)) + send_request = Mock(side_effect=OSError('connection reset')) + monkeypatch.setattr(client, 'send_request', send_request) + + with pytest.raises(OSError, match='connection reset'): + client.call(method='GET', url='https://example.com') + + send_request.assert_called_once() + + +async def test_hooks_only_client_async_does_not_retry_an_unclassified_transport_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The asynchronous pipeline gives up on the first transport failure the client does not classify either.""" + client = StdlibHttpClientAsync(token='hook_token', min_delay_between_retries=timedelta(milliseconds=1)) + send_request = AsyncMock(side_effect=OSError('connection reset')) + monkeypatch.setattr(client, 'send_request', send_request) + + with pytest.raises(OSError, match='connection reset'): + await client.call(method='GET', url='https://example.com') + + send_request.assert_awaited_once() diff --git a/uv.lock b/uv.lock index 51cff51f..03dfaa88 100644 --- a/uv.lock +++ b/uv.lock @@ -54,6 +54,9 @@ dependencies = [ brotli = [ { name = "brotli" }, ] +httpx = [ + { name = "httpx" }, +] [package.dev-dependencies] dev = [ @@ -80,12 +83,13 @@ dev = [ requires-dist = [ { name = "brotli", marker = "extra == 'brotli'", specifier = ">=1.0.9" }, { name = "colorama", specifier = ">=0.4.0" }, + { name = "httpx", marker = "extra == 'httpx'", specifier = ">=0.27.0,<1.0.0" }, { name = "impit", specifier = "~=0.13.0" }, { name = "more-itertools", specifier = ">=10.0.0" }, { name = "pydantic", extras = ["email"], specifier = ">=2.11.0" }, { name = "typing-extensions", specifier = ">=4.6.0" }, ] -provides-extras = ["brotli"] +provides-extras = ["brotli", "httpx"] [package.metadata.requires-dev] dev = [