SelfPrivacy/selfprivacy-rest-api
1
0
Fork 0
mirror of https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git synced 2024-11-22 12:11:26 +00:00
Code Issues Projects Releases Wiki Activity

refactor(services): break out location construction when moving

Browse source
This commit is contained in:
Houkime 2024-02-19 18:37:00 +00:00
parent baaf3299ce
commit 18934a53e6
2 changed files with 17 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
selfprivacy_api/services/moving.py
View file

@ -19,6 +19,10 @@ def get_foldername(p: OwnedPath) -> str:
return p.path.split("/")[-1] return p.path.split("/")[-1]
def location_at_volume(binding_path: OwnedPath, volume_name: str):
return f"/volumes/{volume_name}/{get_foldername(binding_path)}"
def check_volume(volume: BlockDevice, space_needed: int) -> None: def check_volume(volume: BlockDevice, space_needed: int) -> None:
# Check if there is enough space on the new volume # Check if there is enough space on the new volume
if int(volume.fsavail) < space_needed: if int(volume.fsavail) < space_needed:
@ -29,10 +33,10 @@ def check_volume(volume: BlockDevice, space_needed: int) -> None:
raise MoveError("Volume is not mounted.") raise MoveError("Volume is not mounted.")
def check_folders(current_volume: BlockDevice, folders: List[OwnedPath]) -> None: def check_folders(volume_name: str, folders: List[OwnedPath]) -> None:
# Make sure current actual directory exists and if its user and group are correct # Make sure current actual directory exists and if its user and group are correct
for folder in folders: for folder in folders:
path = pathlib.Path(f"/volumes/{current_volume}/{get_foldername(folder)}") path = pathlib.Path(location_at_volume(folder, volume_name))
if not path.exists(): if not path.exists():
raise MoveError(f"directory {path} is not found.") raise MoveError(f"directory {path} is not found.")
@ -63,20 +67,19 @@ def move_folders_to_volume(
if current_progress is None: if current_progress is None:
current_progress = 0 current_progress = 0
folder_percentage = 50 // len(folders) progress_per_folder = 50 // len(folders)
for folder in folders: for folder in folders:
folder_name = get_foldername(folder)
shutil.move( shutil.move(
f"/volumes/{old_volume_name}/{folder_name}", location_at_volume(folder, old_volume_name),
f"/volumes/{new_volume.name}/{folder_name}", location_at_volume(folder, new_volume.name),
) )
progress = current_progress + folder_percentage progress = current_progress + progress_per_folder
report_progress(progress, job, "Moving data to new volume...") report_progress(progress, job, "Moving data to new volume...")
def ensure_folder_ownership(folders: List[OwnedPath], volume: BlockDevice) -> None: def ensure_folder_ownership(folders: List[OwnedPath], volume: BlockDevice) -> None:
for folder in folders: for folder in folders:
true_location = f"/volumes/{volume.name}/{get_foldername(folder)}" true_location = location_at_volume(folder, volume.name)
try: try:
subprocess.run( subprocess.run(
[ [
@ -89,10 +92,10 @@ def ensure_folder_ownership(folders: List[OwnedPath], volume: BlockDevice) -> No
check=True, check=True,
) )
except subprocess.CalledProcessError as error: except subprocess.CalledProcessError as error:
print(error.output)
error_message = ( error_message = (
f"Unable to set ownership of {true_location} :{error.output}" f"Unable to set ownership of {true_location} :{error.output}"
) )
print(error.output)
raise MoveError(error_message) raise MoveError(error_message)
@ -103,7 +106,7 @@ def bind_folders(folders: List[OwnedPath], volume: BlockDevice) -> None:
[ [
"mount", "mount",
"--bind", "--bind",
f"/volumes/{volume.name}/{get_foldername(folder)}", location_at_volume(folder, volume.name),
folder.path, folder.path,
], ],
check=True, check=True,

6
selfprivacy_api/services/service.py
View file

@ -353,7 +353,7 @@ class Service(ABC):
Move a service to another volume. Move a service to another volume.
""" """
service_name = self.get_display_name() service_name = self.get_display_name()
# TODO: validate that this volume exists # TODO : Make sure device exists
old_volume_name = self.get_drive() old_volume_name = self.get_drive()
owned_folders = self.get_owned_folders() owned_folders = self.get_owned_folders()
@ -365,7 +365,7 @@ class Service(ABC):
report_progress(70, job, f"Making sure {service_name} owns its files...") report_progress(70, job, f"Making sure {service_name} owns its files...")
try: try:
ensure_folder_ownership(owned_folders, new_volume, job, self) ensure_folder_ownership(owned_folders, new_volume)
except Exception as error: except Exception as error:
# We have logged it via print and we additionally log it here in the error field # We have logged it via print and we additionally log it here in the error field
# We are continuing anyway but Job has no warning field # We are continuing anyway but Job has no warning field
@ -399,7 +399,7 @@ class Service(ABC):
report_progress(5, job, f"Stopping {service_name}...") report_progress(5, job, f"Stopping {service_name}...")
assert self is not None assert self is not None
with StoppedService(self): with StoppedService(self):
report_progress(9, job, f"Stopped server, starting the move...") report_progress(9, job, "Stopped service, starting the move...")
self.do_move_to_volume(volume, job) self.do_move_to_volume(volume, job)
return job return job