ai-training/cdn_manager.py

import io
import boto3


class CDNCredentials:
    def __init__(self, host, access_key, secret_key):
        self.host = host
        self.access_key = access_key
        self.secret_key = secret_key


class CDNManager:
    def __init__(self, credentials: CDNCredentials):
        self.creds = credentials
        self.minio_client = boto3.client('s3', endpoint_url=self.creds.host,
                                         aws_access_key_id=self.creds.access_key,
                                         aws_secret_access_key=self.creds.secret_key)

    def upload(self, bucket: str, filename: str, file_bytes: bytearray):
        try:
            self.minio_client.upload_fileobj(io.BytesIO(file_bytes), bucket, filename)
            print(f'Uploaded {len(file_bytes)} bytes to {self.creds.host}/{bucket}/{filename}')
            return True
        except Exception as e:
            print(e)
            return False

    def download(self, bucket: str, filename: str):
        try:
            self.minio_client.download_file(bucket, filename, filename)
            return True
        except Exception as e:
            print(e)
            return False