feat: Stream request bodies from files, iterators, and responses

[vdusek]

What changed

set_record, the run_input of start / call / metamorph, and HttpClient.call(data=...) now accept a file-like object, an iterator of bytes or str chunks, or a streamed HttpResponse. The body goes out chunked and uncompressed, so one Actor's OUTPUT record can become another Actor's input without the orchestrating process holding it in memory. The async client also takes async iterators and aiofiles-style objects; the sync client rejects those with a TypeError naming ApifyClientAsync.

A file-like object is read through read(chunk_size). A read that takes no size is called once and its result sent as a single chunk, so a duck-typed reader of that shape keeps working and keeps its old memory profile.

Retries

Only a seekable file-like source is retried, sought back to its starting position before each attempt. Anything else gets a single attempt. An error raised inside the body iterator surfaces as itself with the transport error as its cause, since Impit reports it as a bare impit.HTTPError that the classifier would treat as transient.

Compression

A file-like set_record value is no longer compressed, so uploading from a file handle sends more bytes over the wire, which Apify bills as data transfer. Passing file.read() keeps the old behavior. Not breaking: the call keeps working, and the extra data transfer belongs in the changelog rather than in a major bump.

A streamed value that declares a content_encoding is taken at its word, since the client only sees its bytes as they go out. A str, a JSON-serializable object, and a text-mode file are still rejected.

Custom transports

send_request now receives bytes | Iterator[bytes] | None (async: AsyncIterator[bytes]), and call accepts the new source types. A transport typed against bytes | None only sees an iterator once one of its own callers opts into streaming, so this isn't breaking either.

Docs

A "Streaming uploads" section on the streaming concept page, plus updates to the compression and HTTP client pages, the custom transport examples, and the README. The guide on chaining Actors into a pipeline is a separate PR.

Tests

StreamedRequestBody on its own, the retry pipeline with fake transports, and set_record end to end over Impit and HTTPX2. The two integration tests (3 MiB upload, record piped between stores) weren't run locally.

Issues

• Closes Support streaming request bodies to avoid buffering large uploads in memory #972

✍️ Drafted by Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 94.35028% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 95.21%. Comparing base (d5d4c6e) to head (3e56d01).
⚠️ Report is 7 commits behind head on master.

Files with missing lines	Patch %	Lines
src/apify_client/types.py	0.00%	9 Missing ⚠️
src/apify_client/http_clients/_streamed_body.py	99.19%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #1060      +/-   ##
==========================================
- Coverage   95.24%   95.21%   -0.03%     
==========================================
  Files          59       60       +1     
  Lines        5509     5669     +160     
==========================================
+ Hits         5247     5398     +151     
- Misses        262      271       +9     

Flag	Coverage Δ
integration	91.65% <67.79%> (-0.31%)	⬇️
unit	87.54% <94.35%> (+0.19%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Pijukatel]

StreamedRequestBody.is_source(data) reads more like is data an existing source of StreamedRequestBody

Maybe it should have a name that implies that the method is testing a possibility, not the existing reality.

I am bad with naming, but just to give a hint in which direction I think it should go
StreamedRequestBody.is_possible_source(data)
StreamedRequestBody.is_compatible_source(data)
StreamedRequestBody.can_use_as_source(data)
...

[Pijukatel]

This seems too restrictive for all iterables. (We want to keep some special iterables - like strings, lists, ... out of streaming, but is there really a reason to keep all of them out?l)

For example (simplified example just to demonstrate the point)

class Chunked:
    """Wraps a string and yields it in pieces. Only __iter__, nothing else."""

    def __init__(self, text: str, size: int = 10) -> None:
        self.text, self.size = text, size

    def __iter__(self):
        for i in range(0, len(self.text), self.size):
            yield self.text[i : i + self.size]


client.actor('E2jjCZBezvAZnX8Rb').start(
    run_input=Chunked('{"message": "Hello world!"}'),
    content_type='application/json',
)

On your PR it is serialized as
"<main.Chunked object at 0x7f97fb28a8d0>"

I asked to create a commit with an example implementation that would also support these:
7b34e44

[Pijukatel]

It does not feel very robust.

The detection in StreamedRequestBody is a chain of getattr probes. That works for the standard library, but this code takes arbitrary user objects, and for those the outcome depends on which method names they happen to have.

_accepts_chunk_size is going even further with guessing.

So for user-defined code or libraries we have not considered, this can behave in completely unpredictable ways.