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

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
 
@@ -45,7 +47,28 @@ 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 streamed rather than buffered, so they're never compressed regardless of their content type.
+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.
+
+## 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.
+
+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.
 
 ## Configuration
 
@@ -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,23 @@ 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. Only a
+                bytes-like `value`, or a file-like one that reads into bytes, can carry a compression - anything
+                else raises `TypeError` instead of being stored under a header that misdescribes it.
             timeout: Timeout for the API HTTP request.
         """
-        value, content_type = encode_key_value_store_record_value(value, content_type=content_type)
+        value, content_type = encode_key_value_store_record_value(
+            value,
+            content_type=content_type,
+            content_encoding=content_encoding,
+        )
 
         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 +789,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 +800,23 @@ 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. Only a
+                bytes-like `value`, or a file-like one that reads into bytes, can carry a compression - anything
+                else raises `TypeError` instead of being stored under a header that misdescribes it.
             timeout: Timeout for the API HTTP request.
         """
-        value, content_type = encode_key_value_store_record_value(value, content_type=content_type)
+        value, content_type = encode_key_value_store_record_value(
+            value,
+            content_type=content_type,
+            content_encoding=content_encoding,
+        )
 
         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/_utils/encoding.py

@@ -13,7 +13,7 @@
 
 
 def encode_key_value_store_record_value(
-    value: Any, *, content_type: str | None = None
+    value: Any, *, content_type: str | None = None, content_encoding: str | None = None
 ) -> tuple[bytes | bytearray | str, str]:
     """Encode a value for storage in a key-value store record.
 
@@ -23,12 +23,17 @@ def encode_key_value_store_record_value(
             memory whole - the object is neither rewound nor closed, and async file-like objects are rejected.
             Any other value is JSON-serialized unless it is already bytes or a string.
         content_type: The content type; if None, it's inferred from the value type.
+        content_encoding: The encoding the caller declares the value already carries, if any. Anything other than
+            `identity` means the value is compressed, which only a bytes-like payload can be, so any other value
+            is rejected. The check belongs here because a file-like value has to be read before its payload type
+            is known, and reading it a second time in the caller is not possible.
 
     Returns:
         A tuple of (encoded_value, content_type).
 
     Raises:
-        TypeError: If the value cannot be encoded into a body the transport accepts.
+        TypeError: If the value cannot be encoded into a body the transport accepts, or if it cannot be carrying
+            the declared `content_encoding`.
     """
     # Read file-like values into memory; the transport only accepts bytes-like bodies. Detect them by a
     # callable `read` (not `io.IOBase`) so duck-typed file-likes are read, not JSON-serialized. Impit exposes
@@ -48,6 +53,17 @@ def encode_key_value_store_record_value(
         if not isinstance(value, (bytes, bytearray, str)):
             raise TypeError(f'Reading the file-like value returned {type(value).__name__}, expected bytes or str.')
 
+    # A declared compression describes bytes the caller compressed. A string, a JSON-serializable object, or a
+    # text-mode file cannot be carrying one, and would otherwise be stored under a header that misdescribes it -
+    # the client forwards the header untouched and never inspects the body.
+    declared_encoding = (content_encoding or '').strip().lower()
+    if declared_encoding not in ('', 'identity') and not isinstance(value, (bytes, bytearray)):
+        raise TypeError(
+            f'Cannot upload a {type(value).__name__} value with `Content-Encoding: {content_encoding}`. An encoding '
+            'other than `identity` declares the value is already compressed, so pass the compressed bytes, or a '
+            'file-like object that reads them.'
+        )
+
     if not content_type:
         if isinstance(value, (bytes, bytearray)):
             content_type = 'application/octet-stream'

src/apify_client/http_clients/_base.py

@@ -153,7 +153,7 @@ def set_default_authorization(self, token: str) -> None:
         Args:
             token: The Apify API token to set as the `Bearer` authorization.
         """
-        if not any(key.lower() == 'authorization' for key in self._headers):
+        if self._get_header(self._headers, 'authorization') is None:
             self._headers['Authorization'] = f'Bearer {token}'
 
     @staticmethod
@@ -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)