upload and download stream data from S3 and S3 Compatible with Python

s3.py

import os
import shutil
from minio import Minio
from minio.datatypes import Object
from minio.helpers import ObjectWriteResult
from zipfile import ZipFile

S3_ENDPOINT = 's3.ap-southeast-3.amazonaws.com'
S3_CLIENT_ID = 'ABC'
S3_CLIENT_SECRET = 'BBB'
S3_BUCKET_NAME = 'abc-myproduct-priv-dev'

def create_zip(name, source):
  archived = shutil.make_archive(name, 'zip', source)
  if os.path.exists(name):
    print(archived)
  else:
    print('create zip failed')

def create_zip_file(name, source):
  with ZipFile(name, 'w') as newzipfile:
     newzipfile.write(source)

def unzip(filename, target_folder):
  if not os.path.exists(target_folder):
    os.mkdir(target_folder)

  shutil.unpack_archive(filename, target_folder)

def upload_model(bucket_name: str, object_name: str, file_path: str) -> ObjectWriteResult:
    if not os.path.exists(file_path):
      print(f'upload failed: {file_path} is not exist')
      return

    # Create client with access and secret key
    client = Minio(
        S3_ENDPOINT,
        S3_CLIENT_ID,
        S3_CLIENT_SECRET, secure=True)

    # Make sure bucket exists.
    found = client.bucket_exists(bucket_name)
    if not found:
        print('bucket not found')
        return

    # Upload the file.
    object_write_result = client.fput_object(bucket_name, object_name, file_path)

    return object_write_result

def download_model(bucket_name: str, object_name: str, file_path: str) -> Object:
    # Create client with access and secret key
    client = Minio(
        S3_ENDPOINT,
        S3_CLIENT_ID,
        S3_CLIENT_SECRET, secure=True)

    # Get data of an object.
    object_info = client.fget_object(bucket_name, object_name, file_path)

    return object_info


# Execute download
download_model(S3_BUCKET_NAME, 'data.pkl', 'folder-myproduct')

todo.md

Install Minio Client

pip3 install minio