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

README.md

@@ -133,7 +133,7 @@ For a guided walkthrough — authenticating, running an Actor, and reading its r
 - **Fully typed responses** — every method returns a [Pydantic](https://docs.pydantic.dev/) model generated from the Apify OpenAPI spec, with IDE autocomplete and runtime validation ([Typed models](https://docs.apify.com/api/client/python/docs/concepts/typed-models)).
 - **Automatic retries** — exponential backoff for network errors, HTTP 429, and 5xx responses, configurable per client ([Retries](https://docs.apify.com/api/client/python/docs/concepts/retries)).
 - **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)).
+- **Pagination and streaming** — iterate datasets, key-value store keys, or live logs without manual paging or buffering, and upload large records straight from a file or another stream ([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** — use the default [Impit](https://github.com/apify/impit)-based client, opt in to the built-in [HTTPX2](https://github.com/pydantic/httpx2) client, or plug in any custom implementation ([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)).

docs/02_concepts/09_streaming.mdx

@@ -1,7 +1,7 @@
 ---
 id: streaming-resources
 title: Streaming resources
-description: Stream large datasets, key-value store records, and logs without loading them into memory.
+description: Stream large datasets, key-value store records, and logs in both directions without loading them into memory.
 ---
 
 import Tabs from '@theme/Tabs';
@@ -12,16 +12,22 @@ import ApiLink from '@theme/ApiLink';
 
 import StreamingAsyncExample from '!!raw-loader!./code/09_streaming_async.py';
 import StreamingSyncExample from '!!raw-loader!./code/09_streaming_sync.py';
+import UploadFileAsyncExample from '!!raw-loader!./code/09_upload_file_async.py';
+import UploadFileSyncExample from '!!raw-loader!./code/09_upload_file_sync.py';
+import UploadGeneratorAsyncExample from '!!raw-loader!./code/09_upload_generator_async.py';
+import UploadGeneratorSyncExample from '!!raw-loader!./code/09_upload_generator_sync.py';
 
-Certain resources, such as dataset items, key-value store records, and logs, support streaming directly from the Apify API. This allows you to process large resources incrementally without downloading them entirely into memory, making it ideal for handling large or continuously updated data.
+Large resources don't have to pass through memory whole. The client streams downloads of dataset items, key-value store records, and logs as they arrive, and it streams uploads of key-value store records and Actor inputs from a file or another stream as they're read. Both directions work with the synchronous and the asynchronous client. A download can also feed an upload directly, so one Actor's output becomes another Actor's input without passing through memory.
 
-Supported streaming methods:
+## Streaming downloads
+
+Dataset items, key-value store records, and logs can be streamed directly from the Apify API, so you process them incrementally instead of downloading them whole. The supported methods are:
 
 - <ApiLink to="class/DatasetClient#stream_items">`DatasetClient.stream_items`</ApiLink> - Stream dataset items incrementally. Yields a raw streaming <ApiLink to="class/HttpResponse">`HttpResponse`</ApiLink>.
 - <ApiLink to="class/KeyValueStoreClient#stream_record">`KeyValueStoreClient.stream_record`</ApiLink> - Stream a key-value store record as raw data. Yields a `dict` with the `key`, `value`, and `content_type` fields, where `value` holds the raw streaming response, or `None` when the record doesn't exist.
 - <ApiLink to="class/LogClient#stream">`LogClient.stream`</ApiLink> - Stream logs in real time. Yields a raw streaming <ApiLink to="class/HttpResponse">`HttpResponse`</ApiLink>, or `None` when the log doesn't exist.
 
-All three methods are context managers. Consume the streamed data within a `with` block to ensure that the connection is closed automatically, preventing memory leaks or unclosed connections.
+All three methods are context managers. Consume the streamed data within a `with` block, so the connection is closed automatically.
 
 The following example shows how to stream the logs of an Actor run incrementally:
 
@@ -38,4 +44,70 @@ The following example shows how to stream the logs of an Actor run incrementally
     </TabItem>
 </Tabs>
 
-Streaming is ideal for processing large logs, datasets, or files incrementally without downloading them entirely into memory.
+## Streaming uploads
+
+<ApiLink to="class/KeyValueStoreClient#set_record">`KeyValueStoreClient.set_record`</ApiLink> and the `run_input` of <ApiLink to="class/ActorClient#start">`ActorClient.start`</ApiLink>, <ApiLink to="class/ActorClient#call">`ActorClient.call`</ApiLink>, and <ApiLink to="class/RunClient#metamorph">`RunClient.metamorph`</ApiLink> accept a value the client streams to the API in chunks:
+
+- A file-like object, meaning anything with a `read` method: an open file, an `io.BytesIO`, a member of a ZIP archive, or a pipe. The client reads it in 64 KiB chunks from its current position and doesn't close it. A `read` that takes no size is called once instead, so that source is held in memory whole.
+- An iterator of `bytes` or `str` chunks, such as a generator. The iterator decides the chunk sizes, and it's consumed by the upload, so passing the same one to a second upload sends an empty body.
+- A streaming <ApiLink to="class/HttpResponse">`HttpResponse`</ApiLink> from one of the download methods, whose body is forwarded chunk by chunk.
+
+The asynchronous client also accepts an async iterator and a file-like object with an `async def read`, such as a file opened with `aiofiles`. The synchronous client rejects those with a `TypeError`. A synchronous file or iterator works with both clients. The asynchronous client reads it in a worker thread, so the event loop stays free.
+
+Only the chunk being sent is held in memory, so the process uploading a file doesn't need memory for the whole file. What the API accepts doesn't change: an Actor input is still capped at 9 MB, and a streamed body over that size is rejected with a `413` during the upload.
+
+The following example uploads a file from disk. The asynchronous variant opens it with [aiofiles](https://github.com/Tinche/aiofiles), which isn't an `apify-client` dependency, so install it alongside the client:
+
+```bash
+pip install apify-client aiofiles
+```
+
+<Tabs>
+    <TabItem value="UploadFileAsyncExample" label="Async client" default>
+        <CodeBlock className="language-python">
+            {UploadFileAsyncExample}
+        </CodeBlock>
+    </TabItem>
+    <TabItem value="UploadFileSyncExample" label="Sync client">
+        <CodeBlock className="language-python">
+            {UploadFileSyncExample}
+        </CodeBlock>
+    </TabItem>
+</Tabs>
+
+With a generator, you produce the data while it uploads, for example from pages of a database query:
+
+<Tabs>
+    <TabItem value="UploadGeneratorAsyncExample" label="Async client" default>
+        <CodeBlock className="language-python">
+            {UploadGeneratorAsyncExample}
+        </CodeBlock>
+    </TabItem>
+    <TabItem value="UploadGeneratorSyncExample" label="Sync client">
+        <CodeBlock className="language-python">
+            {UploadGeneratorSyncExample}
+        </CodeBlock>
+    </TabItem>
+</Tabs>
+
+### Content type
+
+A file the standard library opened in text mode is uploaded as `text/plain; charset=utf-8`, encoded to UTF-8 chunk by chunk. Any other streamed value is uploaded as `application/octet-stream` unless you pass `content_type`, an `aiofiles` text file included - the client can't tell it from a binary one. An iterator or a response carries no type of its own, so set `content_type` explicitly, as the examples do.
+
+### Compression
+
+A streamed body is never compressed, because the client sees each chunk only as it's sent. To upload compressed data, compress it yourself and declare the encoding with `content_encoding`. The client then forwards the bytes and the header as they are. For details, see [Pre-compressed bodies](./13_http_compression.mdx#pre-compressed-bodies).
+
+### Retries
+
+The client retries a failed request only when it can send the body again:
+
+- A seekable file-like object, such as an open file or an `io.BytesIO`, is sought back to where it started before each retry, so the upload is retried like any other request.
+- An iterator, a streaming response, a pipe, or any asynchronous source is consumed by the attempt that sends it. The request gets a single attempt, and a failure raises immediately.
+- A failure inside the source itself, such as a disk error while reading the file, is raised as that error and isn't retried, whatever the source.
+
+If an upload from a source that can't be rewound needs retries, download the data to a temporary file first and upload the file. In the asynchronous client, prefer a file opened with `open()` over one opened with `aiofiles` when the upload has to survive a retry: the asynchronous client reads it in a worker thread either way, but only the plain file can be sought back.
+
+### Custom HTTP clients
+
+A custom transport receives a streamed body as an iterator of `bytes` chunks, or as an async iterator in the asynchronous client, so any HTTP library that sends iterable bodies can send it. The <ApiLink to="class/StreamedRequestBody">`StreamedRequestBody`</ApiLink> class documents the rest of the contract. For details, see [The transport contract](./10_custom_http_clients.mdx#the-transport-contract).

docs/02_concepts/10_custom_http_clients.mdx

@@ -98,7 +98,7 @@ The base classes and the response protocol are available from the `apify_client.
 
 The public `call` method provides the shared request pipeline. A concrete transport implements these hooks:
 
-- `send_request(...)` sends one prepared request and returns an `HttpResponse`, error statuses included. It receives the URL with the query parameters already encoded into it, the headers with the client's default headers already merged in, the body already serialized and compressed, and the timeout for this attempt in seconds. The inherited `call` needs it, so every transport adapter has to implement it. Let the HTTP library's exceptions propagate unwrapped, and leave status handling and <ApiLink to="class/ApifyApiError">`ApifyApiError`</ApiLink> to `call`.
+- `send_request(...)` sends one prepared request and returns an `HttpResponse`, error statuses included. It receives the URL with the query parameters already encoded into it, the headers with the client's default headers already merged in, the body, and the timeout for this attempt in seconds. The body is either `bytes`, already serialized and compressed, or an iterator of `bytes` chunks for a streamed body, which is an async iterator in the asynchronous client. The inherited `call` needs it, so every transport adapter has to implement it. Let the HTTP library's exceptions propagate unwrapped, and leave status handling and <ApiLink to="class/ApifyApiError">`ApifyApiError`</ApiLink> to `call`.
 - `is_retryable_transport_error(exc)` classifies transport failures for the shared retry loop. The default classifies nothing as retryable, so a transport that doesn't override it gives up on the first connection failure.
 - `is_timeout_error(exc)` identifies transport-specific timeout exceptions for higher-level client features. The default recognizes Python's `TimeoutError`. Timeout classification is independent of retryability, so a timeout the retry loop should retry has to be listed in `is_retryable_transport_error` too.
 - `close()` or `aclose()` closes resources owned by the transport. The default does nothing, which is correct for a transport that owns no pool or session.
@@ -152,6 +152,8 @@ After that, all API calls made through the client will go through your custom HT
 If you override `call` itself, your implementation becomes responsible for request preparation, retries, timeouts, API error conversion, logging, and statistics. Implementing the transport hooks and inheriting `call` keeps the shared behavior.
 :::
 
+If you override `call` and prepare requests with the inherited helpers, a streamed body arrives as a <ApiLink to="class/StreamedRequestBody">`StreamedRequestBody`</ApiLink>. Send its `iter_bytes()` or `aiter_bytes()`, and send the body a second time only when `rewindable` is true, after calling `rewind()`. For what the client streams and how, see [Streaming uploads](./09_streaming.mdx#streaming-uploads).
+
 ## Use cases
 
 Custom HTTP clients might be useful when the built-in Impit and HTTPX2 clients don't cover your requirements, for example when you need to:

docs/02_concepts/13_http_compression.mdx

@@ -17,7 +17,7 @@ The Apify client compresses request bodies before sending them to the API. It re
 
 ## How it works
 
-The client compresses request bodies using the compressor configured via the `compression` parameter (default `'gzip'`). The server supports both gzip and brotli and decompresses the request body transparently. A body is compressed only when it's large enough to benefit, its content type isn't already compressed, and the request carries no `Content-Encoding` of its own. For details, see [Minimum body size](#minimum-body-size), [Already-compressed payloads](#already-compressed-payloads), and [Pre-compressed bodies](#pre-compressed-bodies).
+The client compresses request bodies using the compressor configured via the `compression` parameter (default `'gzip'`). The server supports both gzip and brotli and decompresses the request body transparently. A body is compressed only when it's large enough to benefit, its content type isn't already compressed, the request carries no `Content-Encoding` of its own, and the body isn't streamed from a file or an iterator. For details, see [Minimum body size](#minimum-body-size), [Already-compressed payloads](#already-compressed-payloads), [Pre-compressed bodies](#pre-compressed-bodies), and [Streaming uploads](./09_streaming.mdx#streaming-uploads).
 
 ## Minimum body size
 
@@ -47,7 +47,7 @@ Two kinds of media type are compressed anyway: raw formats such as `image/bmp`,
     </TabItem>
 </Tabs>
 
-Without an explicit content type, a `bytes` value is sent as `application/octet-stream`, which the client can't tell apart from uncompressed binary data and therefore still compresses. File-like values are read into memory before they're sent, so they follow the same rules as any other body.
+Without an explicit content type, a `bytes` value is sent as `application/octet-stream`, which the client can't tell apart from uncompressed binary data and therefore still compresses. A streamed value, such as an open file or an iterator of chunks, is never compressed. For details, see [Streaming uploads](./09_streaming.mdx#streaming-uploads).
 
 ## Pre-compressed bodies
 
@@ -66,7 +66,7 @@ A payload can reach the client already encoded, for example a gzipped file read
     </TabItem>
 </Tabs>
 
-The header is forwarded verbatim, so it also covers encodings the client ships no compressor for, such as `deflate`. The API accepts `gzip`, `br`, `deflate`, and `identity`. Passing `identity` turns compression off for a single request without changing how the client is configured.
+The header is forwarded verbatim, so it also covers encodings the client ships no compressor for, such as `deflate`. It's forwarded for a streamed value too, so an open gzipped file can be passed in place of its bytes. The API accepts `gzip`, `br`, `deflate`, and `identity`. Passing `identity` turns compression off for a single request without changing how the client is configured.
 
 A value that can't be compressed at all - a string, an object serialized to JSON, or a file-like value opened in text mode - is rejected with a `TypeError` when `content_encoding` names a compression. Beyond that the client can't verify that the bytes match the header, so set `Content-Encoding` only when the payload really is encoded that way. Key-value store records are stored exactly as you upload them, which makes the header part of the stored record rather than a transport detail.
 

docs/02_concepts/code/09_upload_file_async.py

@@ -0,0 +1,16 @@
+import aiofiles
+
+from apify_client import ApifyClientAsync
+
+TOKEN = 'MY-APIFY-TOKEN'
+
+
+async def main() -> None:
+    apify_client = ApifyClientAsync(TOKEN)
+    kvs_client = apify_client.key_value_store('MY-KVS-ID')
+
+    # The file is read in chunks as it uploads, so its size doesn't matter.
+    async with aiofiles.open('backup.tar.gz', 'rb') as backup:
+        await kvs_client.set_record(
+            'backup.tar.gz', backup, content_type='application/gzip'
+        )

docs/02_concepts/code/09_upload_file_sync.py

@@ -0,0 +1,14 @@
+from pathlib import Path
+
+from apify_client import ApifyClient
+
+TOKEN = 'MY-APIFY-TOKEN'
+
+
+def main() -> None:
+    apify_client = ApifyClient(TOKEN)
+    kvs_client = apify_client.key_value_store('MY-KVS-ID')
+
+    # The file is read in chunks as it uploads, so its size doesn't matter.
+    with Path('backup.tar.gz').open('rb') as backup:
+        kvs_client.set_record('backup.tar.gz', backup, content_type='application/gzip')

docs/02_concepts/code/09_upload_generator_async.py

@@ -0,0 +1,21 @@
+from collections.abc import AsyncIterator
+
+from apify_client import ApifyClientAsync
+
+TOKEN = 'MY-APIFY-TOKEN'
+
+
+async def csv_chunks() -> AsyncIterator[str]:
+    """Build the CSV in pieces, for example from pages of a database query."""
+    yield 'id,value\n'
+    for start in range(0, 1_000_000, 10_000):
+        yield ''.join(f'{i},{i * i}\n' for i in range(start, start + 10_000))
+
+
+async def main() -> None:
+    apify_client = ApifyClientAsync(TOKEN)
+    kvs_client = apify_client.key_value_store('MY-KVS-ID')
+
+    # Each chunk is encoded and sent as it's produced. An iterator can't be
+    # rewound, so a failed upload isn't retried.
+    await kvs_client.set_record('report.csv', csv_chunks(), content_type='text/csv')

docs/02_concepts/code/09_upload_generator_sync.py

@@ -0,0 +1,21 @@
+from collections.abc import Iterator
+
+from apify_client import ApifyClient
+
+TOKEN = 'MY-APIFY-TOKEN'
+
+
+def csv_chunks() -> Iterator[str]:
+    """Build the CSV in pieces, for example from pages of a database query."""
+    yield 'id,value\n'
+    for start in range(0, 1_000_000, 10_000):
+        yield ''.join(f'{i},{i * i}\n' for i in range(start, start + 10_000))
+
+
+def main() -> None:
+    apify_client = ApifyClient(TOKEN)
+    kvs_client = apify_client.key_value_store('MY-KVS-ID')
+
+    # Each chunk is encoded and sent as it's produced. An iterator can't be
+    # rewound, so a failed upload isn't retried.
+    kvs_client.set_record('report.csv', csv_chunks(), content_type='text/csv')

docs/02_concepts/code/10_architecture_imports.py

@@ -2,4 +2,5 @@
     HttpClient,
     HttpClientAsync,
     HttpResponse,
+    StreamedRequestBody,
 )

docs/02_concepts/code/10_plugging_in_async.py

@@ -1,3 +1,5 @@
+from collections.abc import AsyncIterator
+
 from typing_extensions import override
 
 from apify_client import ApifyClientAsync
@@ -16,7 +18,7 @@ async def send_request(
         method: str,
         url: str,
         headers: dict[str, str],
-        content: bytes | None,
+        content: bytes | AsyncIterator[bytes] | None,
         timeout: float | None,
         stream: bool,
     ) -> HttpResponse: