1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import codecs
17import io
18import logging
19import os
20import tempfile
21import threading
22from collections.abc import Callable, Iterator
23from typing import IO, Any, TypeVar
24
25import boto3
26import botocore
27from boto3.exceptions import S3UploadFailedError
28from boto3.s3.transfer import TransferConfig
29from botocore.credentials import RefreshableCredentials
30from botocore.exceptions import ClientError, IncompleteReadError, ReadTimeoutError, ResponseStreamingError
31from botocore.session import get_session
32from dateutil.parser import parse as dateutil_parse
33
34from multistorageclient_rust import RustClient, RustClientError, RustRetryableError
35
36from ..constants import DEFAULT_CONNECT_TIMEOUT, DEFAULT_MAX_POOL_CONNECTIONS, DEFAULT_READ_TIMEOUT
37from ..rust_utils import parse_retry_config, run_async_rust_client_method
38from ..signers import CloudFrontURLSigner, URLSigner
39from ..telemetry import Telemetry
40from ..types import (
41 AWARE_DATETIME_MIN,
42 Credentials,
43 CredentialsProvider,
44 ObjectMetadata,
45 PreconditionFailedError,
46 Range,
47 RetryableError,
48 SignerType,
49 SuffixRange,
50 SymlinkHandling,
51)
52from ..utils import (
53 safe_makedirs,
54 split_path,
55 validate_attributes,
56)
57from .base import BaseStorageProvider, _translate_iterator_errors
58
59logger = logging.getLogger(__name__)
60
61_T = TypeVar("_T")
62
63MiB = 1024 * 1024
64
65# Python and Rust share the same multipart_threshold to keep the code simple.
66MULTIPART_THRESHOLD = 64 * MiB
67MULTIPART_CHUNKSIZE = 32 * MiB
68IO_CHUNKSIZE = 32 * MiB
69PYTHON_MAX_CONCURRENCY = 8
70MAX_POOL_CONNECTIONS = DEFAULT_MAX_POOL_CONNECTIONS
71
72PROVIDER = "s3"
73
74EXPRESS_ONEZONE_STORAGE_CLASS = "EXPRESS_ONEZONE"
75
76# Source: https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html
77SUPPORTED_CHECKSUM_ALGORITHMS = frozenset({"CRC32", "CRC32C", "SHA1", "SHA256", "CRC64NVME"})
78
79
[docs]
80class StaticS3CredentialsProvider(CredentialsProvider):
81 """
82 A concrete implementation of the :py:class:`multistorageclient.types.CredentialsProvider` that provides static S3 credentials.
83 """
84
85 _access_key: str
86 _secret_key: str
87 _session_token: str | None
88
89 def __init__(self, access_key: str, secret_key: str, session_token: str | None = None):
90 """
91 Initializes the :py:class:`StaticS3CredentialsProvider` with the provided access key, secret key, and optional
92 session token.
93
94 :param access_key: The access key for S3 authentication.
95 :param secret_key: The secret key for S3 authentication.
96 :param session_token: An optional session token for temporary credentials.
97 """
98 self._access_key = access_key
99 self._secret_key = secret_key
100 self._session_token = session_token
101
[docs]
102 def get_credentials(self) -> Credentials:
103 return Credentials(
104 access_key=self._access_key,
105 secret_key=self._secret_key,
106 token=self._session_token,
107 expiration=None,
108 )
109
[docs]
110 def refresh_credentials(self) -> None:
111 pass
112
113
114DEFAULT_PRESIGN_EXPIRES_IN = 3600
115
116_S3_METHOD_MAPPING: dict[str, str] = {
117 "GET": "get_object",
118 "PUT": "put_object",
119}
120
121
[docs]
122class S3URLSigner(URLSigner):
123 """Generates pre-signed URLs using the boto3 S3 client.
124
125 When the underlying credentials are temporary (STS, IAM role, EC2 instance
126 profile), the effective URL lifetime is the **shorter** of ``expires_in``
127 and the remaining credential lifetime — boto3 will not warn if the
128 credential expires before ``expires_in``.
129
130 See https://docs.aws.amazon.com/AmazonS3/latest/userguide/using-presigned-url.html
131 """
132
133 def __init__(self, s3_client: Any, bucket: str, expires_in: int = DEFAULT_PRESIGN_EXPIRES_IN) -> None:
134 self._s3_client = s3_client
135 self._bucket = bucket
136 self._expires_in = expires_in
137
[docs]
138 def generate_presigned_url(self, path: str, *, method: str = "GET") -> str:
139 client_method = _S3_METHOD_MAPPING.get(method.upper())
140 if client_method is None:
141 raise ValueError(f"Unsupported method for S3 presigning: {method!r}")
142 return self._s3_client.generate_presigned_url(
143 ClientMethod=client_method,
144 Params={"Bucket": self._bucket, "Key": path},
145 ExpiresIn=self._expires_in,
146 )
147
148
[docs]
149class S3StorageProvider(BaseStorageProvider):
150 """
151 A concrete implementation of the :py:class:`multistorageclient.types.StorageProvider` for interacting with Amazon S3 or S3-compatible object stores.
152 """
153
154 def __init__(
155 self,
156 region_name: str = "",
157 endpoint_url: str = "",
158 base_path: str = "",
159 credentials_provider: CredentialsProvider | None = None,
160 profile_name: str | None = None,
161 config_dict: dict[str, Any] | None = None,
162 telemetry_provider: Callable[[], Telemetry] | None = None,
163 verify: bool | str | None = None,
164 **kwargs: Any,
165 ) -> None:
166 """
167 Initializes the :py:class:`S3StorageProvider` with the region, endpoint URL, and optional credentials provider.
168
169 :param region_name: The AWS region where the S3 bucket is located.
170 :param endpoint_url: The custom endpoint URL for the S3 service.
171 :param base_path: The root prefix path within the S3 bucket where all operations will be scoped.
172 :param credentials_provider: The provider to retrieve S3 credentials.
173 :param profile_name: AWS shared configuration + credentials files profile. For :py:class:`boto3.session.Session`. Ignored if ``credentials_provider`` is set.
174 :param config_dict: Resolved MSC config.
175 :param telemetry_provider: A function that provides a telemetry instance.
176 :param verify: Controls SSL certificate verification.
177 Can be ``True`` (verify using system CA bundle, default), ``False`` (skip verification), or a string path to a custom CA certificate bundle.
178 :param request_checksum_calculation: For :py:class:`botocore.config.Config`.
179 When the underlying S3 client should calculate request checksums.
180 :param response_checksum_validation: For :py:class:`botocore.config.Config`.
181 When the underlying S3 client should validate response checksums.
182 :param max_pool_connections: For :py:class:`botocore.config.Config`.
183 The maximum number of connections to keep in a connection pool.
184 :param connect_timeout: For :py:class:`botocore.config.Config`.
185 The time in seconds till a timeout exception is thrown when attempting to make a connection.
186 :param read_timeout: For :py:class:`botocore.config.Config`.
187 The time in seconds till a timeout exception is thrown when attempting to read from a connection.
188 :param retries: For :py:class:`botocore.config.Config`.
189 A dictionary for configuration related to retry behavior.
190 :param s3: For :py:class:`botocore.config.Config`.
191 A dictionary of S3 specific configurations.
192 :param checksum_algorithm: Upload-only object integrity algorithm.
193 One of ``"CRC32"``, ``"CRC32C"``, ``"SHA1"``, ``"SHA256"``, ``"CRC64NVME"`` (case-insensitive).
194 When ``rust_client`` is enabled, only ``"SHA256"`` is accepted.
195 :param multipart_threshold: For :py:class:`boto3.s3.transfer.TransferConfig`.
196 The transfer size threshold for which multipart uploads, downloads, and copies will automatically be triggered.
197 :param max_concurrency: For :py:class:`boto3.s3.transfer.TransferConfig`.
198 The maximum number of threads that will be making requests to perform a transfer.
199 If ``use_threads`` is set to ``False``, the value provided is ignored as the transfer will only ever use the current thread.
200 :param multipart_chunksize: For :py:class:`boto3.s3.transfer.TransferConfig`.
201 The partition size of each part for a multipart transfer.
202 :param io_chunksize: For :py:class:`boto3.s3.transfer.TransferConfig`.
203 The max size of each chunk in the ``io`` queue. Currently, this is the size used when read is called on the downloaded stream as well.
204 Note: This value is ignored when resolved transfer manager type is CRTTransferManager.
205 """
206 super().__init__(
207 base_path=base_path,
208 provider_name=PROVIDER,
209 config_dict=config_dict,
210 telemetry_provider=telemetry_provider,
211 )
212
213 self._region_name = region_name
214 self._endpoint_url = endpoint_url
215 self._credentials_provider = credentials_provider
216 self._profile_name = profile_name
217 self._verify = verify
218
219 self._signature_version = kwargs.get("signature_version", "s3v4")
220 self._boto3_client: Any = None
221 self._s3_client_lock = threading.Lock()
222 self._s3_client_pid = os.getpid()
223 self._s3_client_options: dict[str, Any] = {
224 "request_checksum_calculation": kwargs.get("request_checksum_calculation"),
225 "response_checksum_validation": kwargs.get("response_checksum_validation"),
226 "max_pool_connections": kwargs.get("max_pool_connections", MAX_POOL_CONNECTIONS),
227 "connect_timeout": kwargs.get("connect_timeout", DEFAULT_CONNECT_TIMEOUT),
228 "read_timeout": kwargs.get("read_timeout", DEFAULT_READ_TIMEOUT),
229 "retries": kwargs.get("retries"),
230 "s3": kwargs.get("s3"),
231 }
232 if "rust_client" not in kwargs:
233 self._s3_client = self._create_s3_client(**self._s3_client_options)
234 self._transfer_config = TransferConfig(
235 multipart_threshold=int(kwargs.get("multipart_threshold", MULTIPART_THRESHOLD)),
236 max_concurrency=int(kwargs.get("max_concurrency", PYTHON_MAX_CONCURRENCY)),
237 multipart_chunksize=int(kwargs.get("multipart_chunksize", MULTIPART_CHUNKSIZE)),
238 io_chunksize=int(kwargs.get("io_chunksize", IO_CHUNKSIZE)),
239 use_threads=True,
240 )
241 self._multipart_threshold = self._transfer_config.multipart_threshold
242
243 self._signer_cache: dict[tuple, URLSigner] = {}
244
245 self._checksum_algorithm: str | None = self._validate_checksum_algorithm(kwargs.get("checksum_algorithm"))
246
247 self._rust_client = None
248 self._rust_bucket: str | None = None
249 if "rust_client" in kwargs:
250 # Inherit the rust client options from the kwargs
251 rust_client_options = kwargs["rust_client"]
252 if "max_pool_connections" in kwargs:
253 rust_client_options["max_pool_connections"] = kwargs["max_pool_connections"]
254 if "max_concurrency" in kwargs:
255 rust_client_options["max_concurrency"] = kwargs["max_concurrency"]
256 if "multipart_chunksize" in kwargs:
257 rust_client_options["multipart_chunksize"] = kwargs["multipart_chunksize"]
258 if "read_timeout" in kwargs:
259 rust_client_options["read_timeout"] = kwargs["read_timeout"]
260 if "connect_timeout" in kwargs:
261 rust_client_options["connect_timeout"] = kwargs["connect_timeout"]
262 if "checksum_algorithm" in kwargs:
263 rust_client_options["checksum_algorithm"] = kwargs["checksum_algorithm"]
264 if self._signature_version == "UNSIGNED":
265 rust_client_options["skip_signature"] = True
266
267 # Rust client only supports SHA256 today (object_store limitation).
268 rust_checksum = rust_client_options.get("checksum_algorithm")
269 if rust_checksum is not None:
270 if str(rust_checksum).upper() != "SHA256":
271 raise ValueError(
272 f"Rust client only supports checksum_algorithm='SHA256' today "
273 f"(object_store limitation), got '{rust_checksum!r}'. "
274 f"Disable rust_client or use SHA256."
275 )
276 rust_client_options["checksum_algorithm"] = "sha256"
277 self._rust_client = self._create_rust_client(rust_client_options)
278
279 @property
280 def _s3_client(self) -> Any:
281 if self._boto3_client is None:
282 if self._s3_client_pid != os.getpid():
283 # A forked child cannot release a lock held by another parent thread.
284 self._s3_client_lock = threading.Lock()
285 self._s3_client_pid = os.getpid()
286 # Rust-only reads avoid loading a separate set of botocore models per backend.
287 with self._s3_client_lock:
288 if self._boto3_client is None:
289 self._boto3_client = self._create_s3_client(**self._s3_client_options)
290 return self._boto3_client
291
292 @_s3_client.setter
293 def _s3_client(self, client: Any) -> None:
294 self._boto3_client = client
295
296 @staticmethod
297 def _validate_checksum_algorithm(value: str | None) -> str | None:
298 """
299 Normalizes the optional ``checksum_algorithm`` to uppercase, or returns ``None`` if disabled.
300 """
301 if value is None:
302 return None
303 if not isinstance(value, str) or not value:
304 raise ValueError(
305 f"checksum_algorithm must be one of {sorted(SUPPORTED_CHECKSUM_ALGORITHMS)} or None, got {value!r}."
306 )
307 normalized = value.upper()
308 if normalized not in SUPPORTED_CHECKSUM_ALGORITHMS:
309 raise ValueError(
310 f"Unsupported checksum_algorithm '{value}'. Supported: {sorted(SUPPORTED_CHECKSUM_ALGORITHMS)}."
311 )
312 return normalized
313
314 def _is_directory_bucket(self, bucket: str) -> bool:
315 """
316 Determines if the bucket is a directory bucket based on bucket name.
317 """
318 # S3 Express buckets have a specific naming convention
319 return "--x-s3" in bucket
320
321 def _create_s3_client(
322 self,
323 request_checksum_calculation: str | None = None,
324 response_checksum_validation: str | None = None,
325 max_pool_connections: int = MAX_POOL_CONNECTIONS,
326 connect_timeout: float | None = None,
327 read_timeout: float | None = None,
328 retries: dict[str, Any] | None = None,
329 s3: dict[str, Any] | None = None,
330 ):
331 """
332 Creates and configures the boto3 S3 client, using refreshable credentials if possible.
333
334 :param request_checksum_calculation: For :py:class:`botocore.config.Config`. When the underlying S3 client should calculate request checksums. See the equivalent option in the `AWS configuration file <https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#using-a-configuration-file>`_.
335 :param response_checksum_validation: For :py:class:`botocore.config.Config`. When the underlying S3 client should validate response checksums. See the equivalent option in the `AWS configuration file <https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#using-a-configuration-file>`_.
336 :param max_pool_connections: For :py:class:`botocore.config.Config`. The maximum number of connections to keep in a connection pool.
337 :param connect_timeout: For :py:class:`botocore.config.Config`. The time in seconds till a timeout exception is thrown when attempting to make a connection.
338 :param read_timeout: For :py:class:`botocore.config.Config`. The time in seconds till a timeout exception is thrown when attempting to read from a connection.
339 :param retries: For :py:class:`botocore.config.Config`. A dictionary for configuration related to retry behavior.
340 :param s3: For :py:class:`botocore.config.Config`. A dictionary of S3 specific configurations.
341
342 :return: The configured S3 client.
343 """
344 options = {
345 # https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
346 "config": boto3.session.Config( # pyright: ignore [reportAttributeAccessIssue]
347 max_pool_connections=max_pool_connections,
348 connect_timeout=connect_timeout,
349 read_timeout=read_timeout,
350 retries=retries or {"mode": "standard"},
351 request_checksum_calculation=request_checksum_calculation,
352 response_checksum_validation=response_checksum_validation,
353 s3=s3,
354 ),
355 }
356
357 if self._region_name:
358 options["region_name"] = self._region_name
359
360 if self._endpoint_url:
361 options["endpoint_url"] = self._endpoint_url
362
363 if self._verify is not None:
364 options["verify"] = self._verify
365
366 if self._signature_version:
367 signature_config = botocore.config.Config( # pyright: ignore[reportAttributeAccessIssue]
368 signature_version=botocore.UNSIGNED
369 if self._signature_version == "UNSIGNED"
370 else self._signature_version
371 )
372 options["config"] = options["config"].merge(signature_config)
373
374 if self._credentials_provider:
375 creds = self._fetch_credentials()
376 if creds.get("expiry_time"):
377 # Use RefreshableCredentials if expiry_time provided.
378 refreshable_credentials = RefreshableCredentials.create_from_metadata(
379 metadata=creds, refresh_using=self._fetch_credentials, method="custom-refresh"
380 )
381
382 botocore_session = get_session()
383 botocore_session._credentials = refreshable_credentials
384
385 boto3_session = boto3.Session(botocore_session=botocore_session)
386
387 return boto3_session.client("s3", **options)
388 else:
389 # Add static credentials to the options dictionary
390 options["aws_access_key_id"] = creds["access_key"]
391 options["aws_secret_access_key"] = creds["secret_key"]
392 if creds["token"]:
393 options["aws_session_token"] = creds["token"]
394
395 # Fallback to standard credential chain.
396 session = boto3.Session(profile_name=self._profile_name)
397 return session.client("s3", **options)
398
399 def _create_rust_client(self, rust_client_options: dict[str, Any] | None = None):
400 """
401 Creates and configures the rust client, using refreshable credentials if possible.
402 """
403 configs = dict(rust_client_options) if rust_client_options else {}
404
405 # Extract and parse retry configuration
406 retry_config = parse_retry_config(configs)
407
408 if self._region_name and "region_name" not in configs:
409 configs["region_name"] = self._region_name
410
411 if self._endpoint_url and "endpoint_url" not in configs:
412 configs["endpoint_url"] = self._endpoint_url
413
414 if self._profile_name and "profile_name" not in configs:
415 configs["profile_name"] = self._profile_name
416
417 # If the user specifies a bucket, use it. Otherwise, use the base path.
418 if "bucket" not in configs:
419 bucket, _ = split_path(self._base_path)
420 configs["bucket"] = bucket
421 self._rust_bucket = configs["bucket"]
422
423 if "max_pool_connections" not in configs:
424 configs["max_pool_connections"] = MAX_POOL_CONNECTIONS
425
426 return RustClient(
427 provider=PROVIDER,
428 configs=configs,
429 credentials_provider=self._credentials_provider,
430 retry=retry_config,
431 )
432
433 def _fetch_credentials(self) -> dict:
434 """
435 Refreshes the S3 client if the current credentials are expired.
436 """
437 if not self._credentials_provider:
438 raise RuntimeError("Cannot fetch credentials if no credential provider configured.")
439 self._credentials_provider.refresh_credentials()
440 credentials = self._credentials_provider.get_credentials()
441 return {
442 "access_key": credentials.access_key,
443 "secret_key": credentials.secret_key,
444 "token": credentials.token,
445 "expiry_time": credentials.expiration,
446 }
447
448 def _translate_errors(
449 self,
450 func: Callable[[], _T],
451 operation: str,
452 bucket: str,
453 key: str,
454 ) -> _T:
455 """
456 Translates errors like timeouts and client errors.
457
458 :param func: The function that performs the actual S3 operation.
459 :param operation: The type of operation being performed (e.g., "PUT", "GET", "DELETE").
460 :param bucket: The name of the S3 bucket involved in the operation.
461 :param key: The key of the object within the S3 bucket.
462
463 :return: The result of the S3 operation, typically the return value of the `func` callable.
464 """
465 try:
466 return func()
467 except ClientError as error:
468 status_code = error.response["ResponseMetadata"]["HTTPStatusCode"]
469 request_id = error.response["ResponseMetadata"].get("RequestId")
470 host_id = error.response["ResponseMetadata"].get("HostId")
471 error_code = error.response["Error"]["Code"]
472 error_info = f"request_id: {request_id}, host_id: {host_id}, status_code: {status_code}"
473
474 if status_code == 404:
475 if error_code == "NoSuchUpload":
476 error_message = error.response["Error"]["Message"]
477 raise RetryableError(f"Multipart upload failed for {bucket}/{key}: {error_message}") from error
478 raise FileNotFoundError(f"Object {bucket}/{key} does not exist. {error_info}") # pylint: disable=raise-missing-from
479 elif status_code == 403:
480 raise PermissionError(
481 f"Permission denied to {operation} object(s) at {bucket}/{key}. {error_info}"
482 ) from error
483 elif status_code == 412: # Precondition Failed
484 raise PreconditionFailedError(
485 f"ETag mismatch for {operation} operation on {bucket}/{key}. {error_info}"
486 ) from error
487 elif status_code == 429:
488 raise RetryableError(
489 f"Too many request to {operation} object(s) at {bucket}/{key}. {error_info}"
490 ) from error
491 elif status_code == 503:
492 raise RetryableError(
493 f"Service unavailable when {operation} object(s) at {bucket}/{key}. {error_info}"
494 ) from error
495 elif status_code == 501:
496 raise NotImplementedError(
497 f"Operation {operation} not implemented for object(s) at {bucket}/{key}. {error_info}"
498 ) from error
499 elif status_code == 408:
500 # 408 Request Timeout is from Google Cloud Storage
501 raise RetryableError(
502 f"Request timeout when {operation} object(s) at {bucket}/{key}. {error_info}"
503 ) from error
504 else:
505 raise RuntimeError(
506 f"Failed to {operation} object(s) at {bucket}/{key}. {error_info}, "
507 f"error_type: {type(error).__name__}"
508 ) from error
509 except RustClientError as error:
510 message = error.args[0]
511 status_code = error.args[1]
512 if status_code == 404:
513 raise FileNotFoundError(f"Object {bucket}/{key} does not exist. {message}") from error
514 elif status_code == 403:
515 raise PermissionError(
516 f"Permission denied to {operation} object(s) at {bucket}/{key}. {message}"
517 ) from error
518 else:
519 raise RetryableError(
520 f"Failed to {operation} object(s) at {bucket}/{key}. {message}. status_code: {status_code}"
521 ) from error
522 except (ReadTimeoutError, IncompleteReadError, ResponseStreamingError) as error:
523 raise RetryableError(
524 f"Failed to {operation} object(s) at {bucket}/{key} due to network timeout or incomplete read. "
525 f"error_type: {type(error).__name__}"
526 ) from error
527 except RustRetryableError as error:
528 raise RetryableError(
529 f"Failed to {operation} object(s) at {bucket}/{key} due to retryable error from Rust. "
530 f"error_type: {type(error).__name__}"
531 ) from error
532 except Exception as error:
533 if (
534 isinstance(error, S3UploadFailedError)
535 and isinstance(error.__context__, ClientError)
536 and error.__context__.response["ResponseMetadata"]["HTTPStatusCode"] == 403
537 ):
538 raise PermissionError(
539 f"Permission denied to {operation} object(s) at {bucket}/{key}. {error}"
540 ) from error
541 raise RuntimeError(
542 f"Failed to {operation} object(s) at {bucket}/{key}, error type: {type(error).__name__}, error: {error}"
543 ) from error
544
545 def _put_object(
546 self,
547 path: str,
548 body: bytes,
549 if_match: str | None = None,
550 if_none_match: str | None = None,
551 attributes: dict[str, str] | None = None,
552 content_type: str | None = None,
553 ) -> int:
554 """
555 Uploads an object to the specified S3 path.
556
557 :param path: The S3 path where the object will be uploaded.
558 :param body: The content of the object as bytes.
559 :param if_match: Optional If-Match header value. Use "*" to only upload if the object doesn't exist.
560 :param if_none_match: Optional If-None-Match header value. Use "*" to only upload if the object doesn't exist.
561 :param attributes: Optional attributes to attach to the object.
562 :param content_type: Optional Content-Type header value.
563 """
564 bucket, key = split_path(path)
565
566 def _invoke_api() -> int:
567 kwargs = {"Bucket": bucket, "Key": key, "Body": body}
568 if content_type:
569 kwargs["ContentType"] = content_type
570 if self._is_directory_bucket(bucket):
571 kwargs["StorageClass"] = EXPRESS_ONEZONE_STORAGE_CLASS
572 if if_match:
573 kwargs["IfMatch"] = if_match
574 if if_none_match:
575 kwargs["IfNoneMatch"] = if_none_match
576 validated_attributes = validate_attributes(attributes)
577 if validated_attributes:
578 kwargs["Metadata"] = validated_attributes
579
580 # TODO(NGCDP-5804): Add support to update ContentType header in Rust client
581 rust_unsupported_feature_keys = {"StorageClass", "IfMatch", "IfNoneMatch", "ContentType"}
582 if (
583 self._rust_client
584 # Rust client doesn't support creating objects with trailing /, see https://github.com/apache/arrow-rs/issues/7026
585 and not path.endswith("/")
586 and all(key not in kwargs for key in rust_unsupported_feature_keys)
587 ):
588 rust_attributes = {"attributes": validated_attributes} if validated_attributes else {}
589 run_async_rust_client_method(self._rust_client, "put", key, body, **rust_attributes)
590 else:
591 if self._checksum_algorithm:
592 kwargs["ChecksumAlgorithm"] = self._checksum_algorithm
593 self._s3_client.put_object(**kwargs)
594
595 return len(body)
596
597 return self._translate_errors(_invoke_api, operation="PUT", bucket=bucket, key=key)
598
599 def _supports_suffix_range(self) -> bool:
600 return self._rust_client is None
601
602 def _get_object(self, path: str, byte_range: Range | SuffixRange | None = None) -> bytes:
603 bucket, key = split_path(path)
604
605 if isinstance(byte_range, SuffixRange):
606 suffix_range = byte_range
607
608 def _invoke_suffix_api() -> bytes | None:
609 try:
610 response = self._s3_client.get_object(Bucket=bucket, Key=key, Range=f"bytes=-{suffix_range.size}")
611 except ClientError as error:
612 # Raised for empty objects, or by servers without suffix range support.
613 if error.response["ResponseMetadata"]["HTTPStatusCode"] == 416:
614 return None
615 raise
616 # Servers may ignore the range and return the whole object.
617 return response["Body"].read()[-suffix_range.size :]
618
619 return self._get_object_suffix(
620 path,
621 suffix_range,
622 lambda: self._translate_errors(_invoke_suffix_api, operation="GET", bucket=bucket, key=key),
623 )
624
625 def _invoke_api() -> bytes:
626 if byte_range:
627 bytes_range = f"bytes={byte_range.offset}-{byte_range.offset + byte_range.size - 1}"
628 if self._rust_client:
629 response = run_async_rust_client_method(
630 self._rust_client,
631 "get",
632 key,
633 byte_range,
634 )
635 return response
636 else:
637 response = self._s3_client.get_object(Bucket=bucket, Key=key, Range=bytes_range)
638 else:
639 if self._rust_client:
640 response = run_async_rust_client_method(self._rust_client, "get", key)
641 return response
642 else:
643 response = self._s3_client.get_object(Bucket=bucket, Key=key)
644
645 return response["Body"].read()
646
647 return self._translate_errors(_invoke_api, operation="GET", bucket=bucket, key=key)
648
649 def _copy_object(self, src_path: str, dest_path: str) -> int:
650 src_bucket, src_key = split_path(src_path)
651 dest_bucket, dest_key = split_path(dest_path)
652
653 src_object = self._get_object_metadata(src_path)
654
655 def _invoke_api() -> int:
656 self._s3_client.copy(
657 CopySource={"Bucket": src_bucket, "Key": src_key},
658 Bucket=dest_bucket,
659 Key=dest_key,
660 Config=self._transfer_config,
661 )
662
663 return src_object.content_length
664
665 return self._translate_errors(_invoke_api, operation="COPY", bucket=dest_bucket, key=dest_key)
666
667 def _delete_object(self, path: str, if_match: str | None = None) -> None:
668 bucket, key = split_path(path)
669
670 def _invoke_api() -> None:
671 # Delete conditionally when if_match (etag) is provided; otherwise delete unconditionally
672 if if_match:
673 self._s3_client.delete_object(Bucket=bucket, Key=key, IfMatch=if_match)
674 else:
675 self._s3_client.delete_object(Bucket=bucket, Key=key)
676
677 return self._translate_errors(_invoke_api, operation="DELETE", bucket=bucket, key=key)
678
679 def _delete_objects(self, paths: list[str]) -> None:
680 if not paths:
681 return
682
683 by_bucket: dict[str, list[str]] = {}
684 for p in paths:
685 bucket, key = split_path(p)
686 by_bucket.setdefault(bucket, []).append(key)
687
688 S3_BATCH_LIMIT = 1000
689
690 def _invoke_api() -> None:
691 all_errors: list[str] = []
692 for bucket, keys in by_bucket.items():
693 for i in range(0, len(keys), S3_BATCH_LIMIT):
694 chunk = keys[i : i + S3_BATCH_LIMIT]
695 response = self._s3_client.delete_objects(
696 Bucket=bucket, Delete={"Objects": [{"Key": k} for k in chunk]}
697 )
698 errors = response.get("Errors") or []
699 for e in errors:
700 all_errors.append(f"{bucket}/{e.get('Key', '?')}: {e.get('Code', '')} {e.get('Message', '')}")
701 if all_errors:
702 raise RuntimeError(f"DeleteObjects reported errors: {'; '.join(all_errors)}")
703
704 bucket_desc = "(" + "|".join(by_bucket) + ")"
705 key_desc = "(" + "|".join(str(len(keys)) for keys in by_bucket.values()) + " keys)"
706 self._translate_errors(_invoke_api, operation="DELETE_MANY", bucket=bucket_desc, key=key_desc)
707
708 def _can_use_rust_metadata(self, bucket: str, key: str) -> bool:
709 # S3 keys must retain segments that the Rust store rejects or strips.
710 return (
711 self._rust_client is not None
712 and self._verify is None
713 and bucket == self._rust_bucket
714 and not self._is_directory_bucket(bucket)
715 and RustClient.supports_path(key)
716 )
717
718 def _is_dir(self, path: str) -> bool:
719 # Ensure the path ends with '/' to mimic a directory
720 path = self._append_delimiter(path)
721
722 bucket, key = split_path(path)
723
724 def _invoke_api() -> bool:
725 if self._can_use_rust_metadata(bucket, key.removesuffix("/")):
726 try:
727 return run_async_rust_client_method(self._rust_client, "is_dir", key)
728 except ValueError:
729 # A valid prefix can contain children whose keys Rust cannot represent.
730 pass
731 # List objects with the given prefix
732 response = self._s3_client.list_objects_v2(Bucket=bucket, Prefix=key, MaxKeys=1, Delimiter="/")
733
734 # Check if there are any contents or common prefixes
735 return bool(response.get("Contents", []) or response.get("CommonPrefixes", []))
736
737 return self._translate_errors(_invoke_api, operation="LIST", bucket=bucket, key=key)
738
739 def _make_symlink(self, path: str, target: str) -> None:
740 bucket, key = split_path(path)
741 target_bucket, target_key = split_path(target)
742 if bucket != target_bucket:
743 raise ValueError(f"Cannot create cross-bucket symlink: '{bucket}' -> '{target_bucket}'.")
744 relative_target = ObjectMetadata.encode_symlink_target(key, target_key)
745
746 def _invoke_api() -> None:
747 self._s3_client.put_object(
748 Bucket=bucket,
749 Key=key,
750 Body=b"",
751 Metadata={"msc-symlink-target": relative_target},
752 )
753
754 self._translate_errors(_invoke_api, operation="PUT", bucket=bucket, key=key)
755
756 def _get_object_metadata(self, path: str, strict: bool = True) -> ObjectMetadata:
757 bucket, key = split_path(path)
758 if path.endswith("/") or (bucket and not key):
759 # If path ends with "/" or empty key name is provided, then assume it's a "directory",
760 # which metadata is not guaranteed to exist for cases such as
761 # "virtual prefix" that was never explicitly created.
762 if self._is_dir(path):
763 return ObjectMetadata(
764 key=path,
765 type="directory",
766 content_length=0,
767 last_modified=AWARE_DATETIME_MIN,
768 )
769 else:
770 raise FileNotFoundError(f"Directory {path} does not exist.")
771 else:
772
773 def _invoke_api() -> ObjectMetadata:
774 if self._can_use_rust_metadata(bucket, key):
775 result = run_async_rust_client_method(self._rust_client, "head", key)
776 object_metadata = result.object_metadata
777 return ObjectMetadata(
778 key=path,
779 type="file",
780 content_length=object_metadata.content_length,
781 content_type=result.content_type,
782 last_modified=dateutil_parse(object_metadata.last_modified),
783 etag=object_metadata.etag.strip('"') if object_metadata.etag is not None else None,
784 storage_class=result.storage_class,
785 metadata=result.metadata,
786 symlink_target=result.metadata.get("msc-symlink-target"),
787 )
788 response = self._s3_client.head_object(Bucket=bucket, Key=key)
789 user_metadata = response.get("Metadata")
790 symlink_target = user_metadata.get("msc-symlink-target") if user_metadata else None
791
792 return ObjectMetadata(
793 key=path,
794 type="file",
795 content_length=response["ContentLength"],
796 content_type=response.get("ContentType"),
797 last_modified=response.get("LastModified", AWARE_DATETIME_MIN),
798 etag=response["ETag"].strip('"') if "ETag" in response else None,
799 storage_class=response.get("StorageClass"),
800 metadata=user_metadata,
801 symlink_target=symlink_target,
802 )
803
804 try:
805 return self._translate_errors(_invoke_api, operation="HEAD", bucket=bucket, key=key)
806 except FileNotFoundError:
807 if strict:
808 # If the object does not exist on the given path, we will append a trailing slash and
809 # check if the path is a directory.
810 path = self._append_delimiter(path)
811 if self._is_dir(path):
812 return ObjectMetadata(
813 key=path,
814 type="directory",
815 content_length=0,
816 last_modified=AWARE_DATETIME_MIN,
817 )
818 raise
819
820 def _list_objects(
821 self,
822 path: str,
823 start_after: str | None = None,
824 end_at: str | None = None,
825 include_directories: bool = False,
826 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
827 ) -> Iterator[ObjectMetadata]:
828 bucket, prefix = split_path(path)
829
830 # Get the prefix of the start_after and end_at paths relative to the bucket.
831 if start_after:
832 _, start_after = split_path(start_after)
833 if end_at:
834 _, end_at = split_path(end_at)
835
836 def _invoke_api() -> Iterator[ObjectMetadata]:
837 paginator = self._s3_client.get_paginator("list_objects_v2")
838 if include_directories:
839 page_iterator = paginator.paginate(
840 Bucket=bucket, Prefix=prefix, Delimiter="/", StartAfter=(start_after or "")
841 )
842 else:
843 page_iterator = paginator.paginate(Bucket=bucket, Prefix=prefix, StartAfter=(start_after or ""))
844
845 for page in page_iterator:
846 # A page holds both CommonPrefixes and Contents for the same key window, so a prefix past
847 # end_at must not stop the listing before the page's objects have been processed.
848 past_end_at = False
849 for item in page.get("CommonPrefixes", []):
850 prefix_key = item["Prefix"].rstrip("/")
851 # Filter by start_after and end_at - S3's StartAfter doesn't filter CommonPrefixes
852 if (start_after is None or start_after < prefix_key) and (end_at is None or prefix_key <= end_at):
853 yield ObjectMetadata(
854 key=os.path.join(bucket, prefix_key),
855 type="directory",
856 content_length=0,
857 last_modified=AWARE_DATETIME_MIN,
858 )
859 elif end_at is not None and end_at < prefix_key:
860 past_end_at = True
861 break
862
863 # S3 guarantees lexicographical order for general purpose buckets (for
864 # normal S3) but not directory buckets (for S3 Express One Zone).
865 for response_object in page.get("Contents", []):
866 key = response_object["Key"]
867 if end_at is None or key <= end_at:
868 if key.endswith("/"):
869 if include_directories:
870 yield ObjectMetadata(
871 key=os.path.join(bucket, key.rstrip("/")),
872 type="directory",
873 content_length=0,
874 last_modified=response_object["LastModified"],
875 )
876 else:
877 symlink_target = None
878 if response_object.get("Size", 0) == 0:
879 try:
880 meta = self._get_object_metadata(os.path.join(bucket, key))
881 symlink_target = meta.symlink_target
882 except Exception:
883 symlink_target = None
884 yield ObjectMetadata(
885 key=os.path.join(bucket, key),
886 type="file",
887 content_length=response_object["Size"],
888 last_modified=response_object["LastModified"],
889 etag=response_object["ETag"].strip('"'),
890 storage_class=response_object.get("StorageClass"),
891 symlink_target=symlink_target,
892 )
893 else:
894 return
895
896 if past_end_at:
897 return
898
899 return _translate_iterator_errors(
900 _invoke_api(), lambda f: self._translate_errors(f, operation="LIST", bucket=bucket, key=prefix)
901 )
902
903 @property
904 def supports_parallel_listing(self) -> bool:
905 """
906 S3 supports parallel listing via delimiter-based prefix discovery.
907
908 Note: Directory bucket handling is done in list_objects_recursive().
909 """
910 return True
911
[docs]
912 def list_objects_recursive(
913 self,
914 path: str = "",
915 start_after: str | None = None,
916 end_at: str | None = None,
917 max_workers: int = 32,
918 look_ahead: int = 2,
919 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
920 ) -> Iterator[ObjectMetadata]:
921 """
922 List all objects recursively using parallel prefix discovery for improved performance.
923
924 For S3, uses the Rust client's list_recursive when available for maximum performance.
925 Falls back to Python implementation otherwise.
926
927 Returns files only (no directories), in lexicographic order.
928
929 :param symlink_handling: How to handle symbolic links during listing.
930 """
931 if (start_after is not None) and (end_at is not None) and not (start_after < end_at):
932 raise ValueError(f"start_after ({start_after}) must be before end_at ({end_at})!")
933
934 full_path = self._prepend_base_path(path)
935 bucket, _ = split_path(full_path)
936
937 if self._is_directory_bucket(bucket):
938 yield from self.list_objects(
939 path, start_after, end_at, include_directories=False, symlink_handling=symlink_handling
940 )
941 return
942
943 if self._rust_client:
944 yield from self._emit_metrics(
945 operation=BaseStorageProvider._Operation.LIST,
946 f=lambda: self._list_objects_recursive_rust(path, full_path, bucket, start_after, end_at, max_workers),
947 )
948 else:
949 yield from super().list_objects_recursive(
950 path, start_after, end_at, max_workers, look_ahead, symlink_handling
951 )
952
953 def _list_objects_recursive_rust(
954 self,
955 path: str,
956 full_path: str,
957 bucket: str,
958 start_after: str | None,
959 end_at: str | None,
960 max_workers: int,
961 ) -> Iterator[ObjectMetadata]:
962 """
963 Use Rust client's list_recursive for parallel listing.
964
965 The Rust client already handles parallel listing internally.
966 Returns files only in lexicographic order.
967 """
968 _, prefix = split_path(full_path)
969
970 def _invoke_api() -> Iterator[ObjectMetadata]:
971 result = run_async_rust_client_method(
972 self._rust_client,
973 "list_recursive",
974 [prefix] if prefix else [""],
975 max_concurrency=max_workers,
976 )
977
978 start_after_full = self._prepend_base_path(start_after) if start_after else None
979 end_at_full = self._prepend_base_path(end_at) if end_at else None
980
981 for obj in result.objects:
982 full_key = os.path.join(bucket, obj.key)
983
984 if start_after_full and full_key <= start_after_full:
985 continue
986 if end_at_full and full_key > end_at_full:
987 break
988
989 relative_key = full_key.removeprefix(self._base_path).lstrip("/")
990
991 yield ObjectMetadata(
992 key=relative_key,
993 content_length=obj.content_length,
994 last_modified=dateutil_parse(obj.last_modified),
995 type="file" if obj.object_type == "object" else obj.object_type,
996 etag=obj.etag,
997 )
998
999 yield from _translate_iterator_errors(
1000 _invoke_api(), lambda f: self._translate_errors(f, operation="LIST", bucket=bucket, key=prefix)
1001 )
1002
1003 def _upload_file(
1004 self,
1005 remote_path: str,
1006 f: str | IO,
1007 attributes: dict[str, str] | None = None,
1008 content_type: str | None = None,
1009 ) -> int:
1010 file_size: int = 0
1011
1012 if isinstance(f, str):
1013 bucket, key = split_path(remote_path)
1014 file_size = os.path.getsize(f)
1015
1016 # Upload small files
1017 if file_size <= self._multipart_threshold:
1018 if self._rust_client and not content_type and not self._is_directory_bucket(bucket):
1019 validated_attributes = validate_attributes(attributes)
1020 rust_attributes = {"attributes": validated_attributes} if validated_attributes else {}
1021 self._translate_errors(
1022 lambda: run_async_rust_client_method(self._rust_client, "upload", f, key, **rust_attributes),
1023 operation="PUT",
1024 bucket=bucket,
1025 key=key,
1026 )
1027 else:
1028 with open(f, "rb") as fp:
1029 self._put_object(remote_path, fp.read(), attributes=attributes, content_type=content_type)
1030 return file_size
1031
1032 # Upload large files using TransferConfig
1033 def _invoke_api() -> int:
1034 extra_args = {}
1035 if content_type:
1036 extra_args["ContentType"] = content_type
1037 if self._is_directory_bucket(bucket):
1038 extra_args["StorageClass"] = EXPRESS_ONEZONE_STORAGE_CLASS
1039 validated_attributes = validate_attributes(attributes)
1040 if validated_attributes:
1041 extra_args["Metadata"] = validated_attributes
1042 if self._rust_client and "ContentType" not in extra_args and "StorageClass" not in extra_args:
1043 rust_attributes = {"attributes": validated_attributes} if validated_attributes else {}
1044 run_async_rust_client_method(
1045 self._rust_client, "upload_multipart_from_file", f, key, **rust_attributes
1046 )
1047 else:
1048 if self._checksum_algorithm:
1049 extra_args["ChecksumAlgorithm"] = self._checksum_algorithm
1050 self._s3_client.upload_file(
1051 Filename=f,
1052 Bucket=bucket,
1053 Key=key,
1054 Config=self._transfer_config,
1055 ExtraArgs=extra_args,
1056 )
1057
1058 return file_size
1059
1060 return self._translate_errors(_invoke_api, operation="PUT", bucket=bucket, key=key)
1061 else:
1062 # Upload small files
1063 f.seek(0, io.SEEK_END)
1064 file_size = f.tell()
1065 f.seek(0)
1066
1067 if file_size <= self._multipart_threshold:
1068 if isinstance(f, io.StringIO):
1069 self._put_object(
1070 remote_path, f.read().encode("utf-8"), attributes=attributes, content_type=content_type
1071 )
1072 else:
1073 self._put_object(remote_path, f.read(), attributes=attributes, content_type=content_type)
1074 return file_size
1075
1076 # Upload large files using TransferConfig
1077 bucket, key = split_path(remote_path)
1078
1079 def _invoke_api() -> int:
1080 extra_args = {}
1081 if content_type:
1082 extra_args["ContentType"] = content_type
1083 if self._is_directory_bucket(bucket):
1084 extra_args["StorageClass"] = EXPRESS_ONEZONE_STORAGE_CLASS
1085 validated_attributes = validate_attributes(attributes)
1086 if validated_attributes:
1087 extra_args["Metadata"] = validated_attributes
1088
1089 if (
1090 self._rust_client
1091 and isinstance(f, io.BytesIO)
1092 and "ContentType" not in extra_args
1093 and "StorageClass" not in extra_args
1094 ):
1095 data = f.getbuffer()
1096 rust_attributes = {"attributes": validated_attributes} if validated_attributes else {}
1097 try:
1098 run_async_rust_client_method(
1099 self._rust_client, "upload_multipart_from_bytes", key, data, **rust_attributes
1100 )
1101 finally:
1102 data.release()
1103 else:
1104 if self._checksum_algorithm:
1105 extra_args["ChecksumAlgorithm"] = self._checksum_algorithm
1106 self._s3_client.upload_fileobj(
1107 Fileobj=f,
1108 Bucket=bucket,
1109 Key=key,
1110 Config=self._transfer_config,
1111 ExtraArgs=extra_args,
1112 )
1113
1114 return file_size
1115
1116 return self._translate_errors(_invoke_api, operation="PUT", bucket=bucket, key=key)
1117
1118 def _download_file(self, remote_path: str, f: str | IO, metadata: ObjectMetadata | None = None) -> int:
1119 """
1120 Downloads an object to a local path or file-like object.
1121
1122 Python downloads to paths stage data in a temporary file and replace the destination only after the transfer
1123 succeeds. Transfer or rename failures preserve the existing destination and trigger best-effort cleanup of the
1124 staging file. Cleanup failures are logged without masking the original error. Rust downloads use the Rust
1125 client's destination and resumable state handling.
1126
1127 :param remote_path: The object path to download.
1128 :param f: The destination file path or writable file-like object.
1129 :param metadata: Optional object metadata to avoid an additional lookup.
1130 :return: The downloaded object's content length.
1131 """
1132 if metadata is None:
1133 metadata = self._get_object_metadata(remote_path)
1134
1135 if isinstance(f, str):
1136 bucket, key = split_path(remote_path)
1137 if os.path.dirname(f):
1138 safe_makedirs(os.path.dirname(f))
1139
1140 # Download small files
1141 if metadata.content_length <= self._multipart_threshold:
1142 if self._rust_client:
1143 self._translate_errors(
1144 lambda: run_async_rust_client_method(self._rust_client, "download", key, f),
1145 operation="GET",
1146 bucket=bucket,
1147 key=key,
1148 )
1149 else:
1150 temp_file_path: str | None = None
1151 try:
1152 with tempfile.NamedTemporaryFile(
1153 mode="wb", delete=False, dir=os.path.dirname(f), prefix="."
1154 ) as fp:
1155 temp_file_path = fp.name
1156 fp.write(self._get_object(remote_path))
1157 os.rename(src=temp_file_path, dst=f)
1158 finally:
1159 if temp_file_path:
1160 try:
1161 os.unlink(temp_file_path)
1162 except FileNotFoundError:
1163 pass
1164 except OSError as error:
1165 logger.warning("Could not remove temporary file %s: %s", temp_file_path, error)
1166 return metadata.content_length
1167
1168 # Download large files using TransferConfig
1169 def _invoke_api() -> int:
1170 if self._rust_client:
1171 run_async_rust_client_method(self._rust_client, "download_multipart_to_file", key, f)
1172 else:
1173 temp_file_path: str | None = None
1174 try:
1175 with tempfile.NamedTemporaryFile(
1176 mode="wb", delete=False, dir=os.path.dirname(f), prefix="."
1177 ) as fp:
1178 temp_file_path = fp.name
1179 self._s3_client.download_fileobj(
1180 Bucket=bucket,
1181 Key=key,
1182 Fileobj=fp,
1183 Config=self._transfer_config,
1184 )
1185 os.rename(src=temp_file_path, dst=f)
1186 finally:
1187 if temp_file_path:
1188 try:
1189 os.unlink(temp_file_path)
1190 except FileNotFoundError:
1191 pass
1192 except OSError as error:
1193 logger.warning("Could not remove temporary file %s: %s", temp_file_path, error)
1194
1195 return metadata.content_length
1196
1197 return self._translate_errors(_invoke_api, operation="GET", bucket=bucket, key=key)
1198 else:
1199 # Download small files
1200 if metadata.content_length <= self._multipart_threshold:
1201 response = self._get_object(remote_path)
1202 # Python client returns `bytes`, but Rust client returns an object that implements the buffer protocol,
1203 # so we need to check whether `.decode()` is available.
1204 if isinstance(f, io.StringIO):
1205 if hasattr(response, "decode"):
1206 f.write(response.decode("utf-8"))
1207 else:
1208 f.write(codecs.decode(memoryview(response), "utf-8"))
1209 else:
1210 f.write(response)
1211 return metadata.content_length
1212
1213 # Download large files using TransferConfig
1214 bucket, key = split_path(remote_path)
1215
1216 def _invoke_api() -> int:
1217 self._s3_client.download_fileobj(
1218 Bucket=bucket,
1219 Key=key,
1220 Fileobj=f,
1221 Config=self._transfer_config,
1222 )
1223
1224 return metadata.content_length
1225
1226 return self._translate_errors(_invoke_api, operation="GET", bucket=bucket, key=key)
1227
1228 def _generate_presigned_url(
1229 self,
1230 path: str,
1231 *,
1232 method: str = "GET",
1233 signer_type: SignerType | None = None,
1234 signer_options: dict[str, Any] | None = None,
1235 ) -> str:
1236 options = signer_options or {}
1237 bucket, key = split_path(path)
1238
1239 if signer_type is None or signer_type == SignerType.S3:
1240 expires_in = int(options.get("expires_in", DEFAULT_PRESIGN_EXPIRES_IN))
1241 cache_key: tuple = (SignerType.S3, bucket, expires_in)
1242 if cache_key not in self._signer_cache:
1243 self._signer_cache[cache_key] = S3URLSigner(self._s3_client, bucket, expires_in=expires_in)
1244 elif signer_type == SignerType.CLOUDFRONT:
1245 cache_key = (SignerType.CLOUDFRONT, frozenset(options.items()))
1246 if cache_key not in self._signer_cache:
1247 self._signer_cache[cache_key] = CloudFrontURLSigner(**options)
1248 else:
1249 raise ValueError(f"Unsupported signer type for S3 provider: {signer_type!r}")
1250
1251 return self._signer_cache[cache_key].generate_presigned_url(key, method=method)