add: send_absolute

zfs-backup.py

@@ -314,13 +314,16 @@ class ZFS(CommandInterface):
             # -t   object type
             # -r   recursive
             _base = ["list", "-H", "-d1", "-o", "name"]
-            dataset = _base + ["-t", "filesystem"]
+            dataset = [*_base, "-t", "filesystem"]
-            snapshot = _base + ["-t", "snapshot", "-r"]
+            snapshot = [*_base, "-t", "snapshot", "-r"]
 
-        destroy = ["destroy"]
+        class Send:
             # -R  Replicate filesystem
             # -I  send all intermediary snapshots
-        send = ["send", "-R", "-I"]
+            absolute = ["send", "-R"]
+            incremental = [*absolute, "-I"]
+
+        destroy = ["destroy"]
         # -d  Discard the first element of the "send" snapshot's file system name
         # -F  Force a rollback of the file system to the most recent snapshot before performing the "receive".
         # -u  File system that is associated with the received stream is not mounted.
@@ -357,7 +360,7 @@ class ZFS(CommandInterface):
         cls._pre_send_sanity_checks(old_snapshot, new_snapshot)
         print(f"  {old_snapshot} -> {new_snapshot}")
         return cls._open_stream(
-            *cls._Subcommands.send,
+            *cls._Subcommands.Send.incremental,
             str(old_snapshot),
             str(new_snapshot),
         )
@@ -372,7 +375,7 @@ class ZFS(CommandInterface):
             )
         if old_snapshot.dataset != new_snapshot.dataset:
             raise ValueError(SAME_DATASET)
-        if old_snapshot > new_snapshot:  # chronological ordering
+        if old_snapshot.newer_than(new_snapshot):
             raise ValueError("Old snapshot is newer than new snapshot.")
         if old_snapshot == new_snapshot:
             raise ValueError("Cannot send stream, snapshots are identical.")
@@ -382,6 +385,10 @@ class ZFS(CommandInterface):
         _assert_type(pool, Pool, "Can only receive into pools.")
         cls._receive_stream(*cls._Subcommands.receive, pool.name, stream=stream)
 
+    @classmethod
+    def send_absolute(cls, snapshot: Snapshot) -> Popen:
+        return cls._open_stream(*cls._Subcommands.Send.absolute, str(snapshot))
+
 
 class Time:
     @staticmethod
@@ -640,6 +647,9 @@ class Snapshot:
     def __repr__(self):
         return str(self._path)
 
+    def newer_than(self, other: Snapshot) -> bool:
+        return self > other
+
     @property
     def snapshot_name(self) -> str:
         return self._path.snapshot_name