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

import base64
import hashlib
from datetime import datetime
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 get_last_modified(filename: str):
        """Get last modified time of file

        :param filename: filename
        :return: Date of last modification or None
        """
        try:
            r = requests.get(settings.AWS_HOST + "?prefix="+filename)
            root = ET.fromstring(r.content)
            for child in root.iter('{http://s3.amazonaws.com/doc/2006-03-01/}Contents'):
                if child.find('{http://s3.amazonaws.com/doc/2006-03-01/}Key').text == filename:
                    date_time_str = child.find('{http://s3.amazonaws.com/doc/2006-03-01/}LastModified').text
                    print(date_time_str)
                    date_time_obj = datetime.strptime(date_time_str, '%Y-%m-%dT%H:%M:%S.%fZ')
                    return date_time_obj
        except:
            return None
        return None

    @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 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

        file = MinioService.read_file(filename_from)

        if not MinioService.delete_file(filename_from):
            print("MinioService: Updating (delete) file {} failed".format(filename_from))
            return False

        if not MinioService.create_file(filename_to, file):
            print("MinioService: Updating (create) file {} failed".format(filename_to))
            return False

        return True

    @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