Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _read_pid(config_name):
"""
Reads and returns the PID of the current maestral daemon process from the appropriate
file for the given config name.
"""
from maestral.sync.utils.appdirs import get_runtime_path
pid_file = get_runtime_path("maestral", config_name + ".pid")
with open(pid_file, "r") as f:
pid = f.read().split("\n")[0] # ignore all new lines
pid = int(pid)
logger.debug(f"PID {pid} read from '{pid_file}'.")
return pid
def _get_sock_name(config_name):
"""
Returns the unix socket location to be used for the config. This should default to
the apps runtime directory + '/maestral/CONFIG_NAME.sock'.
"""
os.environ["MAESTRAL_CONFIG"] = config_name
from maestral.sync.utils.appdirs import get_runtime_path
return get_runtime_path("maestral", config_name + ".sock")
def _delete_pid(config_name):
"""
Deletes the PID file for the given config name.
"""
from maestral.sync.utils.appdirs import get_runtime_path
pid_file = get_runtime_path("maestral", config_name + ".pid")
os.unlink(pid_file)
logger.debug(f"Removed PID file '{pid_file}'.")
:param str config_name: The name of the Maestral configuration to use.
:param bool fallback: If ``True``, a new instance of Maestral will be returned when
the daemon cannot be reached. Defaults to ``False``.
:returns: Pyro proxy of Maestral or a new instance.
:raises: ``Pyro5.errors.CommunicationError`` if the daemon cannot be reached and
``fallback`` is ``False``.
"""
os.environ["MAESTRAL_CONFIG"] = config_name
pid = get_maestral_pid(config_name)
if pid:
from maestral.sync.utils.appdirs import get_runtime_path
sock_name = get_runtime_path("maestral", config_name + ".sock")
sys.excepthook = Pyro5.errors.excepthook
maestral_daemon = client.Proxy(URI.format(config_name, "./u:" + sock_name))
try:
maestral_daemon._pyroBind()
return maestral_daemon
except Pyro5.errors.CommunicationError:
maestral_daemon._pyroRelease()
if fallback:
from maestral.sync.main import Maestral, sh
sh.setLevel(logging.CRITICAL)
m = Maestral(run=False)
return m
else:
raise Pyro5.errors.CommunicationError