fix: Respect caller-supplied Content-Encoding for pre-compressed request bodies

Author: vdusek

docs/02_concepts/13_http_compression.mdx

@@ -10,12 +10,14 @@ import CodeBlock from '@theme/CodeBlock';
 
 import SkipCompressionAsyncExample from '!!raw-loader!./code/13_skip_compression_async.py';
 import SkipCompressionSyncExample from '!!raw-loader!./code/13_skip_compression_sync.py';
+import PrecompressedAsyncExample from '!!raw-loader!./code/13_precompressed_async.py';
+import PrecompressedSyncExample from '!!raw-loader!./code/13_precompressed_sync.py';
 
 The Apify client compresses request bodies before sending them to the API. It reduces the amount of data transferred over the network, resulting in faster requests and lower bandwidth usage, especially for large payloads such as Actor inputs, dataset uploads, or key-value store records.
 
 ## 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 is large enough to benefit and its content type isn't already compressed, as the next two sections describe.
+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).
 
 ## Minimum body size
 
@@ -47,6 +49,27 @@ Two kinds of media type are compressed anyway: raw formats such as `image/bmp`,
 
 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 streamed rather than buffered, so they're never compressed regardless of their content type.
 
+## Pre-compressed bodies
+
+A payload can reach the client already encoded, for example a gzipped file read from disk. Set the `Content-Encoding` header to name the encoding the payload carries. The client then sends the body as it is and forwards the header, so nothing gets compressed twice. `set_record` exposes the header as its `content_encoding` argument:
+
+<Tabs>
+    <TabItem value="AsyncExample" label="Async client" default>
+        <CodeBlock className="language-python">
+            {PrecompressedAsyncExample}
+        </CodeBlock>
+    </TabItem>
+    <TabItem value="SyncExample" label="Sync client">
+        <CodeBlock className="language-python">
+            {PrecompressedSyncExample}
+        </CodeBlock>
+    </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 client can't verify that the body matches 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.
+
 ## Configuration
 
 To choose the compression algorithm, pass `compression` to the client constructor:
@@ -88,7 +111,7 @@ client = ApifyClient(token='MY-APIFY-TOKEN', compression=BrotliHttpCompressor(qu
 client = ApifyClient(token='MY-APIFY-TOKEN', compression=GzipHttpCompressor(quality=9))
 ```
 
-You can also implement a fully custom compressor by subclassing `HttpCompressor`. The client calls it only for bodies that reach the [minimum body size](#minimum-body-size) and aren't [already compressed](#already-compressed-payloads):
+You can also implement a fully custom compressor by subclassing `HttpCompressor`. The client calls it only for bodies that reach the [minimum body size](#minimum-body-size) and aren't [already compressed](#already-compressed-payloads) or [pre-compressed by the caller](#pre-compressed-bodies):
 
 ```python
 from apify_client import ApifyClient

docs/02_concepts/code/13_precompressed_async.py

@@ -0,0 +1,27 @@
+import asyncio
+import gzip
+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')
+
+    report = await asyncio.to_thread(Path('report.csv').read_bytes)
+    compressed_report = await asyncio.to_thread(gzip.compress, report)
+
+    # The explicit content encoding stops the client from compressing the bytes again.
+    await kvs_client.set_record(
+        'report',
+        compressed_report,
+        content_type='text/csv',
+        content_encoding='gzip',
+    )
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

docs/02_concepts/code/13_precompressed_sync.py

@@ -0,0 +1,22 @@
+import gzip
+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')
+
+    report = Path('report.csv').read_bytes()
+    compressed_report = gzip.compress(report)
+
+    # The explicit content encoding stops the client from compressing the bytes again.
+    kvs_client.set_record(
+        'report',
+        compressed_report,
+        content_type='text/csv',
+        content_encoding='gzip',
+    )

src/apify_client/_resource_clients/key_value_store.py

@@ -360,6 +360,7 @@ def set_record(
         value: Any,
         *,
         content_type: str | None = None,
+        content_encoding: str | None = None,
         timeout: Timeout = 'long',
     ) -> None:
         """Set a value to the given record in the key-value store.
@@ -370,11 +371,17 @@ def set_record(
             key: The key of the record to save the value to.
             value: The value to save into the record.
             content_type: The content type of the saved value.
+            content_encoding: The encoding already applied to `value`, sent as the `Content-Encoding` header. Pass it
+                to upload a pre-compressed value - the client then forwards the bytes as they are instead of
+                compressing them itself. The API accepts `gzip`, `br`, `deflate`, and `identity`, and stores the
+                record exactly as uploaded, so this also becomes the encoding the record is served with.
             timeout: Timeout for the API HTTP request.
         """
         value, content_type = encode_key_value_store_record_value(value, content_type=content_type)
 
         headers = {'content-type': content_type}
+        if content_encoding is not None:
+            headers['content-encoding'] = content_encoding
 
         self._http_client.call(
             url=self._build_url(f'records/{key}'),
@@ -776,6 +783,7 @@ async def set_record(
         value: Any,
         *,
         content_type: str | None = None,
+        content_encoding: str | None = None,
         timeout: Timeout = 'long',
     ) -> None:
         """Set a value to the given record in the key-value store.
@@ -786,11 +794,17 @@ async def set_record(
             key: The key of the record to save the value to.
             value: The value to save into the record.
             content_type: The content type of the saved value.
+            content_encoding: The encoding already applied to `value`, sent as the `Content-Encoding` header. Pass it
+                to upload a pre-compressed value - the client then forwards the bytes as they are instead of
+                compressing them itself. The API accepts `gzip`, `br`, `deflate`, and `identity`, and stores the
+                record exactly as uploaded, so this also becomes the encoding the record is served with.
             timeout: Timeout for the API HTTP request.
         """
         value, content_type = encode_key_value_store_record_value(value, content_type=content_type)
 
         headers = {'content-type': content_type}
+        if content_encoding is not None:
+            headers['content-encoding'] = content_encoding
 
         await self._http_client.call(
             url=self._build_url(f'records/{key}'),

src/apify_client/http_clients/_base.py

@@ -170,6 +170,11 @@ def _merge_headers(base: dict[str, str] | None, override: dict[str, str] | None)
             merged[key] = value
         return merged
 
+    @staticmethod
+    def _get_header(headers: dict[str, str], name: str) -> str | None:
+        """Look up a header value by name, treated case-insensitively. Returns `None` if the header is not set."""
+        return next((value for key, value in headers.items() if key.lower() == name.lower()), None)
+
     @staticmethod
     def _parse_params(params: dict[str, Any] | None) -> dict[str, Any] | None:
         """Convert request parameters to Apify API-compatible formats.
@@ -228,9 +233,9 @@ def _compute_timeout(self, timeout: Timeout, *, attempt: int) -> int | float | N
     def _is_body_worth_compressing(data: str | bytes | bytearray | None) -> bool:
         """Whether this body clears the size threshold `_prepare_request_call` compresses at, cheaply.
 
-        Below the threshold nothing is ever compressed. At or above it the content type still decides, but
-        checking that here would buy nothing - a body that turns out to be already compressed only wastes the
-        thread hop this answer guards.
+        Below the threshold nothing is ever compressed. At or above it the content type and a caller-supplied
+        `Content-Encoding` still decide, but checking those here would buy nothing - a body that turns out to be
+        already encoded only wastes the thread hop this answer guards.
 
         The threshold is measured on encoded bytes, so a character count alone cannot decide a `str`. It is a
         lower bound, so a `str` long enough in characters is long enough in bytes too. Below that the encoded
@@ -252,12 +257,15 @@ def _prepare_request_call(
     ) -> tuple[dict[str, str], dict[str, Any] | None, bytes | None]:
         """Prepare headers, params, and body for an HTTP request.
 
-        Merges the client's default headers (including authorization) with per-request headers, serializes JSON
-        and compresses the body unless it is smaller than `MIN_COMPRESSION_SIZE` or its content type says the
-        payload is already compressed. Header names are treated case-insensitively and per-request values win
-        over the client defaults. For JSON bodies, a `Content-Type` header is set unless the caller supplied one.
-        `Content-Encoding` always describes what was actually applied to the body, so a caller-supplied value is
-        dropped whenever nothing was compressed.
+        Merges the client's default headers (including authorization) with per-request headers and serializes a
+        JSON body. Header names are treated case-insensitively and per-request values win over the client
+        defaults. For JSON bodies, a `Content-Type` header is set unless the caller supplied one.
+
+        The body is compressed unless a `Content-Encoding` header is already set, the body is smaller than
+        `MIN_COMPRESSION_SIZE`, or its content type says the payload is already compressed. A caller-supplied
+        `Content-Encoding` is forwarded verbatim, which is how a pre-encoded body is uploaded - including one in
+        an encoding the client ships no compressor for. `Content-Encoding: identity` therefore opts a single
+        request out of compression.
         """
         if json is not None and data is not None:
             raise ValueError('Cannot pass both "json" and "data" parameters at the same time!')
@@ -267,27 +275,24 @@ def _prepare_request_call(
         # Dump JSON data to a string so it can be sent as a request body.
         if json is not None:
             data = jsonlib.dumps(json, ensure_ascii=False, allow_nan=False, default=str).encode('utf-8')
-            if not any(key.lower() == 'content-type' for key in headers):
+            if self._get_header(headers, 'content-type') is None:
                 headers['Content-Type'] = 'application/json'
 
-        compressed = False
-
         if isinstance(data, (str, bytes, bytearray)):
             if isinstance(data, str):
                 data = data.encode('utf-8')
             elif isinstance(data, bytearray):
                 data = bytes(data)
 
-            content_type = next((value for key, value in headers.items() if key.lower() == 'content-type'), None)
-            if len(data) >= MIN_COMPRESSION_SIZE and is_compressible_content_type(content_type):
+            # A caller-supplied encoding says the body arrives already encoded, so compressing it here would
+            # both mislabel it and waste the work.
+            if (
+                self._get_header(headers, 'content-encoding') is None
+                and len(data) >= MIN_COMPRESSION_SIZE
+                and is_compressible_content_type(self._get_header(headers, 'content-type'))
+            ):
                 data = self._http_compressor.compress(data)
                 headers = self._merge_headers(headers, {'Content-Encoding': self._http_compressor.content_encoding})
-                compressed = True
-
-        # Anything left uncompressed goes out as-is - a file-like body included - so a caller-supplied encoding
-        # would misdescribe it.
-        if data is not None and not compressed:
-            headers = {key: value for key, value in headers.items() if key.lower() != 'content-encoding'}
 
         return (headers, self._parse_params(params), data)
 

tests/unit/test_http_clients.py

@@ -531,33 +531,18 @@ def test_prepare_request_call_skips_compression_for_already_compressed_content(c
     assert headers['User-Agent'] == client._headers['User-Agent']
 
 
-def test_prepare_request_call_drops_caller_content_encoding_when_compression_is_skipped() -> None:
-    """Skipping compression also strips a caller-supplied `Content-Encoding`, which would misdescribe the body."""
+def test_prepare_request_call_keeps_caller_content_encoding_for_a_streamed_body() -> None:
+    """A body the client streams rather than compresses, such as a file-like object, keeps its `Content-Encoding`."""
     client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor())
-    # Above the size threshold, so the content type is what skips compression here.
-    payload = b'\xff' * MIN_COMPRESSION_SIZE
-
-    headers, _params, data = client._prepare_request_call(
-        headers={'content-type': 'image/jpeg', 'content-encoding': 'br'},
-        data=payload,
-    )
-
-    assert data == payload
-    assert not any(key.lower() == 'content-encoding' for key in headers)
-
-
-def test_prepare_request_call_drops_caller_content_encoding_for_a_streamed_body() -> None:
-    """A body that is streamed rather than compressed, such as a file-like object, also loses `Content-Encoding`."""
-    client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor())
-    stream = BytesIO(b'raw payload')
+    stream = BytesIO(gzip.compress(b'raw payload'))
 
     headers, _params, data = client._prepare_request_call(
         headers={'content-encoding': 'gzip'},
         data=cast('bytes', stream),
     )
 
     assert data is stream
-    assert not any(key.lower() == 'content-encoding' for key in headers)
+    assert headers['content-encoding'] == 'gzip'
 
 
 @pytest.mark.parametrize(
@@ -645,27 +630,44 @@ def test_prepare_request_call_json_keeps_caller_content_type() -> None:
     assert content_type_headers == {'content-type': 'application/json; charset=utf-8'}
 
 
-def test_prepare_request_call_replaces_caller_content_encoding() -> None:
-    """A compressed body reports the compressor actually applied, replacing any caller-supplied Content-Encoding."""
+@pytest.mark.parametrize(
+    ('caller_headers', 'body'),
+    [
+        pytest.param({'content-encoding': 'br'}, b'x' * MIN_COMPRESSION_SIZE, id='body the client would compress'),
+        pytest.param({'content-encoding': 'br'}, b'payload', id='body below the size threshold'),
+        pytest.param(
+            {'content-encoding': 'br', 'content-type': 'image/jpeg'},
+            b'\xff' * MIN_COMPRESSION_SIZE,
+            id='already-compressed content type',
+        ),
+        pytest.param({'content-encoding': 'identity'}, b'x' * MIN_COMPRESSION_SIZE, id='identity opt-out'),
+        pytest.param(
+            {'content-encoding': 'deflate'},
+            b'x' * MIN_COMPRESSION_SIZE,
+            id='encoding the client has no compressor for',
+        ),
+    ],
+)
+def test_prepare_request_call_keeps_caller_content_encoding(caller_headers: dict[str, str], body: bytes) -> None:
+    """A caller-supplied `Content-Encoding` marks the body as pre-encoded, so it goes out untouched and labeled."""
     client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor())
 
-    headers, _params, _data = client._prepare_request_call(
-        headers={'content-encoding': 'br'},
-        data='x' * MIN_COMPRESSION_SIZE,
-    )
+    headers, _params, data = client._prepare_request_call(headers=caller_headers, data=body)
 
+    assert data == body
     encoding_headers = {key: value for key, value in headers.items() if key.lower() == 'content-encoding'}
-    assert encoding_headers == {'Content-Encoding': 'gzip'}
+    assert encoding_headers == {'content-encoding': caller_headers['content-encoding']}
 
 
-def test_prepare_request_call_drops_caller_content_encoding_when_skipping_compression() -> None:
-    """A caller-supplied Content-Encoding is dropped for an uncompressed body, so it cannot mislabel it."""
-    client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor())
+def test_prepare_request_call_keeps_client_wide_content_encoding() -> None:
+    """A `Content-Encoding` configured on the client counts as caller-supplied on every request it sends."""
+    client = _ConcreteHttpClient(headers={'Content-Encoding': 'identity'}, http_compressor=GzipHttpCompressor())
+    body = b'x' * MIN_COMPRESSION_SIZE
 
-    headers, _params, data = client._prepare_request_call(headers={'content-encoding': 'br'}, data='payload')
+    headers, _params, data = client._prepare_request_call(data=body)
 
-    assert data == b'payload'
-    assert not any(key.lower() == 'content-encoding' for key in headers)
+    assert data == body
+    assert headers['Content-Encoding'] == 'identity'
 
 
 def test_build_url_with_params_none() -> None: