Source code for multistorageclient.providers.gcs_s3

 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
16from .s3 import S3StorageProvider
17
18PROVIDER = "gcs_s3"
19
20

[docs]
21class GoogleS3StorageProvider(S3StorageProvider):
22    """
23    A concrete implementation of the :py:class:`multistorageclient.types.StorageProvider` for interacting with GCS via its S3 interface.
24    """
25
26    def __init__(self, *args, **kwargs):
27        rust_client_opts = kwargs.get("rust_client") or {}
28        if kwargs.get("checksum_algorithm") is not None or rust_client_opts.get("checksum_algorithm") is not None:
29            raise ValueError("checksum_algorithm is not supported for gcs_s3 provider.")
30
31        kwargs["request_checksum_calculation"] = "when_required"
32        kwargs["response_checksum_validation"] = "when_required"
33
34        super().__init__(*args, **kwargs)
35
36        # override the provider name from "s3"
37        self._provider_name = PROVIDER
38
39    def _delete_objects(self, paths: list[str]) -> None:
40        """
41        GCS S3 does not support bulk deletion, so we delete one object at a time.
42        """
43        for path in paths:
44            self._delete_object(path)