refactor: ZFS

.idea/inspectionProfiles/profiles_settings.xml

@@ -1,5 +1,6 @@
 <component name="InspectionProjectProfileManager">
   <settings>
+    <option name="PROJECT_PROFILE" value="Default" />
     <option name="USE_PROJECT_PROFILE" value="false" />
     <version value="1.0" />
   </settings>

.idea/misc.xml

@@ -1,4 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.9" project-jdk-type="Python SDK" />
+  <component name="PythonCompatibilityInspectionAdvertiser">
+    <option name="version" value="3" />
+  </component>
 </project>

.idea/zfs-backup.iml

@@ -2,7 +2,7 @@
 <module type="PYTHON_MODULE" version="4">
   <component name="NewModuleRootManager">
     <content url="file://$MODULE_DIR$" />
-    <orderEntry type="inheritedJdk" />
+    <orderEntry type="jdk" jdkName="Python 3.9" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
 </module>

config.py

@@ -32,7 +32,3 @@ keep_snapshots_per_interval = {
     "weekly": 8,
     "monthly": 12,
 }
-
-ZFS = "/usr/bin/zfs"
-ZPOOL = "/usr/bin/zpool"
-CRYPTSETUP = "/usr/bin/cryptsetup"

zfs-backup.py

@@ -10,6 +10,7 @@ from pathlib import Path
 from re import search
 from subprocess import PIPE, Popen, call, check_output
 from sys import exit
+from typing import Any
 from uuid import UUID
 
 import config
@@ -191,11 +192,18 @@ class ZFSPath:
 
 
 class ZFS:
+    # TODO introduce command constants
     @staticmethod
-    def get_datasets(name: str) -> list[Dataset]:
+    def _assert_type(instance: Any, object_type: Any, message: str) -> None:
+        if not isinstance(instance, object_type):
+            raise TypeError(message)
+
+    @classmethod
+    def get_first_level_datasets(cls, pool: Pool) -> list[Dataset]:
+        cls._assert_type(pool, Pool, "Can only get datasets from Pool objects")
         datasets_str = Command.get_output(
             [
-                config.ZFS,
+                ZFS,
                 "list",
                 "-H",
                 "-d",
@@ -204,30 +212,35 @@ class ZFS:
                 "name",
                 "-t",
                 "filesystem",
-                name,
+                pool.name,
             ]
         )
         # we omit the first return value, which is the pool, not the dataset
         datasets_list = datasets_str.split("\n")[1:]
         return [Dataset(ZFSPath.from_string(d)) for d in datasets_list]
 
-    @staticmethod
+    @classmethod
-    def import_pool_from_directory(name: str, directory: Path) -> None:
+    def import_pool_from_directory(cls, pool: Pool, directory: Path) -> None:
+        cls._assert_type(pool, Pool, "Can only import Pool objects")
         # -N: no mount
         # -d: directory to search the pool in
-        Command.run([config.ZPOOL, "import", "-N", "-d", directory, name])
+        Command.run([Binaries.ZPOOL, "import", "-N", "-d", directory, pool.name])
 
-    @staticmethod
+    @classmethod
-    def export_pool(name: str) -> None:
+    def export_pool(cls, pool: Pool) -> None:
-        Command.run([config.ZPOOL, "export", name])
+        cls._assert_type(pool, Pool, "Can only export Pool objects")
+        Command.run([Binaries.ZPOOL, "export", pool.name])
 
-    @staticmethod
+    @classmethod
-    def get_snapshots(name: str) -> list[Snapshot]:
+    def get_snapshots(cls, dataset: Dataset) -> list[Snapshot]:
+        cls._assert_type(
+            dataset, Dataset, "Can only get snapshots from Dataset objects"
+        )
         return [
             Snapshot(ZFSPath.from_string(name))
             for name in Command.get_output(
                 [
-                    config.ZFS,
+                    ZFS,
                     "list",
                     "-r",
                     "-t",
@@ -237,35 +250,48 @@ class ZFS:
                     "-H",
                     "-d",
                     "1",
-                    name,
+                    dataset.qualified_name,
                 ]
             ).split()
         ]
 
-    @staticmethod
+    @classmethod
-    def destroy_snapshot(snapshot: Snapshot, *, recursive=False):
+    def destroy_snapshot(cls, snapshot: Snapshot, *, recursive=False):
-        if not isinstance(snapshot, Snapshot):
+        cls._assert_type(snapshot, Snapshot, "Can only destroy snapshots.")
-            raise TypeError("Can only destroy snapshots.")
         if config.local_pool_to_backup == snapshot.pool:
             raise Exception(DELETE_IN_LOCAL_POOL)
-        cmdline = [config.ZFS, "destroy"]
+        cmdline = [ZFS, "destroy"]
         if recursive:
             cmdline.append("-r")
         cmdline.append(str(snapshot))
         Command.run(cmdline)
 
-    @staticmethod
+    @classmethod
-    def send_incremental(old_snap: Snapshot, new_snap: Snapshot) -> Popen[str]:
+    def send_incremental(
-        if old_snap.dataset != new_snap.dataset:
+        cls, old_snapshot: Snapshot, new_snapshot: Snapshot
-            raise ValueError(SAME_DATASET)
+    ) -> Popen[str]:
-        if old_snap == new_snap:
+        cls._send_sanity_checks(old_snapshot, new_snapshot)
-            raise ValueError("Cannot send stream, snapshots are identical.")
+        print(f"  {old_snapshot} -> {new_snapshot}")
-        print(f"  {old_snap} -> {new_snap}")
         stream = Popen(
-            [config.ZFS, "send", "-R", "-I", str(old_snap), str(new_snap)], stdout=PIPE
+            [ZFS, "send", "-R", "-I", str(old_snapshot), str(new_snapshot)], stdout=PIPE
         )
         return stream
 
+    @classmethod
+    def _send_sanity_checks(cls, old_snapshot: Snapshot, new_snapshot: Snapshot):
+        for snapshot in [old_snapshot, new_snapshot]:
+            cls._assert_type(
+                snapshot,
+                Snapshot,
+                f"Cannot send stream, object is not a Snapshot: {snapshot}",
+            )
+        if old_snapshot.dataset != new_snapshot.dataset:
+            raise ValueError(SAME_DATASET)
+        if old_snapshot > new_snapshot:  # chronological ordering
+            raise ValueError("Old snapshot is newer than new snapshot.")
+        if old_snapshot == new_snapshot:
+            raise ValueError("Cannot send stream, snapshots are identical.")
+
     @staticmethod
     def receive(stream: Popen[str], pool: Pool):
         # -d  Discard the first element of the sent snapshot's file system name
@@ -274,7 +300,7 @@ class ZFS:
         if not isinstance(pool, Pool):
             raise TypeError("Can only receive into pools.")
         Command.receive_from_stream(
-            [config.ZFS, "receive", "-d", "-F", "-u", pool.name],
+            [ZFS, "receive", "-d", "-F", "-u", pool.name],
             stream,
         )
 
@@ -290,7 +316,7 @@ class ZFS:
             raise TypeError("Can only scrub pools.")
         cls._scrub_if_necessary(pool, skip_if_recently_scrubbed)
         if wait_for_finish:
-            Command.run([config.ZPOOL, "wait", "-t", "scrub", pool.name])
+            Command.run([Binaries.ZPOOL, "wait", "-t", "scrub", pool.name])
         if not cls._healthy(pool):
             raise IOError(f"Pool {pool.name} is not healthy.")
 
@@ -299,13 +325,13 @@ class ZFS:
         if cls._scrub_in_progress(pool):
             return
         if not skip_if_recently_scrubbed:
-            Command.run([config.ZPOOL, "scrub", pool.name])
+            Command.run([Binaries.ZPOOL, "scrub", pool.name])
         else:
             if not cls._recently_scrubbed(pool):
-                Command.run([config.ZPOOL, "scrub", pool.name])
+                Command.run([Binaries.ZPOOL, "scrub", pool.name])
 
     @classmethod
-    def _recently_scrubbed(cls, pool):
+    def _recently_scrubbed(cls, pool: Pool) -> bool:
         return datetime.now() - cls._last_scrub(pool) < config.recent_scrub_timedelta
 
     @classmethod
@@ -317,8 +343,8 @@ class ZFS:
         return "ONLINE" in cls._get_pool_status_output(pool)
 
     @staticmethod
-    def _get_pool_status_output(pool) -> str:
+    def _get_pool_status_output(pool: Pool) -> str:
-        return Command.get_output([config.ZFS, "status", pool.name])
+        return Command.get_output([ZFS, "status", pool.name])
 
     @classmethod
     def _last_scrub(cls, pool: Pool) -> datetime:
@@ -384,14 +410,14 @@ class Disk:
             raise ValueError(NOT_DECRYPTED, self._name)
         if self._was_already_decrypted:
             return
-        Command.run([config.CRYPTSETUP, "close", self._mapper_entry])
+        Command.run([Binaries.CRYPTSETUP, "close", self._mapper_entry])
         self._decrypted = False
 
 
 class Cryptsetup:
     @classmethod
     def _status(cls, name: str) -> str:
-        return Command.get_output([config.CRYPTSETUP, "status", name])
+        return Command.get_output([Binaries.CRYPTSETUP, "status", name])
 
     @classmethod
     def decrypt(cls, path: Path, mapper_entry: str) -> bool:
@@ -400,7 +426,7 @@ class Cryptsetup:
                 return True
             print(MAPPER_ENTRY_ALREADY_EXISTS)
             exit(EXIT_ERROR)
-        Command.run([config.CRYPTSETUP, "open", str(path), mapper_entry])
+        Command.run([Binaries.CRYPTSETUP, "open", str(path), mapper_entry])
         return False
 
     @classmethod
@@ -435,7 +461,7 @@ class Pool:
     @property
     @cache
     def datasets(self) -> list[Dataset]:
-        return ZFS.get_datasets(self._name)
+        return ZFS.get_first_level_datasets(self)
 
     @property
     def name(self) -> str:
@@ -452,7 +478,7 @@ class ExternalPool(Pool):
         if self._imported:
             raise ValueError(ALREADY_IMPORTED, self.name)
         self._disk.decrypt()
-        ZFS.import_pool_from_directory(self._name, MAPPER_PATH)
+        ZFS.import_pool_from_directory(self, MAPPER_PATH)
         self._imported = True
 
     def export(self) -> None:
@@ -464,8 +490,7 @@ class ExternalPool(Pool):
         print(f"Exported {self.name}. Disk {self._disk.uuid} can be removed.")
 
     def _export_pool(self) -> None:
-        # TODO cancel scrub if necessary
+        ZFS.export_pool(self)
-        ZFS.export_pool(self._name)
 
     def scrub(self, skip_if_recently_scrubbed: bool) -> None:
         print(f"Scrubbing {self._name}")
@@ -514,7 +539,7 @@ class Dataset:
     @property
     def snapshots(self) -> list[Snapshot]:
         """get qualified snapshot names that are direct children of the dataset"""
-        return ZFS.get_snapshots(self.qualified_name)
+        return ZFS.get_snapshots(self)
 
     def replace_pool(self, name: str) -> Dataset:
         return Dataset(self._path.replace_pool(name))
@@ -595,5 +620,11 @@ def verify_running_as_root():
         exit(EXIT_ERROR)
 
 
+class Binaries:
+    ZFS = Path("/usr/bin/zfs")
+    ZPOOL = Path("/usr/bin/zpool")
+    CRYPTSETUP = Path("/usr/bin/cryptsetup")
+
+
 if __name__ == "__main__":
     main()