feat(): add customize boto core config for OSS storage provider

Author: JunchengXue

flow360/cloud/s3_utils.py

@@ -8,10 +8,11 @@
 from abc import ABCMeta, abstractmethod
 from datetime import datetime
 from enum import Enum
+from typing import Optional
 
 import boto3
 from boto3.s3.transfer import TransferConfig
-
+from botocore.config import Config
 # pylint: disable=unused-import
 from botocore.exceptions import ClientError as CloudFileNotFoundError
 from pydantic.v1 import BaseModel, Field
@@ -110,12 +111,14 @@ class _UserCredential(BaseModel):
     expiration: datetime
     secret_access_key: str = Field(alias="secretAccessKey")
     session_token: str = Field(alias="sessionToken")
+    storage_provider: Optional[str] = Field(alias="storageProvider", default=None)
 
 
 class _S3STSToken(BaseModel):
     cloud_path_prefix: str = Field(alias="cloudpathPrefix")
     cloud_path: str = Field(alias="cloudpath")
     user_credential: _UserCredential = Field(alias="userCredentials")
+    endpoint: Optional[str] = Field(alias="endpoint", default=None)
 
     def get_bucket(self):
         """
@@ -138,14 +141,28 @@ def get_client(self):
         Get s3 client.
         :return:
         """
+        customize_boto3_config = None
+        if (self.user_credential is not None
+                and self.user_credential.storage_provider is not None
+                and self.user_credential.storage_provider == "OSS"):
+            # OSS does not support aws integrity check
+            customize_boto3_config = Config(request_checksum_calculation="when_required",
+                                            response_checksum_validation="when_required",
+                                            s3={
+                                                "addressing_style": "virtual"
+                                            })
         # pylint: disable=no-member
         kwargs = {
             "region_name": Env.current.aws_region,
             "aws_access_key_id": self.user_credential.access_key_id,
             "aws_secret_access_key": self.user_credential.secret_access_key,
             "aws_session_token": self.user_credential.session_token,
+            "config": customize_boto3_config
         }
 
+        if self.endpoint is not None:
+            kwargs["endpoint_url"] = self.endpoint
+
         if Env.current.s3_endpoint_url is not None:
             kwargs["endpoint_url"] = Env.current.s3_endpoint_url
 
@@ -158,8 +175,8 @@ def is_expired(self):
         """
         # pylint: disable=no-member
         return (
-            self.user_credential.expiration
-            - datetime.now(tz=self.user_credential.expiration.tzinfo)
+                self.user_credential.expiration
+                - datetime.now(tz=self.user_credential.expiration.tzinfo)
         ).total_seconds() < 300
 
 
@@ -203,9 +220,9 @@ def get_cloud_path_prefix(self, resource_id, file_name):
         return base_path
 
     def create_multipart_upload(
-        self,
-        resource_id: str,
-        remote_file_name: str,
+            self,
+            resource_id: str,
+            remote_file_name: str,
     ):
         """
         Creates a multipart upload for the specified resource ID and remote file name.
@@ -226,12 +243,12 @@ def create_multipart_upload(
 
     # pylint: disable=too-many-arguments
     def upload_part(
-        self,
-        resource_id: str,
-        remote_file_name: str,
-        upload_id: str,
-        part_number: int,
-        compressed_chunk,
+            self,
+            resource_id: str,
+            remote_file_name: str,
+            upload_id: str,
+            part_number: int,
+            compressed_chunk,
     ):
         """
         Uploads a part of the file as part of a multipart upload.
@@ -260,7 +277,7 @@ def upload_part(
         return {"ETag": response["ETag"], "PartNumber": part_number}
 
     def complete_multipart_upload(
-        self, resource_id: str, remote_file_name: str, upload_id: str, uploaded_parts: dict
+            self, resource_id: str, remote_file_name: str, upload_id: str, uploaded_parts: dict
     ):
         """
         Completes a multipart upload for the specified resource ID, remote file name, upload ID, e_tag, and part number.
@@ -285,7 +302,7 @@ def complete_multipart_upload(
         )
 
     def upload_file(
-        self, resource_id: str, remote_file_name: str, file_name: str, progress_callback=None
+            self, resource_id: str, remote_file_name: str, file_name: str, progress_callback=None
     ):
         """
         Upload a file to s3.
@@ -328,15 +345,15 @@ def _call_back(bytes_in_chunk):
 
     # pylint: disable=too-many-arguments
     def download_file(
-        self,
-        resource_id: str,
-        remote_file_name: str,
-        to_file: str = None,
-        to_folder: str = ".",
-        overwrite: bool = True,
-        progress_callback=None,
-        log_error=True,
-        verbose=True,
+            self,
+            resource_id: str,
+            remote_file_name: str,
+            to_file: str = None,
+            to_folder: str = ".",
+            overwrite: bool = True,
+            progress_callback=None,
+            log_error=True,
+            verbose=True,
     ):
         """
         Download a file from s3.