federated-storage/middleware/app_be/services/minioservice.py

import base64
import hashlib
from typing import Union

import xml.etree.ElementTree as ET
import requests
from django.conf import settings

from app_be.services.StorageServiceInterface import StorageServiceInterface


class MinioService(StorageServiceInterface):
    name = "MinIO"

    @staticmethod
    def check() -> bool:
        print("Checking MinIO availability")
        try:
            url = settings.AWS_HOST
            r = requests.put(url)
            if not (200 <= r.status_code < 500):
                return False
        except:
            return False
        return True

    @staticmethod
    def create_file(filename: str, file: bytes) -> bool:
        """Create / save (or overwrite) an file on the minio storage.

        :param filename: filename
        :param file: bytes representation of the file
        :return: Whether file was successfully uploaded
        """
        try:
            url = settings.AWS_HOST + filename
            headers = {'Content-Type': 'binary/octet-stream'}
            r = requests.put(url, data=file, headers=headers)
            if r.status_code == 200:
                print("Successfully uploaded a file!")
            else:
                print("Something went wrong while creating")
        except:
            return False
        return True

    @staticmethod
    def update_file(filename: str, file: bytes) -> bool:
        """Update a file on the minio storage.

        Finds the latest version of the current file and adds another
        version to the storage.
        :param filename: The filename without the version.
        :param file: Byte representation of the file to save as new version
        :return True if successful, False otherwise
        """
        identifier = filename.strip('.jpg')
        print("MinioService: Trying to update image with identifier ", identifier)
        try:
            headers = {'Content-Type': 'binary/octet-stream'}
            i = 0
            while MinioService.read_file(identifier + '_' + str(i) + '.jpg') is not None:
                print("MinioService: Version {} exists".format(identifier + '_' + str(i) + '.jpg'))
                i = i + 1
            print("MinioService: Creating new version {}".format(identifier + '_' + str(i) + '.jpg'))
            url = settings.AWS_HOST + identifier + '_' + str(i) + '.jpg'
            r = requests.put(url, data=file, headers=headers)
            if r.status_code == 200:
                print("Successfully uploaded a new version with filename {}!".format(identifier + '_' + str(i) + '.jpg'))
            else:
                print("Something went wrong while updating")
        except:
            return False
        return True

    @staticmethod
    def move_file(identifier_from: str, identifier_to: str) -> bool:
        """Move/Rename a file on the S3 storage.

        Consists of a PUT method (copying the object) and a delete method.
        :param identifier_from: Location of the file to be moved
        :param identifier_to: Location the file should be moved to
        :return True on success, False otherwise
        """
        print("MinioService: Trying to move/rename file from {} to {}".format(identifier_from, identifier_to))

        filename_from = identifier_from + '.jpg'
        filename_to = identifier_to + '.jpg'

        url = settings.AWS_HOST + filename_to
        headers = {'x-amz-copy-source': filename_from}
        r = requests.put(url, headers=headers)
        if r.status_code == 200:
            print("Successfully moved/renamed the file {}!".format(filename_from))
        else:
            print("Something went wrong while moving/renaming the file {}".format(filename_from))

    @staticmethod
    def read_file(filename: str) -> bytes:
        """Read  file on the minio storage.

        :param filename: filename
        :return: bytes representation of the file or None on error
        """
        file_bytes: Union[bytes, None] = None
        try:
            url = settings.AWS_HOST + filename
            response = requests.get(url, stream=True)
            if response.status_code == 404:
                return file_bytes
            file_bytes = response.content
        except:
            print("Error downloading file from minio")
        return file_bytes

    @staticmethod
    def delete_file(filename: str) -> bool:
        """Delete an file from the dropbox storage.

        :param filename: filename of the file
        :return: Whether file was successfully deleted
        """
        try:
            url = settings.AWS_HOST + filename
            r = requests.delete(url)
            if r.status_code == 204:
                print("Successfully deleted a file!")
            else:
                print("Something went wrong while deleting")
        except:
            return False
        return True

    @staticmethod
    def delete_all() -> bool:
        """Delete every file from the dropbox storage.

        :return: Whether the storage was successfully emptied
        """
        try:
            r = requests.get(settings.AWS_HOST + "?list")
            root = ET.fromstring(r.content)
            list = []
            cnt = 0
            for child in root.iter('{http://s3.amazonaws.com/doc/2006-03-01/}Contents'):
                list.append(child.find('{http://s3.amazonaws.com/doc/2006-03-01/}Key').text)
                cnt = cnt + 1
            if cnt == 0:
                return True
            body = "<Delete>"
            for key in list:
                body += "<Object><Key>" + key + "</Key></Object>"

            body += "</Delete>"
            print(body)
            header = {"Content-MD5": base64.b64encode(hashlib.md5(body.encode()).digest()),
                      'Content-Type': 'binary/application/xml'}
            r = requests.post(settings.AWS_HOST + "?delete", data=body, headers=header)
            if r.status_code == 200:
                print("Successfully deleted all files!")
            else:
                print("Something went wrong by deleting", r.status_code, r.content)
                exit()
        except:
            return False
        return True