122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
import requests
|
|
from django.conf import settings
|
|
import dropbox
|
|
from dropbox.exceptions import ApiError
|
|
|
|
from app_be.services.StorageServiceInterface import StorageServiceInterface
|
|
|
|
|
|
class DropboxService(StorageServiceInterface):
|
|
name = "Dropbox"
|
|
|
|
@staticmethod
|
|
def get_last_modified(filename: str):
|
|
"""Get last modified time of file
|
|
|
|
:param filename: filename
|
|
:return: Date of last modification or None
|
|
"""
|
|
try:
|
|
dbx = dropbox.Dropbox(settings.DROPBOX_OAUTH2_ACCESS_TOKEN)
|
|
return dbx.files_get_metadata(settings.DROPBOX_IMAGE_FOLDER + filename).server_modified
|
|
except:
|
|
return None
|
|
return None
|
|
|
|
@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 move_file(identifier_from: str, identifier_to: str, file: bytes = None) -> bool:
|
|
"""Move/Rename a file on the dropbox storage.
|
|
|
|
: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("DropboxService: Trying to move/rename file with identifier {} to identifier {}".format(identifier_from,
|
|
identifier_to))
|
|
|
|
dbx = dropbox.Dropbox(settings.DROPBOX_OAUTH2_ACCESS_TOKEN)
|
|
url_from = settings.DROPBOX_IMAGE_FOLDER + identifier_from + '.jpg'
|
|
url_to = settings.DROPBOX_IMAGE_FOLDER + identifier_to
|
|
|
|
try:
|
|
dbx.files_move_v2(url_from, url_to)
|
|
except ApiError:
|
|
return False
|
|
|
|
return True
|
|
|
|
@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
|