fix: Keep HttpResponse isinstance checks from consuming streamed responses (#1010)

`HttpResponse` now comes from `typing_extensions`, whose runtime
protocol check looks attributes up statically. The `typing`
implementation on Python 3.11 calls `hasattr`, which evaluates
properties, so `isinstance(response, HttpResponse)` on an unread
streaming response either raised or silently buffered the whole body.

Adds streaming regression tests (unit and integration) and a
parametrized built-in-client fixture they run on.

Split out of #1006. Stacked on #1009.

*✍️ Drafted by Claude Code*

Author: vdusek

pyproject.toml

@@ -28,6 +28,8 @@ dependencies = [
     "impit~=0.13.0",
     "more_itertools>=10.0.0",
     "pydantic[email]>=2.11.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

@@ -7,9 +7,14 @@
 from abc import ABC, abstractmethod
 from datetime import UTC, datetime, timedelta
 from importlib import metadata
-from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Any
 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/integration/test_dataset.py

@@ -21,6 +21,7 @@
 from apify_client._models import Dataset, DatasetListItem, DatasetStatistics, ListOfDatasets
 from apify_client._resource_clients.dataset import DatasetItemsPage
 from apify_client.errors import ApifyApiError
+from apify_client.http_clients import HttpResponse
 
 if TYPE_CHECKING:
     from apify_client import ApifyClient, ApifyClientAsync
@@ -727,7 +728,7 @@ async def get_items() -> DatasetItemsPage:
         if is_async:
             assert isinstance(stream_ctx, AbstractAsyncContextManager)
             async with stream_ctx as response:
-                assert isinstance(response, impit.Response)
+                assert isinstance(response, HttpResponse)
                 assert response.status_code == 200
                 content = await response.aread()
                 items = json.loads(content)
@@ -736,7 +737,7 @@ async def get_items() -> DatasetItemsPage:
         else:
             assert isinstance(stream_ctx, AbstractContextManager)
             with stream_ctx as response:
-                assert isinstance(response, impit.Response)
+                assert isinstance(response, HttpResponse)
                 assert response.status_code == 200
                 content = response.read()
                 items = json.loads(content)

tests/integration/test_key_value_store.py

@@ -21,6 +21,7 @@
 )
 from apify_client._models import KeyValueStore, KeyValueStoreKey, ListOfKeys, ListOfKeyValueStores
 from apify_client.errors import ApifyApiError
+from apify_client.http_clients import HttpResponse
 
 if TYPE_CHECKING:
     from apify_client import ApifyClient, ApifyClientAsync
@@ -198,14 +199,14 @@ async def test_stream_record_signature(
             signature=test_kvs_of_another_user.keys_signature[key],
         ) as stream:  # ty: ignore[invalid-context-manager]
             assert isinstance(stream, dict)
-            value = json.loads(stream['value'].content.decode('utf-8'))
+            value = json.loads((await stream['value'].aread()).decode('utf-8'))
     else:
         with kvs.stream_record(
             key,
             signature=test_kvs_of_another_user.keys_signature[key],
         ) as stream:  # ty: ignore[invalid-context-manager]
             assert isinstance(stream, dict)
-            value = json.loads(stream['value'].content.decode('utf-8'))
+            value = json.loads(stream['value'].read().decode('utf-8'))
 
     assert test_kvs_of_another_user.expected_content[key] == value
 
@@ -726,11 +727,15 @@ async def added_record_exists() -> bool:
         if is_async:
             async with store_client.stream_record('stream-key') as stream:  # ty: ignore[invalid-context-manager]
                 assert isinstance(stream, dict)
-                value = json.loads(stream['value'].content.decode('utf-8'))
+                response = stream['value']
+                assert isinstance(response, HttpResponse)
+                value = json.loads((await response.aread()).decode('utf-8'))
         else:
             with store_client.stream_record('stream-key') as stream:  # ty: ignore[invalid-context-manager]
                 assert isinstance(stream, dict)
-                value = json.loads(stream['value'].content.decode('utf-8'))
+                response = stream['value']
+                assert isinstance(response, HttpResponse)
+                value = json.loads(response.read().decode('utf-8'))
 
         assert value == {'data': 'streamed'}
     finally:

tests/unit/conftest.py

@@ -6,6 +6,8 @@
 import pytest
 from pytest_httpserver import HTTPServer
 
+from apify_client.http_clients import HttpClient, HttpClientAsync, ImpitHttpClient, ImpitHttpClientAsync
+
 if TYPE_CHECKING:
     from collections.abc import Iterable
 
@@ -28,3 +30,15 @@ def httpserver(make_httpserver: HTTPServer) -> Iterable[HTTPServer]:
     server = make_httpserver
     yield server
     server.clear()
+
+
+@pytest.fixture(params=[pytest.param(ImpitHttpClient, id='impit')])
+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')])
+def http_client_async_class(request: pytest.FixtureRequest) -> type[HttpClientAsync]:
+    """Return each built-in asynchronous HTTP client class."""
+    return request.param

tests/unit/test_client_streaming.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from apify_client import ApifyClient, ApifyClientAsync
+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
+
+
+DATASET_ID = 'test-dataset-id'
+KVS_ID = 'test-kvs-id'
+RECORD_KEY = 'test-record-key'
+STREAM_CONTENT = b'[{"id": 1}]'
+
+
+def test_dataset_stream_items_sync(
+    httpserver: HTTPServer,
+    http_client_class: type[HttpClient],
+) -> None:
+    """Dataset streams expose a transport-independent response that can be read synchronously."""
+    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)
+        assert response.read() == STREAM_CONTENT
+
+
+async def test_dataset_stream_items_async(
+    httpserver: HTTPServer,
+    http_client_async_class: type[HttpClientAsync],
+) -> None:
+    """Dataset streams expose a transport-independent response that can be read asynchronously."""
+    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)
+        assert await response.aread() == STREAM_CONTENT
+
+
+def test_key_value_store_stream_record_sync(
+    httpserver: HTTPServer,
+    http_client_class: type[HttpClient],
+) -> None:
+    """KVS streams require reading the generic response before consuming its content."""
+    httpserver.expect_request(f'/v2/key-value-stores/{KVS_ID}/records/{RECORD_KEY}').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.key_value_store(KVS_ID).stream_record(RECORD_KEY) as record:
+        assert isinstance(record, dict)
+        response = record['value']
+        assert isinstance(response, HttpResponse)
+        assert response.read() == STREAM_CONTENT
+
+
+async def test_key_value_store_stream_record_async(
+    httpserver: HTTPServer,
+    http_client_async_class: type[HttpClientAsync],
+) -> None:
+    """KVS streams require asynchronously reading the generic response before consuming its content."""
+    httpserver.expect_request(f'/v2/key-value-stores/{KVS_ID}/records/{RECORD_KEY}').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.key_value_store(KVS_ID).stream_record(RECORD_KEY) as record:
+        assert isinstance(record, dict)
+        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 the built-in client exposes 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 the built-in client exposes it.
+        raw: Any = response
+        assert raw.is_stream_consumed is False