Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
d03ebc6
feat: Stream request bodies from files, iterators, and responses
vdusek Sep 11, 2026
10c22d9
docs: Drop the Actor pipeline guide from the streaming docs
vdusek Sep 11, 2026
e9445bf
fix: Send a caller-built streamed request body as it is and reject un…
vdusek Sep 11, 2026
d8be115
refactor: Rewind a streamed request body before every request attempt
vdusek Sep 11, 2026
144b0d0
docs: Correct the wording for streamed values in docs and docstrings
vdusek Sep 11, 2026
a6ef26a
docs: Clarify streaming upload retries, size cap, and chunk size units
vdusek Sep 16, 2026
0e036af
refactor: Narrow the types of _is_body_worth_compressing and _is_resp…
vdusek Sep 16, 2026
50e5894
test: Make the streamed record payload detect reordered chunks
vdusek Sep 16, 2026
3e56d01
fix: Accept a file-like read that takes no size argument
vdusek Sep 16, 2026
824eb7c
feat: Stream request bodies from objects that only implement __iter__…
Pijukatel Sep 18, 2026
f3957ba
refactor: Stream only io.IOBase sources in chunks and rename is_sourc…
vdusek Sep 25, 2026
646d970
Merge remote-tracking branch 'origin/master' into feat/streamed-reque…
vdusek Sep 25, 2026
3826cdf
fix: Stream a text file whose position is disabled by next()
vdusek Sep 26, 2026
6460ba8
fix: Close a coroutine chunk before rejecting it
vdusek Sep 26, 2026
fc2fd34
refactor: Drop the unreachable async branch from the chunked read loop
vdusek Sep 26, 2026
d570ce2
test: Use ASCII escapes in streamed body tests
vdusek Sep 26, 2026
acd9874
docs: Say which file-like values stream in the upload docstrings
vdusek Sep 26, 2026
00f24e2
docs: Cover non-IOBase readers and temp files in the streaming upload…
vdusek Sep 26, 2026
4656943
refactor: Read and convert each async chunk in separate steps
vdusek Sep 26, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)).
Expand Down
79 changes: 74 additions & 5 deletions docs/02_concepts/09_streaming.mdx
Original file line number Diff line number Diff line change
@@ -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';
Expand All @@ -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:

Expand All @@ -38,4 +44,67 @@ 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:

- An `io.IOBase` stream: an open file, an `io.BytesIO`, a member of a ZIP archive, or a pipe. The client reads it in chunks of 64 KiB, or 64 Ki characters for a text-mode file, from its current position and doesn't close it.
- Any other object with a `read` method. The client can't tell how that `read` behaves, so it calls `read()` once and sends the result as a single chunk, which holds the whole source in memory.
- An iterable of `bytes` or `str` chunks: an iterator such as a generator, or an object that only implements `__iter__`, for example a class that yields chunks from a generator method. The iterable decides the chunk sizes. An iterator is consumed by the upload, so passing the same one to a second upload sends an empty body. A `str`, `bytes`, `list`, `tuple`, `dict`, or other container, and a pydantic model, are never streamed even though they can be iterated - they are the values the client uploads whole or serializes as JSON.
- 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 iterable - an async generator or an object that only implements `__aiter__` - and an object with an `async def read`, such as a file opened with `aiofiles`, which is read whole like any other non-`io.IOBase` object. 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.

Apart from a non-`io.IOBase` object with a `read` method, 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. Both variants pass a plain file, which the asynchronous client reads in a worker thread.

<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 `io.IOBase` stream, 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, an object with a `read` method outside `io.IOBase`, 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. The object `tempfile.NamedTemporaryFile()` returns, which is also what `tempfile.TemporaryFile()` returns on Windows, isn't an `io.IOBase` stream, so pass its `file` attribute to have it read in chunks and retried. In the asynchronous client, prefer a file opened with `open()` over one opened with `aiofiles`: the plain file is read in chunks in a worker thread and can be sought back for a retry, while the `aiofiles` file is read whole in one call and gets a single attempt.

### 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).
4 changes: 3 additions & 1 deletion docs/02_concepts/10_custom_http_clients.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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:
Expand Down
6 changes: 3 additions & 3 deletions docs/02_concepts/13_http_compression.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand All @@ -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.

Expand Down
19 changes: 19 additions & 0 deletions docs/02_concepts/code/09_upload_file_async.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
import asyncio
from pathlib import Path

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 in a worker thread as it uploads, so its size
# doesn't matter.
backup = await asyncio.to_thread(Path('backup.tar.gz').open, 'rb')
with backup:
await kvs_client.set_record(
'backup.tar.gz', backup, content_type='application/gzip'
)
14 changes: 14 additions & 0 deletions docs/02_concepts/code/09_upload_file_sync.py
Original file line number Diff line number Diff line change
@@ -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')
21 changes: 21 additions & 0 deletions docs/02_concepts/code/09_upload_generator_async.py
Original file line number Diff line number Diff line change
@@ -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')
Loading
Loading