Intermediate changes

commit_hash:1f4ace0f7eebc924221b73c2939b9824cad6b5c5

22 files changed, 441 insertions, 41 deletions

diff --git a/contrib/python/boto3/py3/.dist-info/METADATA b/contrib/python/boto3/py3/.dist-info/METADATA
index 642dc343489..6daed32e2be 100644
--- a/contrib/python/boto3/py3/.dist-info/METADATA
+++ b/contrib/python/boto3/py3/.dist-info/METADATA
@@ -1,6 +1,6 @@
 Metadata-Version: 2.1
 Name: boto3
-Version: 1.29.6
+Version: 1.33.13
 Summary: The AWS SDK for Python
 Home-page: https://github.com/boto/boto3
 Author: Amazon Web Services
@@ -24,9 +24,9 @@ Classifier: Programming Language :: Python :: 3.12
 Requires-Python: >= 3.7
 License-File: LICENSE
 License-File: NOTICE
-Requires-Dist: botocore (<1.33.0,>=1.32.6)
+Requires-Dist: botocore (<1.34.0,>=1.33.13)
 Requires-Dist: jmespath (<2.0.0,>=0.7.1)
-Requires-Dist: s3transfer (<0.8.0,>=0.7.0)
+Requires-Dist: s3transfer (<0.9.0,>=0.8.2)
 Provides-Extra: crt
 Requires-Dist: botocore[crt] (<2.0a0,>=1.21.0) ; extra == 'crt'
 
diff --git a/contrib/python/boto3/py3/boto3/__init__.py b/contrib/python/boto3/py3/boto3/__init__.py
index 046ccce6c24..8ce408ac3d0 100644
--- a/contrib/python/boto3/py3/boto3/__init__.py
+++ b/contrib/python/boto3/py3/boto3/__init__.py
@@ -17,7 +17,7 @@ from boto3.compat import _warn_deprecated_python
 from boto3.session import Session
 
 __author__ = 'Amazon Web Services'
-__version__ = '1.29.6'
+__version__ = '1.33.13'
 
 
 # The default Boto3 session; autoloaded when needed.
diff --git a/contrib/python/boto3/py3/boto3/crt.py b/contrib/python/boto3/py3/boto3/crt.py
new file mode 100644
index 00000000000..4b8df3140e0
--- /dev/null
+++ b/contrib/python/boto3/py3/boto3/crt.py
@@ -0,0 +1,167 @@
+# 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
+#
+# https://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.
+"""
+This file contains private functionality for interacting with the AWS
+Common Runtime library (awscrt) in boto3.
+
+All code contained within this file is for internal usage within this
+project and is not intended for external consumption. All interfaces
+contained within are subject to abrupt breaking changes.
+"""
+
+import threading
+
+import botocore.exceptions
+from botocore.session import Session
+from s3transfer.crt import (
+    BotocoreCRTCredentialsWrapper,
+    BotocoreCRTRequestSerializer,
+    CRTTransferManager,
+    acquire_crt_s3_process_lock,
+    create_s3_crt_client,
+)
+
+# Singletons for CRT-backed transfers
+CRT_S3_CLIENT = None
+BOTOCORE_CRT_SERIALIZER = None
+
+CLIENT_CREATION_LOCK = threading.Lock()
+PROCESS_LOCK_NAME = 'boto3'
+
+
+def _create_crt_client(session, config, region_name, cred_provider):
+    """Create a CRT S3 Client for file transfer.
+
+    Instantiating many of these may lead to degraded performance or
+    system resource exhaustion.
+    """
+    create_crt_client_kwargs = {
+        'region': region_name,
+        'use_ssl': True,
+        'crt_credentials_provider': cred_provider,
+    }
+    return create_s3_crt_client(**create_crt_client_kwargs)
+
+
+def _create_crt_request_serializer(session, region_name):
+    return BotocoreCRTRequestSerializer(
+        session, {'region_name': region_name, 'endpoint_url': None}
+    )
+
+
+def _create_crt_s3_client(
+    session, config, region_name, credentials, lock, **kwargs
+):
+    """Create boto3 wrapper class to manage crt lock reference and S3 client."""
+    cred_wrapper = BotocoreCRTCredentialsWrapper(credentials)
+    cred_provider = cred_wrapper.to_crt_credentials_provider()
+    return CRTS3Client(
+        _create_crt_client(session, config, region_name, cred_provider),
+        lock,
+        region_name,
+        cred_wrapper,
+    )
+
+
+def _initialize_crt_transfer_primatives(client, config):
+    lock = acquire_crt_s3_process_lock(PROCESS_LOCK_NAME)
+    if lock is None:
+        # If we're unable to acquire the lock, we cannot
+        # use the CRT in this process and should default to
+        # the classic s3transfer manager.
+        return None, None
+
+    session = Session()
+    region_name = client.meta.region_name
+    credentials = client._get_credentials()
+
+    serializer = _create_crt_request_serializer(session, region_name)
+    s3_client = _create_crt_s3_client(
+        session, config, region_name, credentials, lock
+    )
+    return serializer, s3_client
+
+
+def get_crt_s3_client(client, config):
+    global CRT_S3_CLIENT
+    global BOTOCORE_CRT_SERIALIZER
+
+    with CLIENT_CREATION_LOCK:
+        if CRT_S3_CLIENT is None:
+            serializer, s3_client = _initialize_crt_transfer_primatives(
+                client, config
+            )
+            BOTOCORE_CRT_SERIALIZER = serializer
+            CRT_S3_CLIENT = s3_client
+
+    return CRT_S3_CLIENT
+
+
+class CRTS3Client:
+    """
+    This wrapper keeps track of our underlying CRT client, the lock used to
+    acquire it and the region we've used to instantiate the client.
+
+    Due to limitations in the existing CRT interfaces, we can only make calls
+    in a single region and does not support redirects. We track the region to
+    ensure we don't use the CRT client when a successful request cannot be made.
+    """
+
+    def __init__(self, crt_client, process_lock, region, cred_provider):
+        self.crt_client = crt_client
+        self.process_lock = process_lock
+        self.region = region
+        self.cred_provider = cred_provider
+
+
+def is_crt_compatible_request(client, crt_s3_client):
+    """
+    Boto3 client must use same signing region and credentials
+    as the CRT_S3_CLIENT singleton. Otherwise fallback to classic.
+    """
+    if crt_s3_client is None:
+        return False
+
+    boto3_creds = client._get_credentials()
+    if boto3_creds is None:
+        return False
+
+    is_same_identity = compare_identity(
+        boto3_creds.get_frozen_credentials(), crt_s3_client.cred_provider
+    )
+    is_same_region = client.meta.region_name == crt_s3_client.region
+    return is_same_region and is_same_identity
+
+
+def compare_identity(boto3_creds, crt_s3_creds):
+    try:
+        crt_creds = crt_s3_creds()
+    except botocore.exceptions.NoCredentialsError:
+        return False
+
+    is_matching_identity = (
+        boto3_creds.access_key == crt_creds.access_key_id
+        and boto3_creds.secret_key == crt_creds.secret_access_key
+        and boto3_creds.token == crt_creds.session_token
+    )
+    return is_matching_identity
+
+
+def create_crt_transfer_manager(client, config):
+    """Create a CRTTransferManager for optimized data transfer."""
+    crt_s3_client = get_crt_s3_client(client, config)
+    if is_crt_compatible_request(client, crt_s3_client):
+        return CRTTransferManager(
+            crt_s3_client.crt_client, BOTOCORE_CRT_SERIALIZER
+        )
+    return None
diff --git a/contrib/python/boto3/py3/boto3/docs/method.py b/contrib/python/boto3/py3/boto3/docs/method.py
index b7520088c2a..86133674737 100644
--- a/contrib/python/boto3/py3/boto3/docs/method.py
+++ b/contrib/python/boto3/py3/boto3/docs/method.py
@@ -28,7 +28,6 @@ def document_model_driven_resource_method(
     resource_action_model=None,
     include_signature=True,
 ):
-
     document_model_driven_method(
         section=section,
         method_name=method_name,
diff --git a/contrib/python/boto3/py3/boto3/dynamodb/conditions.py b/contrib/python/boto3/py3/boto3/dynamodb/conditions.py
index 442f11c4cd9..74b3e8e7835 100644
--- a/contrib/python/boto3/py3/boto3/dynamodb/conditions.py
+++ b/contrib/python/boto3/py3/boto3/dynamodb/conditions.py
@@ -23,7 +23,6 @@ ATTR_NAME_REGEX = re.compile(r'[^.\[\]]+(?![^\[]*\])')
 
 
 class ConditionBase:
-
     expression_format = ''
     expression_operator = ''
     has_grouped_values = False
diff --git a/contrib/python/boto3/py3/boto3/exceptions.py b/contrib/python/boto3/py3/boto3/exceptions.py
index 0068de9cedf..7d9ceaf18df 100644
--- a/contrib/python/boto3/py3/boto3/exceptions.py
+++ b/contrib/python/boto3/py3/boto3/exceptions.py
@@ -40,7 +40,7 @@ class UnknownAPIVersionError(
 ):
     def __init__(self, service_name, bad_api_version, available_api_versions):
         msg = (
-            f"The '{service_name}' resource does not an API version of: {bad_api_version}\n"
+            f"The '{service_name}' resource does not support an API version of: {bad_api_version}\n"
             f"Valid API versions are: {available_api_versions}"
         )
         # Not using super because we don't want the DataNotFoundError
diff --git a/contrib/python/boto3/py3/boto3/resources/factory.py b/contrib/python/boto3/py3/boto3/resources/factory.py
index 5d9531b86ea..4cdd2f01dd4 100644
--- a/contrib/python/boto3/py3/boto3/resources/factory.py
+++ b/contrib/python/boto3/py3/boto3/resources/factory.py
@@ -370,6 +370,7 @@ class ResourceFactory:
         Creates a new property on the resource to lazy-load its value
         via the resource's ``load`` method (if it exists).
         """
+
         # The property loader will check to see if this resource has already
         # been loaded and return the cached value if possible. If not, then
         # it first checks to see if it CAN be loaded (raise if not), then
diff --git a/contrib/python/boto3/py3/boto3/s3/constants.py b/contrib/python/boto3/py3/boto3/s3/constants.py
new file mode 100644
index 00000000000..c7f691fc218
--- /dev/null
+++ b/contrib/python/boto3/py3/boto3/s3/constants.py
@@ -0,0 +1,17 @@
+# 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
+#
+# https://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.
+
+
+# TransferConfig preferred_transfer_client settings
+CLASSIC_TRANSFER_CLIENT = "classic"
+AUTO_RESOLVE_TRANSFER_CLIENT = "auto"
diff --git a/contrib/python/boto3/py3/boto3/s3/inject.py b/contrib/python/boto3/py3/boto3/s3/inject.py
index c62dc3ce22b..440be5a8be6 100644
--- a/contrib/python/boto3/py3/boto3/s3/inject.py
+++ b/contrib/python/boto3/py3/boto3/s3/inject.py
@@ -10,6 +10,8 @@
 # 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 copy as python_copy
+
 from botocore.exceptions import ClientError
 
 from boto3 import utils
@@ -432,7 +434,11 @@ def copy(
     if config is None:
         config = TransferConfig()
 
-    with create_transfer_manager(self, config) as manager:
+    # copy is not supported in the CRT
+    new_config = python_copy.copy(config)
+    new_config.preferred_transfer_client = "classic"
+
+    with create_transfer_manager(self, new_config) as manager:
         future = manager.copy(
             copy_source=CopySource,
             bucket=Bucket,
diff --git a/contrib/python/boto3/py3/boto3/s3/transfer.py b/contrib/python/boto3/py3/boto3/s3/transfer.py
index cce5155d430..1c8efd4016c 100644
--- a/contrib/python/boto3/py3/boto3/s3/transfer.py
+++ b/contrib/python/boto3/py3/boto3/s3/transfer.py
@@ -122,8 +122,11 @@ transfer.  For example:
 
 
 """
-from os import PathLike, fspath
+import logging
+import threading
+from os import PathLike, fspath, getpid
 
+from botocore.compat import HAS_CRT
 from botocore.exceptions import ClientError
 from s3transfer.exceptions import (
     RetriesExceededError as S3TransferRetriesExceededError,
@@ -134,11 +137,19 @@ from s3transfer.manager import TransferManager
 from s3transfer.subscribers import BaseSubscriber
 from s3transfer.utils import OSUtils
 
+import boto3.s3.constants as constants
 from boto3.exceptions import RetriesExceededError, S3UploadFailedError
 
+if HAS_CRT:
+    import awscrt.s3
+
+    from boto3.crt import create_crt_transfer_manager
+
 KB = 1024
 MB = KB * KB
 
+logger = logging.getLogger(__name__)
+
 
 def create_transfer_manager(client, config, osutil=None):
     """Creates a transfer manager based on configuration
@@ -155,6 +166,63 @@ def create_transfer_manager(client, config, osutil=None):
     :rtype: s3transfer.manager.TransferManager
     :returns: A transfer manager based on parameters provided
     """
+    if _should_use_crt(config):
+        crt_transfer_manager = create_crt_transfer_manager(client, config)
+        if crt_transfer_manager is not None:
+            logger.debug(
+                f"Using CRT client. pid: {getpid()}, thread: {threading.get_ident()}"
+            )
+            return crt_transfer_manager
+
+    # If we don't resolve something above, fallback to the default.
+    logger.debug(
+        f"Using default client. pid: {getpid()}, thread: {threading.get_ident()}"
+    )
+    return _create_default_transfer_manager(client, config, osutil)
+
+
+def _should_use_crt(config):
+    # This feature requires awscrt>=0.19.17
+    if HAS_CRT and has_minimum_crt_version((0, 19, 17)):
+        is_optimized_instance = awscrt.s3.is_optimized_for_system()
+    else:
+        is_optimized_instance = False
+    pref_transfer_client = config.preferred_transfer_client.lower()
+
+    if (
+        is_optimized_instance
+        and pref_transfer_client == constants.AUTO_RESOLVE_TRANSFER_CLIENT
+    ):
+        logger.debug(
+            "Attempting to use CRTTransferManager. Config settings may be ignored."
+        )
+        return True
+
+    logger.debug(
+        "Opting out of CRT Transfer Manager. Preferred client: "
+        f"{pref_transfer_client}, CRT available: {HAS_CRT}, "
+        f"Instance Optimized: {is_optimized_instance}."
+    )
+    return False
+
+
+def has_minimum_crt_version(minimum_version):
+    """Not intended for use outside boto3."""
+    if not HAS_CRT:
+        return False
+
+    crt_version_str = awscrt.__version__
+    try:
+        crt_version_ints = map(int, crt_version_str.split("."))
+        crt_version_tuple = tuple(crt_version_ints)
+    except (TypeError, ValueError):
+        return False
+
+    return crt_version_tuple >= minimum_version
+
+
+def _create_default_transfer_manager(client, config, osutil):
+    """Create the default TransferManager implementation for s3transfer."""
     executor_cls = None
     if not config.use_threads:
         executor_cls = NonThreadedExecutor
@@ -177,6 +245,7 @@ class TransferConfig(S3TransferConfig):
         io_chunksize=256 * KB,
         use_threads=True,
         max_bandwidth=None,
+        preferred_transfer_client=constants.AUTO_RESOLVE_TRANSFER_CLIENT,
     ):
         """Configuration object for managed S3 transfers
 
@@ -217,6 +286,15 @@ class TransferConfig(S3TransferConfig):
         :param max_bandwidth: The maximum bandwidth that will be consumed
             in uploading and downloading file content. The value is an integer
             in terms of bytes per second.
+
+        :param preferred_transfer_client: String specifying preferred transfer
+            client for transfer operations.
+
+            Current supported settings are:
+              * auto (default) - Use the CRTTransferManager when calls
+                  are made with supported environment and settings.
+              * classic - Only use the origin S3TransferManager with
+                  requests. Disables possible CRT upgrade on requests.
         """
         super().__init__(
             multipart_threshold=multipart_threshold,
@@ -233,6 +311,7 @@ class TransferConfig(S3TransferConfig):
         for alias in self.ALIAS:
             setattr(self, alias, getattr(self, self.ALIAS[alias]))
         self.use_threads = use_threads
+        self.preferred_transfer_client = preferred_transfer_client
 
     def __setattr__(self, name, value):
         # If the alias name is used, make sure we set the name that it points
diff --git a/contrib/python/boto3/py3/boto3/session.py b/contrib/python/boto3/py3/boto3/session.py
index bdda65ad416..37890ada85e 100644
--- a/contrib/python/boto3/py3/boto3/session.py
+++ b/contrib/python/boto3/py3/boto3/session.py
@@ -478,7 +478,6 @@ class Session:
         return cls(client=client)
 
     def _register_default_handlers(self):
-
         # S3 customizations
         self._session.register(
             'creating-client-class.s3',
diff --git a/contrib/python/boto3/py3/ya.make b/contrib/python/boto3/py3/ya.make
index 673125e7b2a..62c9f40eed5 100644
--- a/contrib/python/boto3/py3/ya.make
+++ b/contrib/python/boto3/py3/ya.make
@@ -2,7 +2,7 @@
 
 PY3_LIBRARY()
 
-VERSION(1.29.6)
+VERSION(1.33.13)
 
 LICENSE(Apache-2.0)
 
@@ -14,10 +14,15 @@ PEERDIR(
 
 NO_LINT()
 
+NO_CHECK_IMPORTS(
+    boto3.crt
+)
+
 PY_SRCS(
     TOP_LEVEL
     boto3/__init__.py
     boto3/compat.py
+    boto3/crt.py
     boto3/docs/__init__.py
     boto3/docs/action.py
     boto3/docs/attr.py
@@ -49,6 +54,7 @@ PY_SRCS(
     boto3/resources/params.py
     boto3/resources/response.py
     boto3/s3/__init__.py
+    boto3/s3/constants.py
     boto3/s3/inject.py
     boto3/s3/transfer.py
     boto3/session.py
diff --git a/contrib/python/s3transfer/py3/.dist-info/METADATA b/contrib/python/s3transfer/py3/.dist-info/METADATA
index 46933ce84f8..8fd5aeaa49f 100644
--- a/contrib/python/s3transfer/py3/.dist-info/METADATA
+++ b/contrib/python/s3transfer/py3/.dist-info/METADATA
@@ -1,6 +1,6 @@
 Metadata-Version: 2.1
 Name: s3transfer
-Version: 0.8.0
+Version: 0.8.2
 Summary: An Amazon S3 Transfer Manager
 Home-page: https://github.com/boto/s3transfer
 Author: Amazon Web Services
@@ -23,9 +23,9 @@ Classifier: Programming Language :: Python :: 3.12
 Requires-Python: >= 3.7
 License-File: LICENSE.txt
 License-File: NOTICE.txt
-Requires-Dist: botocore (<2.0a.0,>=1.32.7)
+Requires-Dist: botocore (<2.0a.0,>=1.33.2)
 Provides-Extra: crt
-Requires-Dist: botocore[crt] (<2.0a.0,>=1.32.7) ; extra == 'crt'
+Requires-Dist: botocore[crt] (<2.0a.0,>=1.33.2) ; extra == 'crt'
 
 =====================================================
 s3transfer - An Amazon S3 Transfer Manager for Python
diff --git a/contrib/python/s3transfer/py3/s3transfer/__init__.py b/contrib/python/s3transfer/py3/s3transfer/__init__.py
index c6014f052f7..fa251490743 100644
--- a/contrib/python/s3transfer/py3/s3transfer/__init__.py
+++ b/contrib/python/s3transfer/py3/s3transfer/__init__.py
@@ -144,7 +144,7 @@ import s3transfer.compat
 from s3transfer.exceptions import RetriesExceededError, S3UploadFailedError
 
 __author__ = 'Amazon Web Services'
-__version__ = '0.8.0'
+__version__ = '0.8.2'
 
 
 class NullHandler(logging.Handler):
diff --git a/contrib/python/s3transfer/py3/s3transfer/crt.py b/contrib/python/s3transfer/py3/s3transfer/crt.py
index 24fa7976569..367f1f6f203 100644
--- a/contrib/python/s3transfer/py3/s3transfer/crt.py
+++ b/contrib/python/s3transfer/py3/s3transfer/crt.py
@@ -15,6 +15,7 @@ import threading
 from io import BytesIO
 
 import awscrt.http
+import awscrt.s3
 import botocore.awsrequest
 import botocore.session
 from awscrt.auth import AwsCredentials, AwsCredentialsProvider
@@ -25,13 +26,7 @@ from awscrt.io import (
     EventLoopGroup,
     TlsContextOptions,
 )
-from awscrt.s3 import (
-    S3Client,
-    S3RequestTlsMode,
-    S3RequestType,
-    S3ResponseError,
-    get_recommended_throughput_target_gbps,
-)
+from awscrt.s3 import S3Client, S3RequestTlsMode, S3RequestType
 from botocore import UNSIGNED
 from botocore.compat import urlsplit
 from botocore.config import Config
@@ -124,7 +119,6 @@ def create_s3_crt_client(
             use. Specify this argument if you want to use a custom CA cert
             bundle instead of the default one on your system.
     """
-
     event_loop_group = EventLoopGroup(num_threads)
     host_resolver = DefaultHostResolver(event_loop_group)
     bootstrap = ClientBootstrap(event_loop_group, host_resolver)
@@ -159,7 +153,7 @@ def create_s3_crt_client(
 
 def _get_crt_throughput_target_gbps(provided_throughput_target_bytes=None):
     if provided_throughput_target_bytes is None:
-        target_gbps = get_recommended_throughput_target_gbps()
+        target_gbps = awscrt.s3.get_recommended_throughput_target_gbps()
         logger.debug(
             'Recommended CRT throughput target in gbps: %s', target_gbps
         )
@@ -544,7 +538,7 @@ class BotocoreCRTRequestSerializer(BaseCRTRequestSerializer):
         return crt_request
 
     def translate_crt_exception(self, exception):
-        if isinstance(exception, S3ResponseError):
+        if isinstance(exception, awscrt.s3.S3ResponseError):
             return self._translate_crt_s3_response_error(exception)
         else:
             return None
diff --git a/contrib/python/s3transfer/py3/s3transfer/manager.py b/contrib/python/s3transfer/py3/s3transfer/manager.py
index b11daeba958..ab9a210f82f 100644
--- a/contrib/python/s3transfer/py3/s3transfer/manager.py
+++ b/contrib/python/s3transfer/py3/s3transfer/manager.py
@@ -35,6 +35,7 @@ from s3transfer.utils import (
     OSUtils,
     SlidingWindowSemaphore,
     TaskSemaphore,
+    add_s3express_defaults,
     get_callbacks,
     signal_not_transferring,
     signal_transferring,
@@ -320,6 +321,7 @@ class TransferManager:
             subscribers = []
         self._validate_all_known_args(extra_args, self.ALLOWED_UPLOAD_ARGS)
         self._validate_if_bucket_supported(bucket)
+        self._add_operation_defaults(bucket, extra_args)
         call_args = CallArgs(
             fileobj=fileobj,
             bucket=bucket,
@@ -502,6 +504,9 @@ class TransferManager:
                     "must be one of: %s" % (kwarg, ', '.join(allowed))
                 )
 
+    def _add_operation_defaults(self, bucket, extra_args):
+        add_s3express_defaults(bucket, extra_args)
+
     def _submit_transfer(
         self, call_args, submission_task_cls, extra_main_kwargs=None
     ):
diff --git a/contrib/python/s3transfer/py3/s3transfer/subscribers.py b/contrib/python/s3transfer/py3/s3transfer/subscribers.py
index cf0dbaa0d79..473d5d94201 100644
--- a/contrib/python/s3transfer/py3/s3transfer/subscribers.py
+++ b/contrib/python/s3transfer/py3/s3transfer/subscribers.py
@@ -10,6 +10,8 @@
 # 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.
+from functools import lru_cache
+
 from s3transfer.compat import accepts_kwargs
 from s3transfer.exceptions import InvalidSubscriberMethodError
 
@@ -28,6 +30,7 @@ class BaseSubscriber:
         return super().__new__(cls)
 
     @classmethod
+    @lru_cache()
     def _validate_subscriber_methods(cls):
         for subscriber_type in cls.VALID_SUBSCRIBER_TYPES:
             subscriber_method = getattr(cls, 'on_' + subscriber_type)
diff --git a/contrib/python/s3transfer/py3/s3transfer/utils.py b/contrib/python/s3transfer/py3/s3transfer/utils.py
index 61407eba5c5..9954dc0a9d2 100644
--- a/contrib/python/s3transfer/py3/s3transfer/utils.py
+++ b/contrib/python/s3transfer/py3/s3transfer/utils.py
@@ -22,6 +22,8 @@ import threading
 from collections import defaultdict
 
 from botocore.exceptions import IncompleteReadError, ReadTimeoutError
+from botocore.httpchecksum import AwsChunkedWrapper
+from botocore.utils import is_s3express_bucket
 
 from s3transfer.compat import SOCKET_ERROR, fallocate, rename_file
 
@@ -54,10 +56,12 @@ def signal_not_transferring(request, operation_name, **kwargs):
 
 
 def signal_transferring(request, operation_name, **kwargs):
-    if operation_name in ['PutObject', 'UploadPart'] and hasattr(
-        request.body, 'signal_transferring'
-    ):
-        request.body.signal_transferring()
+    if operation_name in ['PutObject', 'UploadPart']:
+        body = request.body
+        if isinstance(body, AwsChunkedWrapper):
+            body = getattr(body, '_raw', None)
+        if hasattr(body, 'signal_transferring'):
+            body.signal_transferring()
 
 
 def calculate_num_parts(size, part_size):
@@ -800,3 +804,9 @@ class ChunksizeAdjuster:
             )
 
         return chunksize
+
+
+def add_s3express_defaults(bucket, extra_args):
+    if is_s3express_bucket(bucket) and "ChecksumAlgorithm" not in extra_args:
+        # Default Transfer Operations to S3Express to use CRC32
+        extra_args["ChecksumAlgorithm"] = "crc32"
diff --git a/contrib/python/s3transfer/py3/tests/functional/test_upload.py b/contrib/python/s3transfer/py3/tests/functional/test_upload.py
index 92fe8d803d7..8be9f1eca9b 100644
--- a/contrib/python/s3transfer/py3/tests/functional/test_upload.py
+++ b/contrib/python/s3transfer/py3/tests/functional/test_upload.py
@@ -142,9 +142,12 @@ class TestNonMultipartUpload(BaseUploadTest):
     __test__ = True
 
     def add_put_object_response_with_default_expected_params(
-        self, extra_expected_params=None
+        self, extra_expected_params=None, bucket=None
     ):
-        expected_params = {'Body': ANY, 'Bucket': self.bucket, 'Key': self.key}
+        if bucket is None:
+            bucket = self.bucket
+
+        expected_params = {'Body': ANY, 'Bucket': bucket, 'Key': self.key}
         if extra_expected_params:
             expected_params.update(extra_expected_params)
         upload_response = self.create_stubbed_responses()[0]
@@ -167,9 +170,9 @@ class TestNonMultipartUpload(BaseUploadTest):
         self.assert_put_object_body_was_correct()
 
     def test_upload_with_checksum(self):
-        self.extra_args['ChecksumAlgorithm'] = 'crc32'
+        self.extra_args['ChecksumAlgorithm'] = 'sha256'
         self.add_put_object_response_with_default_expected_params(
-            extra_expected_params={'ChecksumAlgorithm': 'crc32'}
+            extra_expected_params={'ChecksumAlgorithm': 'sha256'}
         )
         future = self.manager.upload(
             self.filename, self.bucket, self.key, self.extra_args
@@ -178,6 +181,21 @@ class TestNonMultipartUpload(BaseUploadTest):
         self.assert_expected_client_calls_were_correct()
         self.assert_put_object_body_was_correct()
 
+    def test_upload_with_s3express_default_checksum(self):
+        s3express_bucket = "mytestbucket--usw2-az6--x-s3"
+        self.assertFalse("ChecksumAlgorithm" in self.extra_args)
+
+        self.add_put_object_response_with_default_expected_params(
+            extra_expected_params={'ChecksumAlgorithm': 'crc32'},
+            bucket=s3express_bucket,
+        )
+        future = self.manager.upload(
+            self.filename, s3express_bucket, self.key, self.extra_args
+        )
+        future.result()
+        self.assert_expected_client_calls_were_correct()
+        self.assert_put_object_body_was_correct()
+
     def test_upload_for_fileobj(self):
         self.add_put_object_response_with_default_expected_params()
         with open(self.filename, 'rb') as f:
@@ -342,9 +360,14 @@ class TestMultipartUpload(BaseUploadTest):
         self.assertEqual(self.sent_bodies, expected_contents)
 
     def add_create_multipart_response_with_default_expected_params(
-        self, extra_expected_params=None
+        self,
+        extra_expected_params=None,
+        bucket=None,
     ):
-        expected_params = {'Bucket': self.bucket, 'Key': self.key}
+        if bucket is None:
+            bucket = self.bucket
+
+        expected_params = {'Bucket': bucket, 'Key': self.key}
         if extra_expected_params:
             expected_params.update(extra_expected_params)
         response = self.create_stubbed_responses()[0]
@@ -352,14 +375,19 @@ class TestMultipartUpload(BaseUploadTest):
         self.stubber.add_response(**response)
 
     def add_upload_part_responses_with_default_expected_params(
-        self, extra_expected_params=None
+        self,
+        extra_expected_params=None,
+        bucket=None,
     ):
+        if bucket is None:
+            bucket = self.bucket
+
         num_parts = 3
         upload_part_responses = self.create_stubbed_responses()[1:-1]
         for i in range(num_parts):
             upload_part_response = upload_part_responses[i]
             expected_params = {
-                'Bucket': self.bucket,
+                'Bucket': bucket,
                 'Key': self.key,
                 'UploadId': self.multipart_id,
                 'Body': ANY,
@@ -378,10 +406,15 @@ class TestMultipartUpload(BaseUploadTest):
             self.stubber.add_response(**upload_part_response)
 
     def add_complete_multipart_response_with_default_expected_params(
-        self, extra_expected_params=None
+        self,
+        extra_expected_params=None,
+        bucket=None,
     ):
+        if bucket is None:
+            bucket = self.bucket
+
         expected_params = {
-            'Bucket': self.bucket,
+            'Bucket': bucket,
             'Key': self.key,
             'UploadId': self.multipart_id,
             'MultipartUpload': {
@@ -600,6 +633,54 @@ class TestMultipartUpload(BaseUploadTest):
         future.result()
         self.assert_expected_client_calls_were_correct()
 
+    def test_multipart_upload_sets_s3express_default_checksum(self):
+        s3express_bucket = "mytestbucket--usw2-az6--x-s3"
+        self.assertFalse('ChecksumAlgorithm' in self.extra_args)
+
+        # ChecksumAlgorithm should be passed on the create_multipart call
+        self.add_create_multipart_response_with_default_expected_params(
+            extra_expected_params={'ChecksumAlgorithm': 'crc32'},
+            bucket=s3express_bucket,
+        )
+
+        # ChecksumAlgorithm should be forwarded and a SHA1 will come back
+        self.add_upload_part_responses_with_default_expected_params(
+            extra_expected_params={'ChecksumAlgorithm': 'crc32'},
+            bucket=s3express_bucket,
+        )
+
+        # The checksums should be used in the complete call like etags
+        self.add_complete_multipart_response_with_default_expected_params(
+            extra_expected_params={
+                'MultipartUpload': {
+                    'Parts': [
+                        {
+                            'ETag': 'etag-1',
+                            'PartNumber': 1,
+                            'ChecksumCRC32': 'sum1==',
+                        },
+                        {
+                            'ETag': 'etag-2',
+                            'PartNumber': 2,
+                            'ChecksumCRC32': 'sum2==',
+                        },
+                        {
+                            'ETag': 'etag-3',
+                            'PartNumber': 3,
+                            'ChecksumCRC32': 'sum3==',
+                        },
+                    ]
+                }
+            },
+            bucket=s3express_bucket,
+        )
+
+        future = self.manager.upload(
+            self.filename, s3express_bucket, self.key, self.extra_args
+        )
+        future.result()
+        self.assert_expected_client_calls_were_correct()
+
     def test_multipart_upload_with_ssec_args(self):
         params = {
             'RequestPayer': 'requester',
diff --git a/contrib/python/s3transfer/py3/tests/unit/test_crt.py b/contrib/python/s3transfer/py3/tests/unit/test_crt.py
index ac1a45ebb7c..47a9bae4a0e 100644
--- a/contrib/python/s3transfer/py3/tests/unit/test_crt.py
+++ b/contrib/python/s3transfer/py3/tests/unit/test_crt.py
@@ -54,7 +54,7 @@ def mock_s3_crt_client():
 @pytest.fixture
 def mock_get_recommended_throughput_target_gbps():
     with mock.patch(
-        's3transfer.crt.get_recommended_throughput_target_gbps'
+        'awscrt.s3.get_recommended_throughput_target_gbps'
     ) as mock_get_target_gbps:
         yield mock_get_target_gbps
 
diff --git a/contrib/python/s3transfer/py3/tests/unit/test_utils.py b/contrib/python/s3transfer/py3/tests/unit/test_utils.py
index 217779943b0..5cfc639ff55 100644
--- a/contrib/python/s3transfer/py3/tests/unit/test_utils.py
+++ b/contrib/python/s3transfer/py3/tests/unit/test_utils.py
@@ -20,6 +20,8 @@ import threading
 import time
 from io import BytesIO, StringIO
 
+import pytest
+
 from s3transfer.futures import TransferFuture, TransferMeta
 from s3transfer.utils import (
     MAX_PARTS,
@@ -36,6 +38,7 @@ from s3transfer.utils import (
     SlidingWindowSemaphore,
     StreamReaderProgress,
     TaskSemaphore,
+    add_s3express_defaults,
     calculate_num_parts,
     calculate_range_parameter,
     get_callbacks,
@@ -1187,3 +1190,34 @@ class TestAdjustChunksize(unittest.TestCase):
         chunksize = MAX_SINGLE_UPLOAD_SIZE + 1
         new_size = self.adjuster.adjust_chunksize(chunksize)
         self.assertEqual(new_size, MAX_SINGLE_UPLOAD_SIZE)
+
+
+class TestS3ExpressDefaults:
+    @pytest.mark.parametrize(
+        "bucket,extra_args,expected",
+        (
+            (
+                "mytestbucket--usw2-az2--x-s3",
+                {},
+                {"ChecksumAlgorithm": "crc32"},
+            ),
+            (
+                "mytestbucket--usw2-az2--x-s3",
+                {"Some": "Setting"},
+                {"ChecksumAlgorithm": "crc32", "Some": "Setting"},
+            ),
+            (
+                "mytestbucket",
+                {},
+                {},
+            ),
+            (
+                "mytestbucket--usw2-az2--x-s3",
+                {"ChecksumAlgorithm": "sha256"},
+                {"ChecksumAlgorithm": "sha256"},
+            ),
+        ),
+    )
+    def test_add_s3express_defaults(self, bucket, extra_args, expected):
+        add_s3express_defaults(bucket, extra_args)
+        assert extra_args == expected
diff --git a/contrib/python/s3transfer/py3/ya.make b/contrib/python/s3transfer/py3/ya.make
index 78692b21c1b..dfb007f8509 100644
--- a/contrib/python/s3transfer/py3/ya.make
+++ b/contrib/python/s3transfer/py3/ya.make
@@ -2,7 +2,7 @@
 
 PY3_LIBRARY()
 
-VERSION(0.8.0)
+VERSION(0.8.2)
 
 LICENSE(Apache-2.0)