class_docstr: An optimized version of the S3 Artifact Repository.
  
  This class is used for uploading and downloading S3 artifacts for UC models. While it largely
  copies the behavior of the S3ArtifactRepository, the `log_artifact`, `log_artifacts`, and
  `_download_file` methods are optimized by replacing boto3 client operations with the use of
  presigned URLs for both uploads and downloads.
class_docstr_tok: ["An","optimized","version","of","the","S3","Artifact","Repository",".","This","class","is","used","for","uploading","and","downloading","S3","artifacts","for","UC","models",".","While","it","largely","copies","the","behavior","of","the","S3ArtifactRepository",",","the","`","log_artifact","`",",","`","log_artifacts","`",",","and","`","_download_file","`","methods","are","optimized","by","replacing","boto3","client","operations","with","the","use","of","presigned","URLs","for","both","uploads","and","downloads","."]
code: def _multipart_upload(
      self, cloud_credential_info, local_file, bucket, key
  ):
      # Create multipart upload
      s3_client = cloud_credential_info
      response = s3_client.create_multipart_upload(
          Bucket=bucket, Key=key
      )
      upload_id = response["UploadId"]
  
      # Create presigned URL for each part
      num_parts = math.ceil(
          os.path.getsize(local_file)
          / MLFLOW_MULTIPART_UPLOAD_CHUNK_SIZE.get()
      )
  
      # define helper functions for uploading data
      def _upload_part(part_number, local_file, start_byte, size):
          presigned_url = s3_client.generate_presigned_url(
              "upload_part",
              Params={
                  "Bucket": bucket,
                  "Key": key,
                  "UploadId": upload_id,
                  "PartNumber": part_number,
              },
          )
          data = read_chunk(local_file, size, start_byte)
          with cloud_storage_http_request(
              "put", presigned_url, data=data
          ) as response:
              augmented_raise_for_status(response)
              return response.headers["ETag"]
  
      try:
          # Upload each part with retries
          futures = {}
          for index in range(num_parts):
              part_number = index + 1
              start_byte = (
                  index * MLFLOW_MULTIPART_UPLOAD_CHUNK_SIZE.get()
              )
              future = self.chunk_thread_pool.submit(
                  _upload_part,
                  part_number=part_number,
                  local_file=local_file,
                  start_byte=start_byte,
                  size=MLFLOW_MULTIPART_UPLOAD_CHUNK_SIZE.get(),
              )
              futures[future] = part_number
  
          results, errors = _complete_futures(futures, local_file)
          if errors:
              raise MlflowException(
                  f"Failed to upload at least one part of {local_file}. Errors: {errors}"
              )
          parts = [
              {"PartNumber": part_number, "ETag": results[part_number]}
              for part_number in sorted(results)
          ]
  
          # Complete multipart upload
          s3_client.complete_multipart_upload(
              Bucket=bucket,
              Key=key,
              UploadId=upload_id,
              MultipartUpload={"Parts": parts},
          )
      except Exception as e:
          _logger.warning(
              "Encountered an unexpected error during multipart upload: %s, aborting",
              e,
          )
          s3_client.abort_multipart_upload(
              Bucket=bucket,
              Key=key,
              UploadId=upload_id,
          )
          raise e
  
imports: import json
  import logging
  import math
  import os
  import posixpath
  import urllib.parse
  from mimetypes import guess_type
  from mlflow.entities import FileInfo
  from mlflow.environment_variables import MLFLOW_ENABLE_MULTIPART_UPLOAD, MLFLOW_MULTIPART_UPLOAD_CHUNK_SIZE, MLFLOW_S3_UPLOAD_EXTRA_ARGS
  from mlflow.exceptions import MlflowException
  from mlflow.protos.databricks_artifacts_pb2 import ArtifactCredentialInfo
  from mlflow.store.artifact.cloud_artifact_repo import CloudArtifactRepository, _complete_futures
  from mlflow.store.artifact.s3_artifact_repo import _get_s3_client
  from mlflow.utils.file_utils import read_chunk
  from mlflow.utils.request_utils import cloud_storage_http_request
  from mlflow.utils.rest_utils import augmented_raise_for_status
label_desc: Use image node_id 10 for calling the OptimizedS3ArtifactRepository obj's underlying member method code with example usage: obj._multipart_upload(cloud_credential_info, local_file, bucket, key) and returns: response