more cleanup

zfs-backup.py

@@ -48,13 +48,60 @@ class Manager:
             self._external_pool.export()
 
     def _backup_all_datasets(self):
-        for local_dataset in self._local_pool.datasets_to_backup:
+        for local_dataset in self._datasets_to_backup:
-            remote_dataset = self._external_pool.search_dataset_like(local_dataset)
+            remote_dataset = self._search_matching_dataset_in_remote_pool(
-            last_common_snapshot = get_last_common_snapshot(remote_dataset, local_dataset)
+                local_dataset, self._external_pool
-            last_local_snapshot: Snapshot = local_dataset.last_snapshot
+            )
-            sender = ZFS.send(last_common_snapshot, last_local_snapshot)
+            start_snapshot = self._get_last_common_snapshot(
+                remote_dataset, local_dataset
+            )
+            end_snapshot = self._get_last_snapshot_with_backup_tag(local_dataset)
+            sender = ZFS.send_incremental(start_snapshot, end_snapshot)
             ZFS.receive(sender, self._external_pool.name)
 
+    def _get_last_snapshot_with_backup_tag(self, local_dataset):
+        snapshots = self._find_snapshots_with_backup_tag(local_dataset)
+        snapshots.sort()
+        end_snapshot = snapshots[-1]
+        return end_snapshot
+
+    @property
+    def _datasets_to_backup(self) -> list[Dataset]:
+        return [
+            dataset
+            for dataset in self._local_pool.datasets
+            if dataset.qualified_name not in config.do_not_backup
+        ]
+
+    @staticmethod
+    def _find_snapshots_with_backup_tag(dataset: Dataset) -> list[Snapshot]:
+        regex = Manager._get_regex_matching_backup_tags()
+        return [
+            snapshot for snapshot in dataset.snapshots if snapshot.matches_regex(regex)
+        ]
+
+    @staticmethod
+    def _get_regex_matching_backup_tags():
+        return "@" + ".*".join([config.snapshot_tag, config.snapshot_interval])
+
+    @staticmethod
+    def _search_matching_dataset_in_remote_pool(dataset: Dataset, pool: Pool):
+        dataset_to_search = dataset.replace_pool(pool.name)
+        for dataset in pool.datasets:
+            if dataset == dataset_to_search:
+                return dataset
+
+    @classmethod
+    def _get_last_common_snapshot(cls, ds1: Dataset, ds2: Dataset) -> Snapshot:
+        common_snapshots = [
+            local_snap
+            for local_snap in list(cls._find_snapshots_with_backup_tag(ds1))
+            for remote_snap in list(cls._find_snapshots_with_backup_tag(ds2))
+            if local_snap.snapname == remote_snap.snapname
+        ]
+        common_snapshots.sort()
+        return common_snapshots[-1]
+
 
 class ZFSPath:
     def __init__(self, elements: list[str], snapname: Optional[str] = None):
@@ -74,6 +121,13 @@ class ZFSPath:
     def __getitem__(self, item):
         return self._qualified_name_as_list[item]
 
+    def __eq__(self, other: ZFSPath):
+        return (
+            self._elements == other._elements
+            and self._is_snapshot == other._is_snapshot
+            and self._snapname == other._snapname
+        )
+
     @property
     def _qualified_name_as_list(self) -> list[str]:
         if self.is_snapshot:
@@ -96,8 +150,16 @@ class ZFSPath:
     def is_snapshot(self) -> bool:
         return self._is_snapshot
 
+    @property
+    def pool(self) -> str:
+        return self._elements[0]
+
+    @property
+    def dataset(self) -> ZFSPath:
+        return ZFSPath(self._elements)
+
     def replace_pool(self, pool: str) -> ZFSPath:
-        return ZFSPath([pool] + self._elements[1:], self.snapname)
+        return ZFSPath([pool] + self._elements[1:].copy(), self.snapname)
 
     @property
     def name_without_pool(self) -> str:
@@ -176,7 +238,11 @@ class ZFS:
         run_command(cmdline)
 
     @staticmethod
-    def send(old_snap: Snapshot, new_snap: Snapshot) -> Popen[str]:
+    def send_incremental(old_snap: Snapshot, new_snap: Snapshot) -> Popen[str]:
+        if old_snap.dataset != new_snap.dataset:
+            raise ValueError(
+                "Cannot send incremental snapshots if start and end snapshot are not based on the same dataset"
+            )
         sender = Popen(
             [config.ZFS, "send", "-R", "-I", old_snap, new_snap], stdout=PIPE
         )
@@ -236,17 +302,6 @@ class Disk:
         self._decrypted = False
 
 
-def get_last_common_snapshot(ds1: Dataset, ds2: Dataset) -> Snapshot:
-    common_snapshots = [
-        local_snap
-        for local_snap in ds1.snapshots_matching_backup_tags
-        for remote_snap in ds2.snapshots_matching_backup_tags
-        if local_snap.snapname == remote_snap.snapname
-    ]
-    common_snapshots.sort()
-    return common_snapshots[-1]
-
-
 class Pool:
     def __init__(self, name: str):
         self._name = name
@@ -261,12 +316,8 @@ class Pool:
     def name(self) -> str:
         return self._name
 
-    @property
+    # def datasets_to_backup(self, do_not_backup: list[str]) -> list[Dataset]:
-    @cache
+    # return [d for d in self.datasets if d.qualified_name not in do_not_backup]
-    def datasets_to_backup(self) -> list[Dataset]:
-        return [
-            d for d in self.datasets if d.qualified_name not in config.do_not_backup
-        ]
 
 
 class ExternalPool(Pool):
@@ -310,25 +361,17 @@ class ExternalPool(Pool):
             print("Could not find a backup drive")
             exit(1)
 
-    @property
+    # def datasets_to_backup(self, do_not_backup: list[str]) -> list[Dataset]:
-    @cache
+    # def replace_pool_name(old_name: str):
-    def datasets_to_backup(self) -> list[Dataset]:
+    # "/".join([self._name, old_name.split("/", maxsplit=1)[1]])
-        def replace_pool_name(old_name: str):
+    #
-            "/".join([self._name, old_name.split("/", maxsplit=1)[1]])
+    # do_not_backup_correct_name = [replace_pool_name(item) for item in do_not_backup]
-
+    # return [d for d in self.datasets if d.qualified_name not in do_not_backup_correct_name]
-        do_not_backup = [replace_pool_name(item) for item in config.do_not_backup]
-        return [d for d in self.datasets if d.qualified_name not in do_not_backup]
 
     def clean_old_snapshots(self):
         for dataset in self.datasets:
             dataset.clean_old_snapshots()
 
-    def search_dataset_like(self, remote_dataset: Dataset) -> Dataset:
-        dataset_to_search = remote_dataset.replace_pool(self.name)
-        for dataset in self.datasets:
-            if dataset == dataset_to_search:
-                return dataset
-
 
 class Dataset:
     def __init__(self, zfs_path: ZFSPath):
@@ -338,7 +381,7 @@ class Dataset:
         return self.qualified_name
 
     def __eq__(self, other: Dataset):
-        return str(self) == str(other)
+        return self._path == other._path
 
     @classmethod
     def from_string(cls, qualified_name: str) -> Dataset:
@@ -357,20 +400,6 @@ class Dataset:
         """get qualified snapshot names that are direct children of the dataset"""
         return ZFS.get_snapshots(self.qualified_name)
 
-    @property
-    def snapshots_matching_backup_tags(self) -> list[Snapshot]:
-        snapshots = [
-            snapshot for snapshot in self.snapshots if snapshot.matches_backup_tags
-        ]
-        snapshots.sort()
-        return snapshots
-
-    @property
-    def last_snapshot(self) -> Snapshot:
-        for snapshot in self.snapshots[::-1]:
-            if snapshot.matches_backup_tags:
-                return snapshot
-
     def replace_pool(self, name: str) -> Dataset:
         return Dataset(self._path.replace_pool(name))
 
@@ -404,17 +433,20 @@ class Snapshot:
     def snapname(self) -> str:
         return self._path.snapname
 
-    @property
-    def matches_backup_tags(self) -> bool:
-        regex = "@" + ".*".join([config.snapshot_tag, config.snapshot_interval])
-        return self.matches_regex(regex)
-
     def matches_regex(self, regex: str) -> bool:
         return bool(re.search(regex, str(self._path)))
 
     def destroy(self):
         ZFS.destroy(str(self))
 
+    @property
+    def pool(self) -> str:
+        return self._path.pool
+
+    @property
+    def dataset(self) -> Dataset:
+        return Dataset(self._path.dataset)
+
 
 def run_command(cmdline: list[str]):
     subprocess.call(cmdline)