Repository: boto/botocore
Pull request #2959 (2023-05-26 19:09:24)

request compression
This pull request enables request compression in the AWS Python SDK. It is on by default, but only used if the `requestcompression` trait is attached to the operation being called. At the moment, the only compression algorithm supported is `gzip`, but this is subject to change in the future. This also addresses a few typos I discovered while working on this project. Below is a summary of the new features in this PR.

* Add `request_min_compression_size_bytes` setting to `Config` object and config file, and support for `AWS_REQUEST_MIN_COMPRESSION_SIZE_BYTES` environment variable.
    * If not provided by a user, the value will default to `10240`. The value MUST be a parseable integer between 0 and 1048576 (1024**2).
* Add `disable_request_compression` setting to `Config` object and config file, and support for `AWS_DISABLE_REQUEST_COMPRESSION` environment variable.
    * If not provided by a user, the value will default to `False`.  If not a parseable boolean, the value will be coerced to `False`, similar to other config settings that are boolean types in botocore.
* Add `request_compression` property to `OperationModel`. This will allow the SDK to check for the presence of this operation trait and run compression if so. 
* Add the `compress.py` file. This file contains logic that is actually responsible for compressing requests. It has one public API:
    * `maybe_compress_request`: This checks whether compression should be performed, actually does compression if so and injects the corresponding `Content-Encoding` header if needed.
* Compression is supported for all types of bodies the SDK already supports: `str`, `bytes`, `bytearray`, and file-like objects like `io.StringIO` and `io.BytesIO` for example.
* Compression is called just before flexible checksums are calculated in `BaseClient._make_api_call`
* Besides the above new settings, compression will also not occur if a customer has configured `signature_version` to be `v2`
* In order to support compression for services using the `query` protocol, dictionary bodies are serialized to bytes if the operation supports compression and the customer hasn't explicitly disabled compression.

codecov-commenter: ## [Codecov](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) Report
Patch coverage: **`95.34`**% and project coverage change: **`-0.04`** :warning:
> Comparison is base [(`0e9d748`)](https://app.codecov.io/gh/boto/botocore/commit/0e9d7486d53f0630bf463010900e0d99da544c7f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) 93.29% compared to head [(`cc12a2e`)](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) 93.25%.

:exclamation: Your organization is not using the GitHub App Integration. As a result you may experience degraded service beginning May 15th. Please [install the Github App Integration](https://github.com/apps/codecov) for your organization. [Read more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto).

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           develop    #2959      +/-   ##
===========================================
- Coverage    93.29%   93.25%   -0.04%     
===========================================
  Files           64       64              
  Lines        13592    13675      +83     
===========================================
+ Hits         12680    12752      +72     
- Misses         912      923      +11     
```


| [Impacted Files](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) | Coverage Δ | |
|---|---|---|
| [botocore/config.py](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvY29uZmlnLnB5) | `100.00% <ø> (ø)` | |
| [botocore/awsrequest.py](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvYXdzcmVxdWVzdC5weQ==) | `97.56% <92.45%> (-0.99%)` | :arrow_down: |
| [botocore/args.py](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvYXJncy5weQ==) | `100.00% <100.00%> (ø)` | |
| [botocore/configprovider.py](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvY29uZmlncHJvdmlkZXIucHk=) | `94.09% <100.00%> (ø)` | |
| [botocore/endpoint.py](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvZW5kcG9pbnQucHk=) | `97.83% <100.00%> (+0.02%)` | :arrow_up: |
| [botocore/exceptions.py](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvZXhjZXB0aW9ucy5weQ==) | `99.57% <100.00%> (+<0.01%)` | :arrow_up: |
| [botocore/model.py](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvbW9kZWwucHk=) | `98.38% <100.00%> (+0.01%)` | :arrow_up: |
| [botocore/utils.py](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvdXRpbHMucHk=) | `79.31% <100.00%> (-0.04%)` | :arrow_down: |

... and [3 files with indirect coverage changes](https://app.codecov.io/gh/boto/botocore/pull/2959/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/boto/botocore/pull/2959?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto).

nateprewitt: One minor comment, but once tests pass, I think this should be set.

Patch:
diff --git a/.changes/next-release/enhancement-compression-36791.json b/.changes/next-release/enhancement-compression-36791.json
new file mode 100644
index 000000000..ea56ede0e
--- /dev/null
+++ b/.changes/next-release/enhancement-compression-36791.json
@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "compression",
+  "description": "Adds support for the ``requestcompression`` operation trait."
+}
diff --git a/botocore/args.py b/botocore/args.py
index 73c8ab45e..83e6b1e27 100644
--- a/botocore/args.py
+++ b/botocore/args.py
@@ -260,10 +260,17 @@ class ClientArgsCreator:
                 tcp_keepalive=client_config.tcp_keepalive,
                 user_agent_extra=client_config.user_agent_extra,
                 user_agent_appid=client_config.user_agent_appid,
+                request_min_compression_size_bytes=(
+                    client_config.request_min_compression_size_bytes
+                ),
+                disable_request_compression=(
+                    client_config.disable_request_compression
+                ),
             )
         self._compute_retry_config(config_kwargs)
         self._compute_connect_timeout(config_kwargs)
         self._compute_user_agent_appid_config(config_kwargs)
+        self._compute_request_compression_config(config_kwargs)
         s3_config = self.compute_s3_config(client_config)
 
         is_s3_service = self._is_s3_service(service_name)
@@ -543,6 +550,53 @@ class ClientArgsCreator:
         if connect_timeout:
             config_kwargs['connect_timeout'] = connect_timeout
 
+    def _compute_request_compression_config(self, config_kwargs):
+        min_size = config_kwargs.get('request_min_compression_size_bytes')
+        disabled = config_kwargs.get('disable_request_compression')
+        if min_size is None:
+            min_size = self._config_store.get_config_variable(
+                'request_min_compression_size_bytes'
+            )
+        # conversion func is skipped so input validation must be done here
+        # regardless if the value is coming from the config store or the
+        # config object
+        min_size = self._validate_min_compression_size(min_size)
+        config_kwargs['request_min_compression_size_bytes'] = min_size
+
+        if disabled is None:
+            disabled = self._config_store.get_config_variable(
+                'disable_request_compression'
+            )
+        else:
+            # if the user provided a value we must check if it's a boolean
+            disabled = ensure_boolean(disabled)
+        config_kwargs['disable_request_compression'] = disabled
+
+    def _validate_min_compression_size(self, min_size):
+        min_allowed_min_size = 1
+        max_allowed_min_size = 1048576
+        if min_size is not None:
+            error_msg_base = (
+                f'Invalid value "{min_size}" for '
+                'request_min_compression_size_bytes.'
+            )
+            try:
+                min_size = int(min_size)
+            except (ValueError, TypeError):
+                msg = (
+                    f'{error_msg_base} Value must be an integer. '
+                    f'Received {type(min_size)} instead.'
+                )
+                raise botocore.exceptions.InvalidConfigError(error_msg=msg)
+            if not min_allowed_min_size <= min_size <= max_allowed_min_size:
+                msg = (
+                    f'{error_msg_base} Value must be between '
+                    f'{min_allowed_min_size} and {max_allowed_min_size}.'
+                )
+                raise botocore.exceptions.InvalidConfigError(error_msg=msg)
+
+        return min_size
+
     def _ensure_boolean(self, val):
         if isinstance(val, bool):
             return val
diff --git a/botocore/client.py b/botocore/client.py
index ec34bf7c0..9f202029b 100644
--- a/botocore/client.py
+++ b/botocore/client.py
@@ -16,6 +16,7 @@ from botocore import waiter, xform_name
 from botocore.args import ClientArgsCreator
 from botocore.auth import AUTH_TYPE_MAPS
 from botocore.awsrequest import prepare_request_dict
+from botocore.compress import maybe_compress_request
 from botocore.config import Config
 from botocore.discovery import (
     EndpointDiscoveryHandler,
@@ -955,6 +956,9 @@ class BaseClient:
         if event_response is not None:
             http, parsed_response = event_response
         else:
+            maybe_compress_request(
+                self.meta.config, request_dict, operation_model
+            )
             apply_request_checksum(request_dict)
             http, parsed_response = self._make_request(
                 operation_model, request_dict, request_context
diff --git a/botocore/compress.py b/botocore/compress.py
new file mode 100644
index 000000000..1f8577e84
--- /dev/null
+++ b/botocore/compress.py
@@ -0,0 +1,126 @@
+# Copyright 2023 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+"""
+NOTE: All functions in this module are considered private and are
+subject to abrupt breaking changes. Please do not use them directly.
+
+"""
+
+import io
+import logging
+from gzip import GzipFile
+from gzip import compress as gzip_compress
+
+from botocore.compat import urlencode
+from botocore.utils import determine_content_length
+
+logger = logging.getLogger(__name__)
+
+
+def maybe_compress_request(config, request_dict, operation_model):
+    """Attempt to compress the request body using the modeled encodings."""
+    if _should_compress_request(config, request_dict, operation_model):
+        for encoding in operation_model.request_compression['encodings']:
+            encoder = COMPRESSION_MAPPING.get(encoding)
+            if encoder is not None:
+                logger.debug('Compressing request with %s encoding.', encoding)
+                request_dict['body'] = encoder(request_dict['body'])
+                _set_compression_header(request_dict['headers'], encoding)
+                return
+            else:
+                logger.debug('Unsupported compression encoding: %s', encoding)
+
+
+def _should_compress_request(config, request_dict, operation_model):
+    if (
+        config.disable_request_compression is not True
+        and config.signature_version != 'v2'
+        and operation_model.request_compression is not None
+    ):
+        if not _is_compressible_type(request_dict):
+            body_type = type(request_dict['body'])
+            log_msg = 'Body type %s does not support compression.'
+            logger.debug(log_msg, body_type)
+            return False
+
+        if operation_model.has_streaming_input:
+            streaming_input = operation_model.get_streaming_input()
+            streaming_metadata = streaming_input.metadata
+            return 'requiresLength' not in streaming_metadata
+
+        body_size = _get_body_size(request_dict['body'])
+        min_size = config.request_min_compression_size_bytes
+        return min_size <= body_size
+
+    return False
+
+
+def _is_compressible_type(request_dict):
+    body = request_dict['body']
+    # Coerce dict to a format compatible with compression.
+    if isinstance(body, dict):
+        body = urlencode(body, doseq=True, encoding='utf-8').encode('utf-8')
+        request_dict['body'] = body
+    is_supported_type = isinstance(body, (str, bytes, bytearray))
+    return is_supported_type or hasattr(body, 'read')
+
+
+def _get_body_size(body):
+    size = determine_content_length(body)
+    if size is None:
+        logger.debug(
+            'Unable to get length of the request body: %s. '
+            'Skipping compression.',
+            body,
+        )
+        size = 0
+    return size
+
+
+def _gzip_compress_body(body):
+    if isinstance(body, str):
+        return gzip_compress(body.encode('utf-8'))
+    elif isinstance(body, (bytes, bytearray)):
+        return gzip_compress(body)
+    elif hasattr(body, 'read'):
+        if hasattr(body, 'seek') and hasattr(body, 'tell'):
+            current_position = body.tell()
+            compressed_obj = _gzip_compress_fileobj(body)
+            body.seek(current_position)
+            return compressed_obj
+        return _gzip_compress_fileobj(body)
+
+
+def _gzip_compress_fileobj(body):
+    compressed_obj = io.BytesIO()
+    with GzipFile(fileobj=compressed_obj, mode='wb') as gz:
+        while True:
+            chunk = body.read(8192)
+            if not chunk:
+                break
+            if isinstance(chunk, str):
+                chunk = chunk.encode('utf-8')
+            gz.write(chunk)
+    compressed_obj.seek(0)
+    return compressed_obj
+
+
+def _set_compression_header(headers, encoding):
+    ce_header = headers.get('Content-Encoding')
+    if ce_header is None:
+        headers['Content-Encoding'] = encoding
+    else:
+        headers['Content-Encoding'] = f'{ce_header},{encoding}'
+
+
+COMPRESSION_MAPPING = {'gzip': _gzip_compress_body}
diff --git a/botocore/config.py b/botocore/config.py
index be3a475fa..a968eacc9 100644
--- a/botocore/config.py
+++ b/botocore/config.py
@@ -206,6 +206,20 @@ class Config:
         creating new connections if set to True.
 
         Defaults to False.
+
+    :type request_min_compression_size_bytes: int
+    :param request_min_compression_bytes: The minimum size in bytes that a
+    request body should be to trigger compression. All requests with streaming
+    input that don't contain the `requiresLength` trait will be compressed
+    regardless of this setting.
+
+        Defaults to None.
+
+    :type disable_request_compression: bool
+    :param disable_request_compression: Disables request body compression if
+    set to True.
+
+        Defaults to None.
     """
 
     OPTION_DEFAULTS = OrderedDict(
@@ -231,6 +245,8 @@ class Config:
             ('ignore_configured_endpoint_urls', None),
             ('defaults_mode', None),
             ('tcp_keepalive', None),
+            ('request_min_compression_size_bytes', None),
+            ('disable_request_compression', None),
         ]
     )
 
diff --git a/botocore/configprovider.py b/botocore/configprovider.py
index d7b2e19de..1f9155e32 100644
--- a/botocore/configprovider.py
+++ b/botocore/configprovider.py
@@ -10,7 +10,7 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
-"""This module contains the inteface for controlling how configuration
+"""This module contains the interface for controlling how configuration
 is loaded.
 """
 import copy
@@ -147,6 +147,18 @@ BOTOCORE_DEFAUT_SESSION_VARIABLES = {
     # whatever the defaults are in _retry.json.
     'max_attempts': ('max_attempts', 'AWS_MAX_ATTEMPTS', None, int),
     'user_agent_appid': ('sdk_ua_app_id', 'AWS_SDK_UA_APP_ID', None, None),
+    'request_min_compression_size_bytes': (
+        'request_min_compression_size_bytes',
+        'AWS_REQUEST_MIN_COMPRESSION_SIZE_BYTES',
+        10240,
+        None,
+    ),
+    'disable_request_compression': (
+        'disable_request_compression',
+        'AWS_DISABLE_REQUEST_COMPRESSION',
+        False,
+        utils.ensure_boolean,
+    ),
 }
 # A mapping for the s3 specific configuration vars. These are the configuration
 # vars that typically go in the s3 section of the config file. This mapping
diff --git a/botocore/model.py b/botocore/model.py
index ea888ce1b..dec3e95eb 100644
--- a/botocore/model.py
+++ b/botocore/model.py
@@ -87,6 +87,7 @@ class Shape:
         'union',
         'contextParam',
         'clientContextParams',
+        'requiresLength',
     ]
     MAP_TYPE = OrderedDict
 
@@ -172,6 +173,9 @@ class Shape:
             * idempotencyToken
             * document
             * union
+            * contextParam
+            * clientContextParams
+            * requiresLength
 
         :rtype: dict
         :return: Metadata about the shape.
@@ -614,6 +618,10 @@ class OperationModel:
             and 'name' in shape.metadata['contextParam']
         ]
 
+    @CachedProperty
+    def request_compression(self):
+        return self._operation_model.get('requestcompression')
+
     @CachedProperty
     def auth_type(self):
         return self._operation_model.get('authtype')


Test patch:
diff --git a/tests/functional/test_compress.py b/tests/functional/test_compress.py
new file mode 100644
index 000000000..63f1510b1
--- /dev/null
+++ b/tests/functional/test_compress.py
@@ -0,0 +1,129 @@
+# Copyright 2023 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+import gzip
+
+import pytest
+
+from botocore.compress import COMPRESSION_MAPPING
+from botocore.config import Config
+from tests import ALL_SERVICES, ClientHTTPStubber, patch_load_service_model
+
+FAKE_MODEL = {
+    "version": "2.0",
+    "documentation": "",
+    "metadata": {
+        "apiVersion": "2020-02-02",
+        "endpointPrefix": "otherservice",
+        "protocol": "query",
+        "serviceFullName": "Other Service",
+        "serviceId": "Other Service",
+        "signatureVersion": "v4",
+        "signingName": "otherservice",
+        "uid": "otherservice-2020-02-02",
+    },
+    "operations": {
+        "MockOperation": {
+            "name": "MockOperation",
+            "http": {"method": "POST", "requestUri": "/"},
+            "input": {"shape": "MockOperationRequest"},
+            "documentation": "",
+            "requestcompression": {
+                "encodings": ["gzip"],
+            },
+        },
+    },
+    "shapes": {
+        "MockOpParamList": {
+            "type": "list",
+            "member": {"shape": "MockOpParam"},
+        },
+        "MockOpParam": {
+            "type": "structure",
+            "members": {"MockOpParam": {"shape": "MockOpParamValue"}},
+        },
+        "MockOpParamValue": {
+            "type": "string",
+        },
+        "MockOperationRequest": {
+            "type": "structure",
+            "required": ["MockOpParamList"],
+            "members": {
+                "MockOpParamList": {
+                    "shape": "MockOpParamList",
+                    "documentation": "",
+                },
+            },
+        },
+    },
+}
+
+FAKE_RULESET = {
+    "version": "1.0",
+    "parameters": {},
+    "rules": [
+        {
+            "conditions": [],
+            "type": "endpoint",
+            "endpoint": {
+                "url": "https://foo.bar",
+                "properties": {},
+                "headers": {},
+            },
+        }
+    ],
+}
+
+
+def _all_compression_operations():
+    for service_model in ALL_SERVICES:
+        for operation_name in service_model.operation_names:
+            operation_model = service_model.operation_model(operation_name)
+            if operation_model.request_compression is not None:
+                yield operation_model
+
+
+@pytest.mark.parametrize("operation_model", _all_compression_operations())
+def test_no_unknown_compression_encodings(operation_model):
+    for encoding in operation_model.request_compression["encodings"]:
+        assert encoding in COMPRESSION_MAPPING.keys(), (
+            f"Found unknown compression encoding '{encoding}' "
+            f"in operation {operation_model.name}."
+        )
+
+
+def test_compression(patched_session, monkeypatch):
+    patch_load_service_model(
+        patched_session, monkeypatch, FAKE_MODEL, FAKE_RULESET
+    )
+    client = patched_session.create_client(
+        "otherservice",
+        region_name="us-west-2",
+        config=Config(request_min_compression_size_bytes=100),
+    )
+    with ClientHTTPStubber(client, strict=True) as http_stubber:
+        http_stubber.add_response(status=200, body=b"<response/>")
+        params_list = [
+            {"MockOpParam": f"MockOpParamValue{i}"} for i in range(1, 21)
+        ]
+        client.mock_operation(MockOpParamList=params_list)
+        param_template = (
+            "MockOpParamList.member.{i}.MockOpParam=MockOpParamValue{i}"
+        )
+        serialized_params = "&".join(
+            param_template.format(i=i) for i in range(1, 21)
+        )
+        additional_params = "Action=MockOperation&Version=2020-02-02"
+        serialized_body = f"{additional_params}&{serialized_params}"
+        actual_body = gzip.decompress(http_stubber.requests[0].body)
+        assert serialized_body.encode('utf-8') == actual_body
diff --git a/tests/unit/test_args.py b/tests/unit/test_args.py
index e82226b00..8f1c99242 100644
--- a/tests/unit/test_args.py
+++ b/tests/unit/test_args.py
@@ -529,6 +529,91 @@ class TestCreateClientArgs(unittest.TestCase):
             )
             m.assert_not_called()
 
+    def test_request_compression_client_config(self):
+        input_config = Config(
+            disable_request_compression=True,
+            request_min_compression_size_bytes=100,
+        )
+        client_args = self.call_get_client_args(client_config=input_config)
+        config = client_args['client_config']
+        self.assertEqual(config.request_min_compression_size_bytes, 100)
+        self.assertTrue(config.disable_request_compression)
+
+    def test_request_compression_config_store(self):
+        self.config_store.set_config_variable(
+            'request_min_compression_size_bytes', 100
+        )
+        self.config_store.set_config_variable(
+            'disable_request_compression', True
+        )
+        config = self.call_get_client_args()['client_config']
+        self.assertEqual(config.request_min_compression_size_bytes, 100)
+        self.assertTrue(config.disable_request_compression)
+
+    def test_request_compression_client_config_overrides_config_store(self):
+        self.config_store.set_config_variable(
+            'request_min_compression_size_bytes', 100
+        )
+        self.config_store.set_config_variable(
+            'disable_request_compression', True
+        )
+        input_config = Config(
+            disable_request_compression=False,
+            request_min_compression_size_bytes=1,
+        )
+        client_args = self.call_get_client_args(client_config=input_config)
+        config = client_args['client_config']
+        self.assertEqual(config.request_min_compression_size_bytes, 1)
+        self.assertFalse(config.disable_request_compression)
+
+    def test_coercible_value_request_min_compression_size_bytes(self):
+        config = Config(request_min_compression_size_bytes='100')
+        client_args = self.call_get_client_args(client_config=config)
+        config = client_args['client_config']
+        self.assertEqual(config.request_min_compression_size_bytes, 100)
+
+    def test_coercible_value_disable_request_compression(self):
+        config = Config(disable_request_compression='true')
+        client_args = self.call_get_client_args(client_config=config)
+        config = client_args['client_config']
+        self.assertTrue(config.disable_request_compression)
+
+    def test_bad_type_request_min_compression_size_bytes(self):
+        with self.assertRaises(exceptions.InvalidConfigError):
+            config = Config(request_min_compression_size_bytes='foo')
+            self.call_get_client_args(client_config=config)
+        self.config_store.set_config_variable(
+            'request_min_compression_size_bytes', 'foo'
+        )
+        with self.assertRaises(exceptions.InvalidConfigError):
+            self.call_get_client_args()
+
+    def test_low_min_request_min_compression_size_bytes(self):
+        with self.assertRaises(exceptions.InvalidConfigError):
+            config = Config(request_min_compression_size_bytes=0)
+            self.call_get_client_args(client_config=config)
+        self.config_store.set_config_variable(
+            'request_min_compression_size_bytes', 0
+        )
+        with self.assertRaises(exceptions.InvalidConfigError):
+            self.call_get_client_args()
+
+    def test_high_max_request_min_compression_size_bytes(self):
+        with self.assertRaises(exceptions.InvalidConfigError):
+            config = Config(request_min_compression_size_bytes=9999999)
+            self.call_get_client_args(client_config=config)
+        self.config_store.set_config_variable(
+            'request_min_compression_size_bytes', 9999999
+        )
+        with self.assertRaises(exceptions.InvalidConfigError):
+            self.call_get_client_args()
+
+    def test_bad_value_disable_request_compression(self):
+        input_config = Config(disable_request_compression='foo')
+        client_args = self.call_get_client_args(client_config=input_config)
+        config = client_args['client_config']
+        self.assertFalse(config.disable_request_compression)
+
 
 class TestEndpointResolverBuiltins(unittest.TestCase):
     def setUp(self):
diff --git a/tests/unit/test_compress.py b/tests/unit/test_compress.py
new file mode 100644
index 000000000..c149d7c6b
--- /dev/null
+++ b/tests/unit/test_compress.py
@@ -0,0 +1,322 @@
+# Copyright 2023 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+import gzip
+import io
+import sys
+
+import pytest
+
+import botocore
+from botocore.compress import COMPRESSION_MAPPING, maybe_compress_request
+from botocore.config import Config
+from tests import mock
+
+
+def _make_op(
+    request_compression=None,
+    has_streaming_input=False,
+    streaming_metadata=None,
+):
+    op = mock.Mock()
+    op.request_compression = request_compression
+    op.has_streaming_input = has_streaming_input
+    if streaming_metadata is not None:
+        streaming_shape = mock.Mock()
+        streaming_shape.metadata = streaming_metadata
+        op.get_streaming_input.return_value = streaming_shape
+    return op
+
+
+OP_NO_COMPRESSION = _make_op()
+OP_WITH_COMPRESSION = _make_op({'encodings': ['gzip']})
+OP_UNKNOWN_COMPRESSION = _make_op({'encodings': ['foo']})
+OP_MULTIPLE_COMPRESSIONS = _make_op({'encodings': ['gzip', 'foo']})
+STREAMING_OP_WITH_COMPRESSION = _make_op(
+    {'encodings': ['gzip']},
+    True,
+    {},
+)
+STREAMING_OP_WITH_COMPRESSION_REQUIRES_LENGTH = _make_op(
+    {'encodings': ['gzip']},
+    True,
+    {'requiresLength': True},
+)
+
+
+REQUEST_BODY = (
+    b'Action=PutMetricData&Version=2010-08-01&Namespace=Namespace'
+    b'&MetricData.member.1.MetricName=metric&MetricData.member.1.Unit=Bytes'
+    b'&MetricData.member.1.Value=128'
+)
+REQUEST_BODY_COMPRESSED = (
+    b'\x1f\x8b\x08\x00\x01\x00\x00\x00\x02\xffsL.\xc9\xcc\xcf\xb3\r(-\xf1M-)'
+    b'\xcaLvI,IT\x0bK-*\x06\x89\x1a\x19\x18\x1a\xe8\x1aX\xe8\x1a\x18\xaa\xf9%'
+    b'\xe6\xa6\x16\x17$&\xa7\xda\xc2Yj\x08\x1dz\xb9\xa9\xb9I\xa9Ez\x86z\x101\x90'
+    b'\x1a\xdb\\0\x13\xab\xaa\xd0\xbc\xcc\x12[\xa7\xca\x92\xd4b\xac\xd2a\x899\xa5'
+    b'\xa9\xb6\x86F\x16\x00\x1e\xdd\t\xfd\x9e\x00\x00\x00'
+)
+
+
+COMPRESSION_CONFIG_128_BYTES = Config(
+    disable_request_compression=False,
+    request_min_compression_size_bytes=128,
+)
+COMPRESSION_CONFIG_1_BYTE = Config(
+    disable_request_compression=False,
+    request_min_compression_size_bytes=1,
+)
+
+
+class NonSeekableStream:
+    def __init__(self, buffer):
+        self._buffer = buffer
+
+    def read(self, size=None):
+        return self._buffer.read(size)
+
+
+def _request_dict(body=REQUEST_BODY, headers=None):
+    if headers is None:
+        headers = {}
+
+    return {
+        'body': body,
+        'headers': headers,
+    }
+
+
+def request_dict_non_seekable_text_stream():
+    stream = NonSeekableStream(io.StringIO(REQUEST_BODY.decode('utf-8')))
+    return _request_dict(stream)
+
+
+def request_dict_non_seekable_bytes_stream():
+    return _request_dict(NonSeekableStream(io.BytesIO(REQUEST_BODY)))
+
+
+class StaticGzipFile(gzip.GzipFile):
+    def __init__(self, *args, **kwargs):
+        kwargs['mtime'] = 1
+        super().__init__(*args, **kwargs)
+
+
+def static_compress(*args, **kwargs):
+    kwargs['mtime'] = 1
+    return gzip.compress(*args, **kwargs)
+
+
+def _bad_compression(body):
+    raise ValueError('Reached unintended compression algorithm "foo"')
+
+
+MOCK_COMPRESSION = {'foo': _bad_compression}
+MOCK_COMPRESSION.update(COMPRESSION_MAPPING)
+
+
+def _assert_compression_body(compressed_body, expected_body):
+    data = compressed_body
+    if hasattr(compressed_body, 'read'):
+        data = compressed_body.read()
+    assert data == expected_body
+
+
+def _assert_compression_header(headers, encoding='gzip'):
+    assert 'Content-Encoding' in headers
+    assert encoding in headers['Content-Encoding']
+
+
+def assert_request_compressed(request_dict, expected_body):
+    _assert_compression_body(request_dict['body'], expected_body)
+    _assert_compression_header(request_dict['headers'])
+
+
+@pytest.mark.parametrize(
+    'request_dict, operation_model',
+    [
+        (
+            _request_dict(),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            _request_dict(),
+            OP_MULTIPLE_COMPRESSIONS,
+        ),
+        (
+            _request_dict(),
+            STREAMING_OP_WITH_COMPRESSION,
+        ),
+        (
+            _request_dict(bytearray(REQUEST_BODY)),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            _request_dict(headers={'Content-Encoding': 'identity'}),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            _request_dict(REQUEST_BODY.decode('utf-8')),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            _request_dict(io.BytesIO(REQUEST_BODY)),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            _request_dict(io.StringIO(REQUEST_BODY.decode('utf-8'))),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            request_dict_non_seekable_bytes_stream(),
+            STREAMING_OP_WITH_COMPRESSION,
+        ),
+        (
+            request_dict_non_seekable_text_stream(),
+            STREAMING_OP_WITH_COMPRESSION,
+        ),
+    ],
+)
+@mock.patch.object(botocore.compress, 'GzipFile', StaticGzipFile)
+@mock.patch.object(botocore.compress, 'gzip_compress', static_compress)
+@pytest.mark.skipif(
+    sys.version_info < (3, 8), reason='requires python3.8 or higher'
+)
+def test_compression(request_dict, operation_model):
+    maybe_compress_request(
+        COMPRESSION_CONFIG_128_BYTES, request_dict, operation_model
+    )
+    assert_request_compressed(request_dict, REQUEST_BODY_COMPRESSED)
+
+
+@pytest.mark.parametrize(
+    'config, request_dict, operation_model',
+    [
+        (
+            Config(
+                disable_request_compression=True,
+                request_min_compression_size_bytes=1,
+            ),
+            _request_dict(),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            Config(
+                disable_request_compression=False,
+                request_min_compression_size_bytes=256,
+            ),
+            _request_dict(),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            Config(
+                disable_request_compression=False,
+                request_min_compression_size_bytes=1,
+                signature_version='v2',
+            ),
+            _request_dict(),
+            OP_WITH_COMPRESSION,
+        ),
+        (
+            COMPRESSION_CONFIG_128_BYTES,
+            _request_dict(),
+            STREAMING_OP_WITH_COMPRESSION_REQUIRES_LENGTH,
+        ),
+        (
+            COMPRESSION_CONFIG_128_BYTES,
+            _request_dict(),
+            OP_NO_COMPRESSION,
+        ),
+        (
+            COMPRESSION_CONFIG_128_BYTES,
+            _request_dict(),
+            OP_UNKNOWN_COMPRESSION,
+        ),
+        (
+            COMPRESSION_CONFIG_128_BYTES,
+            _request_dict(headers={'Content-Encoding': 'identity'}),
+            OP_UNKNOWN_COMPRESSION,
+        ),
+        (
+            COMPRESSION_CONFIG_128_BYTES,
+            request_dict_non_seekable_bytes_stream(),
+            OP_WITH_COMPRESSION,
+        ),
+    ],
+)
+def test_no_compression(config, request_dict, operation_model):
+    ce_header = request_dict['headers'].get('Content-Encoding')
+    original_body = request_dict['body']
+    maybe_compress_request(config, request_dict, operation_model)
+    assert request_dict['body'] == original_body
+    assert ce_header == request_dict['headers'].get('Content-Encoding')
+
+
+@pytest.mark.parametrize(
+    'operation_model, expected_body',
+    [
+        (
+            OP_WITH_COMPRESSION,
+            (
+                b'\x1f\x8b\x08\x00\x01\x00\x00\x00\x02\xffK\xcb'
+                b'\xcf\xb7MJ,\x02\x00v\x8e5\x1c\x07\x00\x00\x00'
+            ),
+        ),
+        (OP_NO_COMPRESSION, {'foo': 'bar'}),
+    ],
+)
+@mock.patch.object(botocore.compress, 'gzip_compress', static_compress)
+@pytest.mark.skipif(
+    sys.version_info < (3, 8), reason='requires python3.8 or higher'
+)
+def test_dict_compression(operation_model, expected_body):
+    request_dict = _request_dict({'foo': 'bar'})
+    maybe_compress_request(
+        COMPRESSION_CONFIG_1_BYTE, request_dict, operation_model
+    )
+    body = request_dict['body']
+    assert body == expected_body
+
+
+@pytest.mark.parametrize('body', [1, object(), True, 1.0])
+def test_maybe_compress_bad_types(body):
+    request_dict = _request_dict(body)
+    maybe_compress_request(
+        COMPRESSION_CONFIG_1_BYTE, request_dict, OP_WITH_COMPRESSION
+    )
+    assert request_dict['body'] == body
+
+
+@mock.patch.object(botocore.compress, 'GzipFile', StaticGzipFile)
+def test_body_streams_position_reset():
+    request_dict = _request_dict(io.BytesIO(REQUEST_BODY))
+    maybe_compress_request(
+        COMPRESSION_CONFIG_128_BYTES,
+        request_dict,
+        OP_WITH_COMPRESSION,
+    )
+    assert request_dict['body'].tell() == 0
+    assert_request_compressed(request_dict, REQUEST_BODY_COMPRESSED)
+
+
+@mock.patch.object(botocore.compress, 'gzip_compress', static_compress)
+@mock.patch.object(botocore.compress, 'COMPRESSION_MAPPING', MOCK_COMPRESSION)
+@pytest.mark.skipif(
+    sys.version_info < (3, 8), reason='requires python3.8 or higher'
+)
+def test_only_compress_once():
+    request_dict = _request_dict()
+    maybe_compress_request(
+        COMPRESSION_CONFIG_128_BYTES,
+        request_dict,
+        OP_MULTIPLE_COMPRESSIONS,
+    )
+    assert_request_compressed(request_dict, REQUEST_BODY_COMPRESSED)
diff --git a/tests/unit/test_model.py b/tests/unit/test_model.py
index 62c8f568d..303ed1a31 100644
--- a/tests/unit/test_model.py
+++ b/tests/unit/test_model.py
@@ -208,6 +208,7 @@ class TestOperationModelFromService(unittest.TestCase):
                     },
                     'errors': [{'shape': 'NoSuchResourceException'}],
                     'documentation': 'Docs for PayloadOperation',
+                    'requestcompression': {'encodings': ['gzip']},
                 },
                 'NoBodyOperation': {
                     'http': {
@@ -521,12 +522,19 @@ class TestOperationModelFromService(unittest.TestCase):
         self.assertEqual(static_ctx_param2.name, 'booleanStaticContextParam')
         self.assertEqual(static_ctx_param2.value, True)
 
-    def test_static_context_parameter_abent(self):
+    def test_static_context_parameter_absent(self):
         service_model = model.ServiceModel(self.model)
         operation = service_model.operation_model('OperationTwo')
         self.assertIsInstance(operation.static_context_parameters, list)
         self.assertEqual(len(operation.static_context_parameters), 0)
 
+    def test_request_compression(self):
+        service_model = model.ServiceModel(self.model)
+        operation = service_model.operation_model('PayloadOperation')
+        self.assertEqual(
+            operation.request_compression, {'encodings': ['gzip']}
+        )
+
 
 class TestOperationModelEventStreamTypes(unittest.TestCase):
     def setUp(self):