From 34e3e41382a6d8599fc6443cc02d01118c6fe869 Mon Sep 17 00:00:00 2001 From: Fuxino Date: Wed, 21 Oct 2026 22:44:19 +0200 Subject: [PATCH] Share one removal loop between local and remote backups remove_old_backups() repeated the counting and sorting in both branches (the keep < -1 bug was in both); listing and removing are now per-branch helpers. --- simple_backup/simple_backup.py | 137 +++++++++++++++++---------------- 1 file changed, 69 insertions(+), 68 deletions(-) diff --git a/simple_backup/simple_backup.py b/simple_backup/simple_backup.py index 1a0ed72..db00aac 100755 --- a/simple_backup/simple_backup.py +++ b/simple_backup/simple_backup.py @@ -318,6 +318,41 @@ class Backup: def remove_old_backups(self) -> None: """Remove old backups if there are more than indicated by 'keep'""" + dirs = self._list_backups() + + if dirs is None: + return + + n_backup = len(dirs) + + # The new backup is already in the list, unless old ones are removed before it + if not self._remove_before: + n_backup -= 1 + + count = 0 + + if n_backup > self.keep: + logger.info('Removing old backups...') + dirs.sort() + + n_remove = n_backup - self.keep + + for i in range(n_remove): + logger.info('Removing backup %s (%d/%d)...', dirs[i], i + 1, n_remove) + + if self._remove_backup(dirs[i]): + count += 1 + + self._removed_count = count + + if count == 1: + logger.info('Removed %d backup', count) + elif count > 1: + logger.info('Removed %d backups', count) + + def _list_backups(self) -> list[str] | None: + """Names of the backup directories, or None if they can't be listed (then nothing is removed)""" + if self._remote: assert self._ssh is not None @@ -335,88 +370,54 @@ class Backup: logger.error('Cannot list old backups on the server. Old backups will not be removed') logger.error(err) - return + return None - dirs = [os.path.basename(d) for d in output.split('\n') if d != ''] + return [os.path.basename(d) for d in output.split('\n') if d != ''] - n_backup = len(dirs) + try: + # Only count directories, so that stray files don't count as backups + return [f.name for f in os.scandir(f'{self.output}/simple_backup') if f.is_dir(follow_symlinks=False)] + except FileNotFoundError: + return None + except PermissionError: + logger.error('Cannot list old backups. Permission denied. Old backups will not be removed') - if not self._remove_before: - n_backup -= 1 + return None - count = 0 + def _remove_backup(self, name: str) -> bool: + """Remove one backup directory; errors are logged""" - if n_backup > self.keep: - logger.info('Removing old backups...') - dirs.sort() + path = f'{self.output}/simple_backup/{name}' - n_remove = n_backup - self.keep + if self._remote: + assert self._ssh is not None - for i in range(n_remove): - logger.info('Removing backup %s (%d/%d)...', dirs[i], i + 1, n_remove) + sudo = 'sudo ' if self.remote_sudo else '' + _, _, stderr = self._ssh.exec_command(f'{sudo}rm -r {shlex.quote(path)}') - if self.remote_sudo: - _, _, stderr = self._ssh.exec_command( - f'sudo rm -r {shlex.quote(f"{self.output}/simple_backup/{dirs[i]}")}') - else: - _, _, stderr = self._ssh.exec_command( - f'rm -r {shlex.quote(f"{self.output}/simple_backup/{dirs[i]}")}') + err = stderr.read().decode('utf-8').strip().split('\n')[0] - err = stderr.read().decode('utf-8').strip().split('\n')[0] + if err != '': + logger.error('Error while removing backup %s.', name) + logger.error(err) - if err != '': - logger.error( - 'Error while removing backup %s.', dirs[i]) - logger.error(err) - else: - count += 1 - else: - try: - # Only count directories, so that stray files don't count as backups - dirs = [f.name for f in os.scandir(f'{self.output}/simple_backup') - if f.is_dir(follow_symlinks=False)] - except FileNotFoundError: - return - except PermissionError: - logger.error('Cannot list old backups. Permission denied. Old backups will not be removed') + return False - return + return True - n_backup = len(dirs) + try: + rmtree(path) - if not self._remove_before: - n_backup -= 1 + return True + except FileNotFoundError: + logger.error('Error while removing backup %s. Directory not found', name) + except PermissionError: + logger.error('Error while removing backup %s. Permission denied', name) + except OSError as e: + # E.g. read-only filesystem or device busy + logger.error('Error while removing backup %s. %s', name, e.strerror or e) - count = 0 - - if n_backup > self.keep: - logger.info('Removing old backups...') - dirs.sort() - - n_remove = n_backup - self.keep - - for i in range(n_remove): - logger.info('Removing backup %s (%d/%d)...', dirs[i], i + 1, n_remove) - - try: - rmtree(f'{self.output}/simple_backup/{dirs[i]}') - count += 1 - except FileNotFoundError: - logger.error( - 'Error while removing backup %s. Directory not found', dirs[i]) - except PermissionError: - logger.error( - 'Error while removing backup %s. Permission denied', dirs[i]) - except OSError as e: - # E.g. read-only filesystem or device busy - logger.error('Error while removing backup %s. %s', dirs[i], e.strerror or e) - - self._removed_count = count - - if count == 1: - logger.info('Removed %d backup', count) - elif count > 1: - logger.info('Removed %d backups', count) + return False def find_last_backup(self) -> None: """Get path of last backup (from last_backup symlink) for rsync --link-dest"""