diff --git a/zfs-backup.py b/zfs-backup.py index 28a5687..93093c2 100755 --- a/zfs-backup.py +++ b/zfs-backup.py @@ -34,7 +34,6 @@ def main(): # TODO skip post-backup scrub if nothing was sent (except when pre-scrub was also skipped) # TODO print time estimation while scrub is in progress # TODO logging -# TODO type hints class Manager: @@ -46,7 +45,7 @@ class Manager: self._local_pool = local_pool self._external_pool = external_pool - def backup(self): + def backup(self) -> None: """Backup workflow.""" self._external_pool.import_() try: @@ -57,7 +56,7 @@ class Manager: finally: self._external_pool.export() - def _backup_all_datasets(self): + def _backup_all_datasets(self) -> None: """Send snapshots of local datasets to the backup pool.""" print("Backing up datasets") for ( @@ -241,7 +240,7 @@ class ZFSPath: name = "@".join([name, self.snapshot_name]) return name - def _sanity_check(self): + def _sanity_check(self) -> None: for item in self._elements: if len(item) == 0: raise ValueError(Messages.NOT_VALID_ZFS_PATH) @@ -388,7 +387,7 @@ class ZFS(CommandInterface): return [Snapshot(ZFSPath.from_string(name)) for name in snapshot_names] @classmethod - def destroy_snapshot(cls, snapshot: Snapshot): + def destroy_snapshot(cls, snapshot: Snapshot) -> None: _assert_type(snapshot, Snapshot, "Can only destroy snapshots.") if config.local_pool_to_backup == snapshot.pool: raise Exception(Messages.DELETE_IN_LOCAL_POOL) @@ -407,7 +406,9 @@ class ZFS(CommandInterface): ) @classmethod - def _pre_send_sanity_checks(cls, old_snapshot: Snapshot, new_snapshot: Snapshot): + def _pre_send_sanity_checks( + cls, old_snapshot: Snapshot, new_snapshot: Snapshot + ) -> None: for snapshot in [old_snapshot, new_snapshot]: _assert_type( snapshot, @@ -486,7 +487,7 @@ class Disk: def _mapper_entry(self) -> str: return f"crypt-{self._name}" - def encrypt(self): + def encrypt(self) -> None: if not self._decrypted: raise ValueError(Messages.NOT_DECRYPTED, self._name) if self._was_already_decrypted: @@ -618,7 +619,7 @@ class ExternalPool(Pool): print(Messages.CANNOT_FIND_BACKUP_DRIVE) exit(EXIT_ERROR) - def clean_old_snapshots(self): + def clean_old_snapshots(self) -> None: print("Cleaning old snapshots") for dataset in self.datasets: dataset.clean_old_snapshots() @@ -654,11 +655,11 @@ class Dataset: def replace_pool(self, name: str) -> Dataset: return Dataset(self._path.replace_pool(name)) - def clean_old_snapshots(self): + def clean_old_snapshots(self) -> None: for interval, number in config.keep_snapshots_per_interval.keys(): self._clean_old_snapshots_for_interval(interval, number) - def _clean_old_snapshots_for_interval(self, interval: str, max_count: int): + def _clean_old_snapshots_for_interval(self, interval: str, max_count: int) -> None: regex = _get_regex_matching_snapshots_with_tags([config.snapshot_tag, interval]) snapshots = [ snapshot for snapshot in self.snapshots if snapshot.matches_regex(regex) @@ -699,7 +700,7 @@ class Snapshot: def matches_regex(self, regex: str) -> bool: return bool(search(regex, str(self._path))) - def destroy(self): + def destroy(self) -> None: ZFS.destroy_snapshot(self) @property @@ -762,7 +763,7 @@ class Messages: SAME_DATASET = "Cannot send incremental snapshots if start and end snapshot are not based on the same dataset" -def _verify_running_as_root(): +def _verify_running_as_root() -> None: if getuid() > 0: print(Messages.RUN_AS_ROOT) exit(EXIT_ERROR)