41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
import requests
|
|
from django.conf import settings
|
|
|
|
class MinioService:
|
|
|
|
@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")
|
|
except:
|
|
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: bytes
|
|
try:
|
|
url = settings.AWS_HOST + filename
|
|
response = requests.get(url, stream=True)
|
|
file_bytes=response.content
|
|
except:
|
|
print("Error downloading file from minio")
|
|
return file_bytes |