Compare commits

4 Commits

Author SHA1 Message Date
furtest 7cb96fcd45 synchroniser : misc pylint suggested fixes
- Do not use mutables as default arguments
- Remove trailing whitespaces
- Initialise using [] and {} instead of list() and dict()
- Use is not None instead of != None
- Some other small stuff
2026-05-05 12:25:52 +02:00
furtest 22d30b4df7 synchroniser : use with when using Popen
Creating a subprocess with Popen is allocating ressources (according to
pylint) so wrap it in a with to avoid problems.

Also link to unison documentation for the return code value.
2026-05-05 12:11:07 +02:00
furtest 8733167ae3 synchroniser : do not use mutable as default argument
Default arguments are evaluated only once meaning if they are mutables
they are shared between every function invocation.
See the warning here:
https://docs.python.org/3/tutorial/controlflow.html#default-argument-values

So use None instead of [] and replace it by the list in the body of the
function.
2026-05-05 12:05:18 +02:00
furtest b15bb9052d synchroniser : remove wait for foreground links update
The run function already waits for the command to finish and wheter or
not it actually runs in the background is handled before using & and
nohup.
Furthermore it also appears that this code might be broken.
So there is only one thing to do : remove it
2026-05-05 11:25:52 +02:00
+48 -35
View File
@@ -10,7 +10,6 @@ the remote.
import subprocess
import os
import sys
import time
import logging
from pathlib import Path
@@ -50,26 +49,27 @@ class Synchroniser:
"""
def __init__(self, remote:str, local:str, user:str, ip:str, port:int=22,
args_bool:list=[], args_value:dict={}, ssh_settings:dict={},
backup:BackupConfig | None = None
args_bool:list[str] | None = None, args_value:dict[str, str] | None = None,
ssh_settings:dict[str, str] | None = None, backup:BackupConfig | None = None
):
"""Initialises an instance of Synchroniser.
"""
self.remote_dir:str = remote
self.local:str = local
self.args_bool:list[str] = args_bool
self.args_value:dict[str, str] = args_value
self.ssh_settings:dict[str, str] = dict()
self.args_bool:list[str] = args_bool if args_bool is not None else []
self.args_value:dict[str, str] = args_value if args_value is not None else {}
self.ssh_settings:dict[str, str] = ssh_settings if ssh_settings is not None else {}
self.remote_user:str = user
self.remote_ip:str = ip
self.remote_ip:str = ip
self.remote_port:int = port
self.files_extra:list = list()
self.links_extra:list = list()
self.files_extra:list = []
self.links_extra:list = []
self.control_path:str = ""
if(backup != None and backup.enabled):
if backup is not None and backup.enabled:
backup = cast(BackupConfig, backup)
self.files_extra.append("-backup")
if(backup.selection != ""):
if backup.selection != "":
self.files_extra.append(backup.selection)
else:
self.files_extra.append("Name *")
@@ -91,8 +91,10 @@ class Synchroniser:
"-ignore",
f"Name {backup.backupprefix[:-1]}"
])
def create_ssh_master_connection(self, control_path:str="~/.ssh/control_%C", connection_timeout:int=60) -> None:
def create_ssh_master_connection(self, control_path:str="~/.ssh/control_%C",
connection_timeout:int=60
) -> None:
"""Creates an ssh master connection.
It is used so the user only has to authenticate once to the remote server.
@@ -123,15 +125,15 @@ class Synchroniser:
f"{self.remote_user}@{self.remote_ip}",
"-p", str(self.remote_port)
]
master_ssh = subprocess.Popen(command)
# TODO: Raise an exception instead of changing the return value
try:
ret_code = master_ssh.wait(timeout=connection_timeout)
except subprocess.TimeoutExpired as e:
print("Time to login expired", file=sys.stderr)
raise e
except KeyboardInterrupt as e:
raise e
with subprocess.Popen(command) as master_ssh:
# TODO: Raise an exception instead of changing the return value
try:
ret_code = master_ssh.wait(timeout=connection_timeout)
except subprocess.TimeoutExpired as e:
print("Time to login expired", file=sys.stderr)
raise e
except KeyboardInterrupt as e:
raise e
if ret_code != 0:
print("Login to remote failed", file=sys.stderr)
@@ -151,8 +153,10 @@ class Synchroniser:
f"{self.remote_user}@{self.remote_ip}",
"-p", str(self.remote_port)
]
close = subprocess.Popen(command)
return close.wait()
with subprocess.Popen(command) as close:
retval = close.wait()
return retval
def sync_files(self, paths:list, force:bool=False) -> None:
"""Synchronises the files.
@@ -190,8 +194,8 @@ class Synchroniser:
)
def sync(self, remote_root:str, local_root:str,
paths:list=[], ignore:list=[], force:bool=False,
other:list=[]
paths:list | None=None, ignore:list | None = None, force:bool=False,
other:list | None = None
) -> None:
"""Performs the synchronisation by calling unison.
@@ -217,6 +221,13 @@ class Synchroniser:
was interrupted.
If this happens propagate the error to unisync.
"""
if paths is None:
paths = []
if ignore is None:
ignore = []
if other is None:
other = []
command = [ "/usr/bin/unison", "-root", remote_root, "-root", local_root ]
for arg in self.args_bool:
command.append(f"-{arg}")
@@ -247,8 +258,10 @@ class Synchroniser:
for arg in other:
command.append(arg)
proc = subprocess.Popen(command)
ret_code = proc.wait()
with subprocess.Popen(command) as proc:
ret_code = proc.wait()
# See unison manual section 6.11
if ret_code == 3:
raise FatalSyncError("Synchronisation could not be completed")
@@ -279,12 +292,12 @@ class Synchroniser:
link_background_wrapper
]
link_update_process = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
if not background:
print("Starting links update.")
link_update_process.wait()
print("Done")
print("Updating links")
subprocess.run(command,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False
)
def mount_remote_dir(self):
"""Mounts the remote directory to make the local links work.
@@ -311,5 +324,5 @@ class Synchroniser:
f"{self.remote_user}@{self.remote_ip}:{self.remote_dir}/.data",
str(path_to_mount)
]
completed_process = subprocess.run(command)
completed_process = subprocess.run(command, check=True)
completed_process.check_returncode()