81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
|
|
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 a valid ZFSPath")
|