138 lines
4.7 KiB
Python
138 lines
4.7 KiB
Python
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 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
|