code: def __init__(
      self, config, measure_timestamp_diff=False, compression=None
  ):
      self.logger = logs.get_logger(self.__class__.__name__)
      self.logger.setLevel(get_storage_verbose_level())
      self.credentials: Credentials = Credentials.get_credentials(
          config
      )
  
      self.endpoint = config.get("endpoint", None)
  
      if self.credentials is None:
          msg: str = "NO CREDENTIALS provided for {0}.".format(
              self.endpoint
          )
          self._report_fatal(msg)
  
      if self.credentials.get_type() != AWS_TYPE:
          msg: str = "EXPECTED aws credentials for {0}: {1}".format(
              self.endpoint, repr(self.credentials.to_dict())
          )
          self._report_fatal(msg)
  
      aws_key: str = self.credentials.get_key()
      aws_secret_key = self.credentials.get_secret_key()
      region_name = self.credentials.get_region()
      profile_name = self.credentials.get_profile()
  
      if profile_name is not None:
          # it seems that explicitly specified profile name
          # should not be used with explicitly specified credentials:
          aws_key = None
          aws_secret_key = None
  
      session = Session(
          aws_access_key_id=aws_key,
          aws_secret_access_key=aws_secret_key,
          region_name=region_name,
          profile_name=profile_name,
      )
  
      session.events.unregister(
          "before-parameter-build.s3.ListObjects",
          set_list_objects_encoding_type_url,
      )
  
      self.client = session.client(
          "s3",
          endpoint_url=self.endpoint,
          config=Config(signature_version="s3v4"),
      )
  
      if compression is None:
          compression = config.get("compression", None)
  
      self.cleanup_bucket = config.get("cleanup_bucket", False)
      if isinstance(self.cleanup_bucket, str):
          self.cleanup_bucket = self.cleanup_bucket.lower() == "true"
      self.bucket_cleaned_up: bool = False
  
      self.endpoint = self.client._endpoint.host
  
      self.bucket = config["bucket"]
      try:
          buckets = self.client.list_buckets()
      except Exception as exc:
          msg: str = "FAILED to list buckets for {0}: {1}".format(
              self.endpoint, exc
          )
          self._report_fatal(msg)
  
      if self.bucket not in [b["Name"] for b in buckets["Buckets"]]:
          try:
              if region_name is not None:
                  self.client.create_bucket(
                      Bucket=self.bucket,
                      CreateBucketConfiguration={
                          "LocationConstraint": region_name
                      },
                  )
              else:
                  self.client.create_bucket(Bucket=self.bucket)
  
          except Exception as exc:
              msg: str = (
                  "FAILED to create bucket {0} for {1}: {2}".format(
                      self.bucket, self.endpoint, exc
                  )
              )
              self._report_fatal(msg)
  
      super().__init__(
          StorageType.storageS3,
          self.logger,
          measure_timestamp_diff,
          compression=compression,
      )
  
imports: import calendar
  import os
  from urllib.parse import urlparse
  from boto3.s3.transfer import TransferConfig
  from boto3.session import Session
  import botocore
  from botocore.client import Config
  from botocore.handlers import set_list_objects_encoding_type_url
  from typing import Dict
  import re
  from studio.util import logs
  from studio.util import util
  from studio.credentials.credentials import Credentials, AWS_TYPE
  from studio.storage.storage_handler import StorageHandler
  from studio.storage.storage_type import StorageType
  from studio.storage.storage_setup import get_storage_verbose_level
label_desc: Use image node_id 1 to create a new S3StorageHandler object from inherited base classes: StorageHandler with example: obj = S3StorageHandler(config, measure_timestamp_diff, compression)