fix: Stop HttpResponse protocol checks from consuming streaming bodies

Author: vdusek

pyproject.toml

@@ -28,7 +28,8 @@ dependencies = [
     "impit~=0.13.0",
     "more_itertools>=10.0.0",
     "pydantic[email]>=2.11.0",
-    "typing-extensions>=4.4.0",
+    # 4.6.0 is the first release whose runtime protocol checks look attributes up statically.
+    "typing-extensions>=4.6.0",
 ]
 
 [project.optional-dependencies]

src/apify_client/http_clients/_base.py

@@ -11,9 +11,14 @@
 from datetime import UTC, datetime, timedelta
 from http import HTTPStatus
 from importlib import metadata
-from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable
+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
+# up statically. The `typing` implementation on Python 3.11 calls `hasattr`, which evaluates properties: on an
+# unread streaming response that either raises or silently buffers the whole body.
+from typing_extensions import Protocol, runtime_checkable
+
 from apify_client._consts import (
     DEFAULT_MAX_RETRIES,
     DEFAULT_MIN_DELAY_BETWEEN_RETRIES,

tests/unit/test_client_streaming.py

@@ -6,6 +6,8 @@
 from apify_client.http_clients import HttpResponse
 
 if TYPE_CHECKING:
+    from typing import Any
+
     from pytest_httpserver import HTTPServer
 
     from apify_client.http_clients import HttpClient, HttpClientAsync
@@ -87,3 +89,41 @@ async def test_key_value_store_stream_record_async(
         response = record['value']
         assert isinstance(response, HttpResponse)
         assert await response.aread() == STREAM_CONTENT
+
+
+def test_protocol_check_leaves_stream_unread_sync(
+    httpserver: HTTPServer,
+    http_client_class: type[HttpClient],
+) -> None:
+    """Checking a streaming response against the protocol inspects it without pulling the body off the wire."""
+    httpserver.expect_request(f'/v2/datasets/{DATASET_ID}/items').respond_with_data(STREAM_CONTENT)
+    api_url = httpserver.url_for('/').removesuffix('/')
+    client = ApifyClient.with_custom_http_client(
+        api_url=api_url,
+        http_client=http_client_class(),
+    )
+
+    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 both built-in clients expose it.
+        raw: Any = response
+        assert raw.is_stream_consumed is False
+
+
+async def test_protocol_check_leaves_stream_unread_async(
+    httpserver: HTTPServer,
+    http_client_async_class: type[HttpClientAsync],
+) -> None:
+    """Checking a streaming response against the protocol inspects it without pulling the body off the wire."""
+    httpserver.expect_request(f'/v2/datasets/{DATASET_ID}/items').respond_with_data(STREAM_CONTENT)
+    api_url = httpserver.url_for('/').removesuffix('/')
+    client = ApifyClientAsync.with_custom_http_client(
+        api_url=api_url,
+        http_client=http_client_async_class(),
+    )
+
+    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 both built-in clients expose it.
+        raw: Any = response
+        assert raw.is_stream_consumed is False