66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
from webdav3.exceptions import WebDavException, ResponseErrorCode
|
|
from webdav3.client import Client
|
|
import storage.exceptions as StorageException
|
|
from storage.utils import human_filesize
|
|
from storage.storage import BaseStorage
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# WebDAV Support - https://pypi.org/project/webdavclient3/
|
|
|
|
|
|
class WebDAVStorage(BaseStorage):
|
|
|
|
TYPE = 'webdav'
|
|
|
|
def __connect(self):
|
|
# Connect to the external storage. This function can be run multiple times. It will check if it has already a connection to re-use
|
|
try:
|
|
# When this fails with an Attribute error, that means that the 'client' variable is not set and we need to make a new connection
|
|
assert(self.client)
|
|
except AttributeError:
|
|
# Because the 'client' variable is not known, the WebDAV connections is not created yet. So do it now!
|
|
self.client = Client({
|
|
'webdav_hostname': self.url,
|
|
'webdav_login': self.username,
|
|
'webdav_password': self.password,
|
|
})
|
|
|
|
try:
|
|
# Here we abuse the .free check to see if the login credentials do work
|
|
free_space = self.client.free()
|
|
logger.info(f'Created WebDAV connection to url: \'{self.url}\', with space left: {human_filesize(free_space)}')
|
|
except ResponseErrorCode as ex:
|
|
# Login went wrong, so delete the client variable for next run/try
|
|
del(self.client)
|
|
|
|
# If there was an authentication error, raise exception and quit.
|
|
if 401 == ex.code:
|
|
raise StorageException.InvalidAuthentication(self.username)
|
|
|
|
# TODO: More errors.....
|
|
|
|
def _file_exists_action(self, path):
|
|
self.__connect()
|
|
return self.client.check(path)
|
|
|
|
def _directory_exists_action(self, path):
|
|
self.__connect()
|
|
return self.client.check(path)
|
|
|
|
def _make_folder_action(self, path):
|
|
self.__connect()
|
|
self.client.mkdir(path)
|
|
return True
|
|
|
|
def _upload_file_action(self, source, destination):
|
|
self.__connect()
|
|
self.client.upload(local_path=source, remote_path=destination)
|
|
return True
|
|
|
|
def _download_file_action(self, source, destination):
|
|
self.__connect()
|
|
self.client.download(source, destination)
|
|
return True
|