Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def load_maestral(self):
pending_link = not _is_linked()
pending_dbx_folder = not os.path.isdir(CONF.get("main", "path"))
if pending_link or pending_dbx_folder:
from maestral.gui.setup_dialog import SetupDialog
logger.info("Setting up Maestral...")
done = SetupDialog.configureMaestral(pending_link)
if done:
logger.info("Successfully set up Maestral")
self.restart()
else:
logger.info("Setup aborted.")
self.quit()
else:
self.mdbx = self._get_or_start_maestral_daemon()
self.setup_ui_linked()
has_changes = sync.wait_for_remote_changes(sync.last_cursor, timeout=30)
if not running.is_set(): # if stopped, return
return
syncing.wait() # if paused, wait until resumed
# apply remote changes
if has_changes:
logger.info(SYNCING)
with sync.lock:
# get changes
changes = sync.list_remote_changes(sync.last_cursor)
# notify user about changes
if CONF.get("app", "notifications"):
sync.notify_user(changes)
# apply remote changes to local Dropbox folder
sync.apply_remote_changes(changes)
logger.info(IDLE)
except ConnectionError:
syncing.clear()
connected.clear()
disconnected_signal.send()
logger.debug(DISCONNECTED, exc_info=True)
logger.info(DISCONNECTED)
except MaestralApiError as e:
syncing.clear() # stop syncing
running.clear() # shutdown threads
def _save_to_history(dbx_path):
# add new file to recent_changes
recent_changes = CONF.get("internal", "recent_changes")
recent_changes.append(dbx_path)
# eliminate duplicates
recent_changes = list(OrderedDict.fromkeys(recent_changes))
# save last 30 changes
CONF.set("internal", "recent_changes", recent_changes[-30:])
def __init__(self, client, local_file_event_queue, queue_uploading, queue_downloading):
self.client = client
self.local_file_event_queue = local_file_event_queue
self.queue_uploading = queue_uploading
self.queue_downloading = queue_downloading
# load cached properties
self._dropbox_path = CONF.get("main", "path")
self._excluded_files = CONF.get("main", "excluded_files")
self._excluded_folders = CONF.get("main", "excluded_folders")
self._rev_dict_cache = self._load_rev_dict_from_file()
def _ask_for_path(default=osp.join("~", CONF.get("main", "default_dir_name"))):
"""
Asks for Dropbox path.
"""
while True:
msg = f"Please give Dropbox folder location or press enter for default ['{default}']:"
res = input(msg).strip("'\" ")
dropbox_path = osp.expanduser(res or default)
old_path = CONF.get("main", "path")
try:
if osp.samefile(old_path, dropbox_path):
return
except FileNotFoundError:
pass
def __repr__(self):
email = CONF.get("account", "email")
account_type = CONF.get("account", "type")
return f"<{self.__class__}({email}, {account_type})>"
def __init__(self):
self.account_id = CONF.get("account", "account_id")
self.access_token = ""