use a lighter task for suggesting file path in backup/fileset

Author: Fallen-Breath

prime_backup/action/get_file_action.py

@@ -67,6 +67,17 @@ def run(self) -> Dict[str, FileInfo]:
 			return {file.path: FileInfo.of(file) for file in files}
 
 
+class GetBackupFilePathsAction(Action[List[str]]):
+	def __init__(self, backup_id: int):
+		super().__init__()
+		self.backup_id = backup_id
+
+	@override
+	def run(self) -> List[str]:
+		with DbAccess.open_session() as session:
+			return session.get_backup_file_paths(self.backup_id)
+
+
 class GetFilesetFileAction(_GetSingleFileActionBase):
 	def __init__(self, fileset_id: int, file_path: PathLike, *, count_backups: bool = False, sample_backup_num: Optional[int] = None):
 		super().__init__(count_backups=count_backups, sample_backup_num=sample_backup_num)
@@ -91,6 +102,17 @@ def run(self) -> Dict[str, FileInfo]:
 			return {file.path: FileInfo.of(file) for file in files}
 
 
+class GetFilesetFilePathsAction(Action[List[str]]):
+	def __init__(self, fileset_id: int):
+		super().__init__()
+		self.fileset_id = fileset_id
+
+	@override
+	def run(self) -> List[str]:
+		with DbAccess.open_session() as session:
+			return session.get_fileset_file_paths(self.fileset_id)
+
+
 class NotDirectoryError(PrimeBackupError):
 	pass
 

prime_backup/db/session.py

@@ -538,6 +538,21 @@ def get_fileset_files(self, fileset_id: Optional[int]) -> List[schema.File]:
 			where(schema.File.fileset_id == fileset_id)
 		).scalars().all())
 
+	def get_fileset_file_paths(self, fileset_id: int) -> List[str]:
+		return _list_it(self.session.execute(
+			select(schema.File.path).
+			where(schema.File.fileset_id == fileset_id)
+		).scalars().all())
+
+	def get_fileset_file_path_and_role(self, fileset_id: int) -> Dict[str, int]:
+		return {
+			row.path: row.role
+			for row in self.session.execute(
+				select(schema.File.path, schema.File.role).
+				where(schema.File.fileset_id == fileset_id)
+			)
+		}
+
 	def get_fileset_ids_by_blob_hashes(self, hashes: List[str]) -> List[int]:
 		fileset_ids = set()
 		for v_hashes in collection_utils.slicing_iterate(hashes, self.__safe_var_limit):
@@ -775,6 +790,16 @@ def get_backup_files(self, backup_or_backup_id: Union[int, schema.Backup]) -> Li
 		files_delta = self.get_fileset_files(backup.fileset_id_delta)
 		return self.merge_fileset_files(files_base, files_delta)
 
+	def get_backup_file_paths(self, backup_or_backup_id: Union[int, schema.Backup]) -> List[str]:
+		backup = self.__convert_backup_or_backup_id_to_backup(backup_or_backup_id)
+		file_paths: Dict[str, None] = dict.fromkeys(self.get_fileset_file_paths(backup.fileset_id_base), None)
+		for path, role in self.get_fileset_file_path_and_role(backup.fileset_id_delta).items():
+			if role == FileRole.delta_add.value:
+				file_paths[path] = None
+			elif role == FileRole.delta_remove:
+				file_paths.pop(path, None)
+		return list(file_paths.keys())
+
 	def get_backups(self, backup_ids: List[int]) -> Dict[int, schema.Backup]:
 		"""
 		:return: a dict, backup id -> Backup. All given ids are in the dict

prime_backup/mcdr/task/db/get_backup_file_paths_task.py

@@ -3,7 +3,7 @@
 from mcdreforged.api.all import CommandSource
 from typing_extensions import override
 
-from prime_backup.action.get_file_action import GetBackupFilesAction
+from prime_backup.action.get_file_action import GetBackupFilePathsAction
 from prime_backup.mcdr.task.backup.transform_backup_id_task import TransformBackupIdTask
 from prime_backup.mcdr.task.basic_task import LightTask
 
@@ -24,5 +24,4 @@ def run(self) -> List[str]:
 			backup_id = self.backup_id
 		else:  # backup id raw argument
 			backup_id = TransformBackupIdTask(self.source, [self.backup_id]).run()[0]
-		file_dict = GetBackupFilesAction(backup_id).run()
-		return list(file_dict.keys())
+		return GetBackupFilePathsAction(backup_id).run()

prime_backup/mcdr/task/db/get_fileset_file_paths_task.py

@@ -3,7 +3,7 @@
 from mcdreforged.api.all import CommandSource
 from typing_extensions import override
 
-from prime_backup.action.get_file_action import GetFilesetFilesAction
+from prime_backup.action.get_file_action import GetFilesetFilePathsAction
 from prime_backup.mcdr.task.basic_task import LightTask
 
 
@@ -19,5 +19,4 @@ def id(self) -> str:
 
 	@override
 	def run(self) -> List[str]:
-		file_dict = GetFilesetFilesAction(self.fileset_id).run()
-		return list(file_dict.keys())
+		return GetFilesetFilePathsAction(self.fileset_id).run()