Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def move(self, dbx_path, new_path, **kwargs):
"""
Moves/renames files or folders on Dropbox.
:param str dbx_path: Path to file/folder on Dropbox.
:param str new_path: New path on Dropbox to move to.
:param kwargs: Keyword arguments for Dropbox SDK files_move_v2.
:returns: Metadata of moved file/folder.
:raises: :class:`MaestralApiError`
"""
try:
res = self.dbx.files_move_v2(dbx_path, new_path, allow_shared_folder=True,
allow_ownership_transfer=True, **kwargs)
md = res.metadata
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc, new_path)
logger.debug("File moved from '{0}' to '{1}' on Dropbox.".format(
dbx_path, md.path_display))
return md
"""
if not 30 <= timeout <= 480:
raise ValueError("Timeout must be in range [30, 480]")
logger.debug("Waiting for remote changes since cursor:\n{0}".format(last_cursor))
# honour last request to back off
if self._last_longpoll is not None:
while time.time() - self._last_longpoll < self._backoff:
time.sleep(1)
try:
result = self.dbx.files_list_folder_longpoll(last_cursor, timeout=timeout)
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc)
# keep track of last long poll, back off if requested by SDK
if result.backoff:
self._backoff = result.backoff + 5
else:
self._backoff = 0
logger.debug("Detected remote changes: {}.".format(str(result.changes)))
self._last_longpoll = time.time()
return result.changes # will be True or False
"""
# create local directory if not present
dst_path_directory = osp.dirname(dst_path)
if not osp.exists(dst_path_directory):
try:
os.makedirs(dst_path_directory)
except FileExistsError:
pass
except OS_FILE_ERRORS as exc:
raise construct_local_error(exc, dbx_path)
try:
md = self.dbx.files_download_to_file(dst_path, dbx_path, **kwargs)
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc, dbx_path)
except OS_FILE_ERRORS as exc:
raise construct_local_error(exc, dbx_path)
logger.debug("File '{0}' (rev {1}) from '{2}' was successfully downloaded "
"as '{3}'.".format(md.name, md.rev, md.path_display, dst_path))
return md
Lists changes to remote Dropbox since :param:`last_cursor`. Call this
after :method:`wait_for_remote_changes` returns `True`. Only remote changes
in currently synced folders will be returned by default.
:param str last_cursor: Last to cursor to compare for changes.
:returns: :class:`dropbox.files.ListFolderResult` instance.
:rtype: :class:`dropbox.files.ListFolderResult`
:raises:
"""
results = []
try:
results.append(self.dbx.files_list_folder_continue(last_cursor))
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc)
while results[-1].has_more:
try:
more_results = self.dbx.files_list_folder_continue(results[-1].cursor)
results.append(more_results)
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc)
# combine all results into one
results = self.flatten_results(results)
logger.debug("Listed remote changes: {} changes.".format(len(results.entries)))
return results
def get_account_info(self):
"""
Gets current account information.
:returns: :class:`dropbox.users.FullAccount` instance or `None` if failed.
:rtype: dropbox.users.FullAccount
"""
try:
res = self.dbx.users_get_current_account() # should only raise auth errors
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc)
if res.account_type.is_basic():
account_type = 'basic'
elif res.account_type.is_business():
account_type = 'business'
elif res.account_type.is_pro():
account_type = 'pro'
else:
account_type = ''
CONF.set("account", "account_id", res.account_id)
CONF.set("account", "email", res.email)
CONF.set("account", "display_name", res.name.display_name)
CONF.set("account", "abbreviated_name", res.name.abbreviated_name)
CONF.set("account", "type", account_type)
def get_space_usage(self):
"""
Gets current account space usage.
:returns: :class:`SpaceUsage` instance or `False` if failed.
:rtype: SpaceUsage
"""
try:
res = self.dbx.users_get_space_usage() # should only raise auth errors
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc)
# convert from dropbox.users.SpaceUsage to SpaceUsage with nice string
# representation
res.__class__ = SpaceUsage
if res.allocation.is_team():
CONF.set("account", "usage_type", "team")
elif res.allocation.is_individual():
CONF.set("account", "usage_type", "individual")
CONF.set("account", "usage", repr(res))
return res
**kwargs
)
results.append(res)
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc, dbx_path)
idx = 0
while results[-1].has_more:
idx += len(results[-1].entries)
logger.info("Indexing {0}...".format(idx))
try:
more_results = self.dbx.files_list_folder_continue(results[-1].cursor)
results.append(more_results)
except dropbox.exceptions.DropboxException as exc:
new_exc = to_maestral_error(exc, dbx_path)
if isinstance(new_exc, CursorResetError) and self._retry_count < retry:
# retry up to three times, then raise
self._retry_count += 1
self.list_folder(dbx_path, include_non_downloadable_files, **kwargs)
else:
self._retry_count = 0
raise new_exc
logger.debug("Listed contents of folder '{0}'".format(dbx_path))
self._retry_count = 0
return self.flatten_results(results)
def unlink(self):
"""
Unlinks the Dropbox account and deletes local sync information.
"""
self.auth.delete_creds()
try:
self.dbx.auth_token_revoke() # should only raise auth errors
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc)
:raises: :class:`MaestralApiError` if deletion fails for any other reason than
a non-existing file.
"""
try:
# try to move file (response will be metadata, probably)
res = self.dbx.files_delete_v2(dbx_path, **kwargs)
md = res.metadata
except dropbox.exceptions.DropboxException as exc:
if (isinstance(exc.error, dropbox.files.DeleteError) and
exc.error.is_path_lookup()):
# don't log as error if file did not exist
logger.debug("An error occurred when deleting '{0}': the file does "
"not exist on Dropbox".format(dbx_path))
return True
else:
raise to_maestral_error(exc, dbx_path)
logger.debug("File / folder '{0}' removed from Dropbox.".format(dbx_path))
return md
def make_dir(self, dbx_path, **kwargs):
"""
Creates folder on Dropbox.
:param str dbx_path: Path o fDropbox folder.
:param kwargs: Keyword arguments for Dropbox SDK files_create_folder_v2.
:returns: Metadata of created folder.
:raises: :class:`MaestralApiError`
"""
try:
res = self.dbx.files_create_folder_v2(dbx_path, **kwargs)
md = res.metadata
except dropbox.exceptions.DropboxException as exc:
raise to_maestral_error(exc, dbx_path)
logger.debug("Created folder '%s' on Dropbox.", md.path_display)
return md