refactor: type hinting

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)