111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
import requests
|
|
from django.conf import settings
|
|
import dropbox
|
|
|
|
from app_be.services.StorageServiceInterface import StorageServiceInterface
|
|
|
|
|
|
class DropboxService(StorageServiceInterface):
|
|
name = "Dropbox"
|
|
|
|
@staticmethod
|
|
def check() -> bool:
|
|
try:
|
|
dbx = dropbox.Dropbox(settings.DROPBOX_OAUTH2_ACCESS_TOKEN)
|
|
result = dbx.check_user(query=u'Hello')
|
|
if not result.result == 'Hello':
|
|
print("String does not match")
|
|
return False
|
|
except:
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def create_file(filename: str, file: bytes) -> bool:
|
|
"""Create / save (or overwrite) an file on the dropbox storage.
|
|
|
|
:param filename: filename
|
|
:param file: bytes representation of the file
|
|
:return: Whether file was successfully uploaded
|
|
"""
|
|
try:
|
|
dbx = dropbox.Dropbox(settings.DROPBOX_OAUTH2_ACCESS_TOKEN)
|
|
dbx.files_upload(file, settings.DROPBOX_IMAGE_FOLDER + filename)
|
|
except:
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def read_file(filename: str) -> bytes:
|
|
"""Read file on the dropbox storage.
|
|
|
|
:param filename: filename
|
|
:return: bytes representation of the file or None on error
|
|
"""
|
|
file_bytes: bytes
|
|
try:
|
|
dbx = dropbox.Dropbox(settings.DROPBOX_OAUTH2_ACCESS_TOKEN)
|
|
f: requests.models.Response
|
|
metadata, f = dbx.files_download(settings.DROPBOX_IMAGE_FOLDER + filename)
|
|
file_bytes = f.content
|
|
if f is not None:
|
|
f.close()
|
|
except:
|
|
print("Error downloading file from dropbox")
|
|
return file_bytes
|
|
|
|
@staticmethod
|
|
def update_file(filename: str, file: bytes) -> bool:
|
|
"""Update a file on the dropbox storage.
|
|
|
|
Increments a counter until we get to the newest version, then
|
|
create the file with the newest version + 1.
|
|
:param filename: filename of the file
|
|
:param file: bytes representation of the file
|
|
:return: Whether file was successfully uploaded
|
|
"""
|
|
identifier = filename.strip('.jpg')
|
|
print("DropboxService: Trying to update file with identifier ", identifier)
|
|
i = 0
|
|
while True:
|
|
try:
|
|
dbx = dropbox.Dropbox(settings.DROPBOX_OAUTH2_ACCESS_TOKEN)
|
|
print("DropboxService: Checking if version {} exists".format(identifier + '_' + str(i) + '.jpg'))
|
|
dbx.files_get_metadata(identifier + '_' + str(i) + '.jpg')
|
|
i = i + 1
|
|
except:
|
|
print("DropboxService: Version {} exists".format(identifier + '_' + str(i) + '.jpg'))
|
|
break
|
|
|
|
print("DropboxService: Creating new version {}".format(identifier + '_' + str(i) + '.jpg'))
|
|
return DropboxService.create_file(identifier + '_' + str(i) + '.jpg', file)
|
|
|
|
@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:
|
|
dbx = dropbox.Dropbox(settings.DROPBOX_OAUTH2_ACCESS_TOKEN)
|
|
dbx.files_delete_v2(settings.DROPBOX_IMAGE_FOLDER + filename)
|
|
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:
|
|
dbx = dropbox.Dropbox(settings.DROPBOX_OAUTH2_ACCESS_TOKEN)
|
|
dbx.files_delete_v2(settings.DROPBOX_IMAGE_FOLDER[:-1])
|
|
dbx.files_create_folder_v2(settings.DROPBOX_IMAGE_FOLDER[:-1])
|
|
except dropbox.exceptions.ApiError as e:
|
|
print(e)
|
|
return False
|
|
return True
|