From 923d380e3b38a52e8878b165cd90a48715b90134 Mon Sep 17 00:00:00 2001 From: Steve Yoo Date: Fri, 11 Sep 2026 15:39:47 -0400 Subject: [PATCH 1/2] Support multipart_threshold for downloads using CRT --- .../next-release/enhancement-crt-69912.json | 5 + awscli/customizations/s3/factory.py | 19 +++- awscli/s3transfer/crt.py | 56 ++++++++++- tests/functional/s3transfer/test_crt.py | 99 +++++++++++++++++++ 4 files changed, 171 insertions(+), 8 deletions(-) create mode 100644 .changes/next-release/enhancement-crt-69912.json diff --git a/.changes/next-release/enhancement-crt-69912.json b/.changes/next-release/enhancement-crt-69912.json new file mode 100644 index 000000000000..0725537f4642 --- /dev/null +++ b/.changes/next-release/enhancement-crt-69912.json @@ -0,0 +1,5 @@ +{ + "type": "enhancement", + "category": "crt", + "description": "Configure CRT client to download with single GET when object size is below ``multipart_threshold``" +} diff --git a/awscli/customizations/s3/factory.py b/awscli/customizations/s3/factory.py index 2dcbf3368cb2..18b4cfb90fc2 100644 --- a/awscli/customizations/s3/factory.py +++ b/awscli/customizations/s3/factory.py @@ -20,6 +20,7 @@ from s3transfer.crt import ( BotocoreCRTCredentialsWrapper, BotocoreCRTRequestSerializer, + CRTTransferConfig, CRTTransferManager, acquire_crt_s3_process_lock, create_s3_crt_client, @@ -241,12 +242,22 @@ def _acquire_crt_s3_process_lock(self): def _create_crt_transfer_manager(self, params, runtime_config): self._acquire_crt_s3_process_lock() + config_kwargs = self._resolve_crt_client_config_kwargs(runtime_config) return CRTTransferManager( - self._create_crt_client(params, runtime_config), + self._create_crt_client(params, runtime_config, config_kwargs), self._create_crt_request_serializer(params), + transfer_config=self._create_crt_transfer_config(config_kwargs), ) - def _create_crt_client(self, params, runtime_config): + def _create_crt_transfer_config(self, config_kwargs): + # The crt client only applies its multipart threshold to uploads, so + # downloads rely on the transfer config to match it. Leaving the + # threshold unset keeps the client's own download behavior. + return CRTTransferConfig( + multipart_threshold=config_kwargs.get('multipart_upload_threshold') + ) + + def _create_crt_client(self, params, runtime_config, config_kwargs): create_crt_client_kwargs = { 'region': self._resolve_region(params), 'verify': self._resolve_verify(params), @@ -257,9 +268,7 @@ def _create_crt_client(self, params, runtime_config): target_throughput = runtime_config.get('target_bandwidth', None) if target_throughput: create_crt_client_kwargs['target_throughput'] = target_throughput - create_crt_client_kwargs.update( - self._resolve_crt_client_config_kwargs(runtime_config) - ) + create_crt_client_kwargs.update(config_kwargs) if params.get('sign_request', True): crt_credentials_provider = self._get_crt_credentials_provider() create_crt_client_kwargs['crt_credentials_provider'] = ( diff --git a/awscli/s3transfer/crt.py b/awscli/s3transfer/crt.py index a69478a80ecc..32b470f68d14 100644 --- a/awscli/s3transfer/crt.py +++ b/awscli/s3transfer/crt.py @@ -222,8 +222,31 @@ def _get_crt_throughput_target_gbps(provided_throughput_target_bytes=None): return target_gbps +class CRTTransferConfig: + def __init__(self, multipart_threshold=None): + """Configuration the CRT transfer manager applies itself + + This only covers configuration that the CRT s3 client cannot apply + on its own. Everything else is configured on the client, so it + deliberately has no equivalent here. + + :type multipart_threshold: Optional[int] + :param multipart_threshold: The size, in bytes, that a download must + exceed to be split into ranged requests. The CRT s3 client only + applies its own threshold to uploads. If not set, the client + decides how to split every download. + """ + self.multipart_threshold = multipart_threshold + + class CRTTransferManager: - def __init__(self, crt_s3_client, crt_request_serializer, osutil=None): + def __init__( + self, + crt_s3_client, + crt_request_serializer, + osutil=None, + transfer_config=None, + ): """A transfer manager interface for Amazon S3 on CRT s3 client. :type crt_s3_client: awscrt.s3.S3Client @@ -237,12 +260,19 @@ def __init__(self, crt_s3_client, crt_request_serializer, osutil=None): :type osutil: s3transfer.utils.OSUtils :param osutil: OSUtils object to use for os-related behavior when using with transfer manager. + + :type transfer_config: s3transfer.crt.CRTTransferConfig + :param transfer_config: The transfer configuration to apply. If not + provided, the CRT s3 client's own configuration applies to every + transfer. """ if osutil is None: self._osutil = OSUtils() self._crt_s3_client = crt_s3_client self._s3_args_creator = S3ClientArgsCreator( - crt_request_serializer, self._osutil + crt_request_serializer, + self._osutil, + transfer_config=transfer_config, ) self._crt_exception_translator = ( crt_request_serializer.translate_crt_exception @@ -398,6 +428,7 @@ def __init__(self, transfer_id=None, call_args=None): self._transfer_id = transfer_id self._call_args = call_args self._user_context = {} + self._size = None @property def call_args(self): @@ -411,6 +442,13 @@ def transfer_id(self): def user_context(self): return self._user_context + @property + def size(self): + return self._size + + def provide_transfer_size(self, size): + self._size = size + class CRTTransferFuture(BaseTransferFuture): def __init__(self, meta=None, coordinator=None): @@ -760,9 +798,10 @@ def set_s3_request(self, s3_request): class S3ClientArgsCreator: - def __init__(self, crt_request_serializer, os_utils): + def __init__(self, crt_request_serializer, os_utils, transfer_config=None): self._request_serializer = crt_request_serializer self._os_utils = os_utils + self._transfer_config = transfer_config self._client_config = crt_request_serializer.client_config self._service_model = crt_request_serializer.service_model @@ -892,8 +931,19 @@ def _get_make_request_args_get_object( make_request_args['recv_filepath'] = recv_filepath make_request_args['on_body'] = on_body make_request_args['checksum_config'] = checksum_config + if self._should_download_in_single_request(future.meta.size): + make_request_args['type'] = S3RequestType.DEFAULT + make_request_args['operation_name'] = _get_operation_name( + request_type + ) return make_request_args + def _should_download_in_single_request(self, size): + if self._transfer_config is None or size is None: + return False + threshold = self._transfer_config.multipart_threshold + return threshold is not None and size <= threshold + def _should_calculate_upload_checksum(self, request_type, extra_args): if any( checksum_arg in extra_args diff --git a/tests/functional/s3transfer/test_crt.py b/tests/functional/s3transfer/test_crt.py index 790d80c46452..185be2e41b25 100644 --- a/tests/functional/s3transfer/test_crt.py +++ b/tests/functional/s3transfer/test_crt.py @@ -16,6 +16,7 @@ import time from concurrent.futures import Future +import pytest from botocore.session import Session from s3transfer.subscribers import BaseSubscriber @@ -831,3 +832,101 @@ def test_crt_s3_client_error_handling(self): ) with self.assertRaises(awscrt.exceptions.AwsCrtError): future.result() + + +class ProvideSizeSubscriber(BaseSubscriber): + def __init__(self, size): + self._size = size + + def on_queued(self, future, **kwargs): + future.meta.provide_transfer_size(self._size) + + +MULTIPART_THRESHOLD = 8 * 1024 * 1024 +DOWNLOADED_CONTENT = 'content' + + +@pytest.fixture +def files(): + file_creator = FileCreator() + yield file_creator + file_creator.remove_all() + + +@pytest.fixture +def filename(files): + return files.full_path('myfile') + + +@pytest.fixture +def crt_client(files): + client = mock.Mock(awscrt.s3.S3Client) + + def simulate_make_request(**kwargs): + files.create_file( + kwargs['recv_filepath'], DOWNLOADED_CONTENT, mode='w' + ) + kwargs['on_done'](error=None) + return mock.Mock(awscrt.s3.S3Request) + + client.make_request.side_effect = simulate_make_request + return client + + +@pytest.fixture +def request_serializer(): + session = Session() + session.set_config_variable('region', 'us-west-2') + return s3transfer.crt.BotocoreCRTRequestSerializer(session) + + +@pytest.fixture +def download(crt_client, request_serializer, filename): + """Downloads an object of a given size and returns the crt request args""" + + def _download(size, multipart_threshold): + transfer_manager = s3transfer.crt.CRTTransferManager( + crt_s3_client=crt_client, + crt_request_serializer=request_serializer, + transfer_config=s3transfer.crt.CRTTransferConfig( + multipart_threshold=multipart_threshold + ), + ) + subscribers = [] if size is None else [ProvideSizeSubscriber(size)] + transfer_manager.download( + 'test_bucket', 'test_key', filename, {}, subscribers + ).result() + return crt_client.make_request.call_args[1] + + return _download + + +@requires_crt() +class TestDownloadMultipartThreshold: + @pytest.mark.parametrize( + 'size', [MULTIPART_THRESHOLD - 1, MULTIPART_THRESHOLD] + ) + def test_within_threshold_downloads_in_single_request( + self, download, size + ): + request_args = download(size, MULTIPART_THRESHOLD) + assert request_args['type'] == awscrt.s3.S3RequestType.DEFAULT + assert request_args['operation_name'] == 'GetObject' + + def test_above_threshold_splits_download(self, download): + request_args = download(MULTIPART_THRESHOLD + 1, MULTIPART_THRESHOLD) + assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT + assert 'operation_name' not in request_args + + def test_unset_threshold_splits_download(self, download): + request_args = download(1, None) + assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT + + def test_unknown_size_splits_download(self, download): + request_args = download(None, MULTIPART_THRESHOLD) + assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT + + def test_single_request_download_writes_file(self, download, filename): + download(MULTIPART_THRESHOLD - 1, MULTIPART_THRESHOLD) + with open(filename) as f: + assert f.read() == DOWNLOADED_CONTENT From fc7a4e089584a0822c8e36366de875c459fc592e Mon Sep 17 00:00:00 2001 From: Steve Yoo Date: Fri, 11 Sep 2026 16:52:53 -0400 Subject: [PATCH 2/2] clean tests --- tests/functional/s3transfer/test_crt.py | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/tests/functional/s3transfer/test_crt.py b/tests/functional/s3transfer/test_crt.py index 185be2e41b25..8e36e7b542d9 100644 --- a/tests/functional/s3transfer/test_crt.py +++ b/tests/functional/s3transfer/test_crt.py @@ -23,6 +23,7 @@ from tests import ( HAS_CRT, FileCreator, + FileSizeProvider, NonSeekableReader, NonSeekableWriter, mock, @@ -834,14 +835,6 @@ def test_crt_s3_client_error_handling(self): future.result() -class ProvideSizeSubscriber(BaseSubscriber): - def __init__(self, size): - self._size = size - - def on_queued(self, future, **kwargs): - future.meta.provide_transfer_size(self._size) - - MULTIPART_THRESHOLD = 8 * 1024 * 1024 DOWNLOADED_CONTENT = 'content' @@ -892,7 +885,7 @@ def _download(size, multipart_threshold): multipart_threshold=multipart_threshold ), ) - subscribers = [] if size is None else [ProvideSizeSubscriber(size)] + subscribers = [FileSizeProvider(size)] transfer_manager.download( 'test_bucket', 'test_key', filename, {}, subscribers ).result() @@ -901,7 +894,6 @@ def _download(size, multipart_threshold): return _download -@requires_crt() class TestDownloadMultipartThreshold: @pytest.mark.parametrize( 'size', [MULTIPART_THRESHOLD - 1, MULTIPART_THRESHOLD] @@ -922,10 +914,6 @@ def test_unset_threshold_splits_download(self, download): request_args = download(1, None) assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT - def test_unknown_size_splits_download(self, download): - request_args = download(None, MULTIPART_THRESHOLD) - assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT - def test_single_request_download_writes_file(self, download, filename): download(MULTIPART_THRESHOLD - 1, MULTIPART_THRESHOLD) with open(filename) as f: