start introducing modules

zfs_backup/zfs_path.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from zfs_backup.messages import NOT_VALID_ZFS_PATH
+
+
+class ZFSPath:
+    """Represents a path in ZFS.
+
+    Used to represent the path for objects of Pool, Dataset and Snapshot classes."""
+
+    def __init__(self, elements: list[str], snapshot_name: str | None = None):
+        self._elements = elements
+        self._snapshot_name = snapshot_name
+        self._is_snapshot = False
+        if snapshot_name:
+            self._is_snapshot = True
+        self._sanity_check(self._elements)
+
+    def __repr__(self):
+        """Return the path in the same manner as `zfs`."""
+        path = "/".join(self._elements)
+        if self._is_snapshot:
+            assert self._snapshot_name is not None
+            return "@".join([path, self._snapshot_name])
+        return path
+
+    def __eq__(self, other: ZFSPath):
+        return (
+            self._elements == other._elements
+            and self._is_snapshot == other._is_snapshot
+            and self._snapshot_name == other._snapshot_name
+        )
+
+    def __hash__(self):
+        return str(self).__hash__()
+
+    @classmethod
+    def from_string(cls, string: str) -> ZFSPath:
+        """Take a string from the `zfs` cli, return a ZFSPath object."""
+        snapshot_name = None
+        if "@" in string:
+            string, snapshot_name = string.split("@")
+        elements: list[str] = string.split("/")
+        return ZFSPath(elements, snapshot_name)
+
+    @property
+    def snapshot_name(self) -> str | None:
+        """Return the name of the snapshot, or None if ZFSPath does not point to a snapshot."""
+        return self._snapshot_name
+
+    @property
+    def is_snapshot(self) -> bool:
+        return self._is_snapshot
+
+    @property
+    def pool_name(self) -> str:
+        """Return the name."""
+        return self._elements[0]
+
+    @property
+    def dataset(self) -> ZFSPath:
+        """Return the `zfs` path down to the dataset (omit snapshot)."""
+        return ZFSPath(self._elements)
+
+    def replace_pool(self, pool: str) -> ZFSPath:
+        """Return a new ZFSPath. Replace the old pool with a pool with new name."""
+        return ZFSPath([pool] + self._elements[1:].copy(), self.snapshot_name)
+
+    @property
+    def name_without_pool(self) -> str:
+        """Return the `zfs` path, but without the pool name."""
+        name = "/".join(self._elements[1:])
+        if self.is_snapshot:
+            assert self.snapshot_name
+            name = "@".join([name, self.snapshot_name])
+        return name
+
+    @staticmethod
+    def _sanity_check(elements: list[str]) -> None:
+        for item in elements:
+            if len(item) == 0:
+                raise ValueError(NOT_VALID_ZFS_PATH)