docstrings

zfs-backup.py

@@ -29,7 +29,6 @@ def main():
 
 
 # TODO tests
-# TODO docstrings
 # TODO --skip-import
 # TODO skip post-backup scrub if nothing was sent (except when pre-scrub was also skipped)
 # TODO print time estimation while scrub is in progress
@@ -70,6 +69,7 @@ class Manager:
             ZFS.receive(stream, self._external_pool)
 
     def _get_user_filtered_dataset_backup_map(self) -> _map_type:
+        """Get the dataset map, let user confirm new datasets."""
         return self._user_confirm_new_datasets(self._get_raw_dataset_map())
 
     def _user_confirm_new_datasets(self, dataset_map: _map_type) -> _map_type:
@@ -185,6 +185,10 @@ def _get_regex_matching_snapshots_with_tags(tags: list[str]):
 
 
 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
@@ -194,6 +198,7 @@ class ZFSPath:
         self._sanity_check()
 
     def __repr__(self):
+        """Return the path in the same manner as `zfs`."""
         path = "/".join(self._elements)
         if self._is_snapshot:
             return "@".join([path, self._snapshot_name])
@@ -208,6 +213,7 @@ class ZFSPath:
 
     @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("@")
@@ -215,7 +221,8 @@ class ZFSPath:
         return ZFSPath(elements, snapshot_name)
 
     @property
-    def snapshot_name(self) -> str:
+    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
@@ -223,18 +230,22 @@ class ZFSPath:
         return self._is_snapshot
 
     @property
-    def pool(self) -> str:
+    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:
             name = "@".join([name, self.snapshot_name])
@@ -285,11 +296,13 @@ class ZPool(CommandInterface):
 
     @classmethod
     def import_from_directory(cls, pool: Pool, directory: Path) -> None:
+        """Search `directory` for `pool` and import it."""
         _assert_type(pool, Pool, "Can only import Pool objects")
         cls._run_command(*cls._Subcommands.import_, str(directory), pool.name)
 
     @classmethod
     def export(cls, pool: Pool) -> None:
+        """Export the pool."""
         _assert_type(pool, Pool, "Can only export Pool objects")
         cls._run_command(*cls._Subcommands.export, pool.name)
 
@@ -301,6 +314,9 @@ class ZPool(CommandInterface):
         skip_if_recently_scrubbed: bool,
         wait_for_finish: bool = True,
     ) -> None:
+        """Scrub the pool.
+
+        Waits for the scrub to finish. Raise an IOError if the pool reports as not healthy after the scrub."""
         _assert_type(pool, Pool, "Can only scrub pools.")
         cls._scrub_if_necessary(pool, skip_if_recently_scrubbed)
         if wait_for_finish:
@@ -310,6 +326,8 @@ class ZPool(CommandInterface):
 
     @classmethod
     def _scrub_if_necessary(cls, pool: Pool, skip_if_recently_scrubbed: bool) -> None:
+        """Scrub the pool. If"""
+
         def do_scrub():
             cls._run_command(*cls._Subcommands.scrub, pool.name)
 
@@ -323,6 +341,7 @@ class ZPool(CommandInterface):
 
     @classmethod
     def _recently_scrubbed(cls, pool: Pool) -> bool:
+        """Is the interval after the last scrub smaller than the minimum timedelta?"""
         return datetime.now() - cls._last_scrub(pool) < config.recent_scrub_timedelta
 
     @classmethod
@@ -335,10 +354,12 @@ class ZPool(CommandInterface):
 
     @classmethod
     def _get_pool_status_output(cls, pool: Pool) -> str:
+        """Return output of `zpool status [pool]`."""
         return cls._get_output(*cls._Subcommands.status, pool.name)
 
     @classmethod
     def _last_scrub(cls, pool: Pool) -> datetime:
+        """Get time of the last scrub."""
         return Time.get_last_scrub_from_status_output(cls._get_pool_status_output(pool))
 
 
@@ -373,6 +394,9 @@ class ZFS(CommandInterface):
 
     @classmethod
     def get_first_level_datasets(cls, pool: Pool) -> list[Dataset]:
+        """Get the first level datasets of a pool.
+
+        If a pool `p` has the datasets `p/a`, `p/a/b` and `p/c`, return `[p/a, p/c]`."""
         _assert_type(pool, Pool, "Can only get datasets from Pool objects")
         datasets_str = cls._get_output(*cls._Subcommands.List.dataset, pool.name)
         # we omit the first return value, which is the pool, not the dataset
@@ -381,6 +405,7 @@ class ZFS(CommandInterface):
 
     @classmethod
     def get_snapshots(cls, dataset: Dataset) -> list[Snapshot]:
+        """Get list of snapshots that are associated with a dataset."""
         _assert_type(dataset, Dataset, "Can only get snapshots from Dataset object")
         command = [*cls._Subcommands.List.snapshot, dataset.qualified_name]
         snapshot_names: list[str] = cls._get_output(*command).split("\n")
@@ -388,6 +413,7 @@ class ZFS(CommandInterface):
 
     @classmethod
     def destroy_snapshot(cls, snapshot: Snapshot) -> None:
+        """Non-recursively destroy a snapshot."""
         _assert_type(snapshot, Snapshot, "Can only destroy snapshots.")
         if config.local_pool_to_backup == snapshot.pool:
             raise Exception(Messages.DELETE_IN_LOCAL_POOL)
@@ -397,6 +423,7 @@ class ZFS(CommandInterface):
     def send_incremental(
         cls, old_snapshot: Snapshot, new_snapshot: Snapshot
     ) -> Popen[str]:
+        """Send the incremental replicating stream between two snapshots."""
         cls._pre_send_sanity_checks(old_snapshot, new_snapshot)
         print(f"  {old_snapshot} -> {new_snapshot}")
         return cls._open_stream(
@@ -424,11 +451,13 @@ class ZFS(CommandInterface):
 
     @classmethod
     def receive(cls, stream: Popen[str], pool: Pool):
+        """Receive a stream generated with `zfs send` into a pool."""
         _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:
+        """Send the non-incremental replicating stream of the snapshot."""
         print(f"  {snapshot} -> [new]")
         return cls._open_stream(*cls._Subcommands.Send.absolute, str(snapshot))
 
@@ -448,10 +477,12 @@ class Time:
 
     @classmethod
     def _extract_datetime_segment(cls, line_with_datetime: str) -> str:
+        """From the line in `zpool status` that includes the date and time, extract only the part with date and time."""
         return line_with_datetime.split(" errors on ")[-1]
 
     @classmethod
     def _get_line_with_datetime_from_status_output(cls, output: str) -> str:
+        """From `zpool status`, extract the line that contains the date and time of the last scrub."""
         lines = output.split("\n")
         for line in lines:
             if line.startswith("  scan: scrub repaired"):
@@ -461,6 +492,7 @@ class Time:
 
     @classmethod
     def get_last_scrub_from_status_output(cls, output: str) -> datetime:
+        """Given the output of `zfs status`, return the `datetime` object of the last scrub."""
         line_with_datetime = cls._get_line_with_datetime_from_status_output(output)
         date_str = cls._extract_datetime_segment(line_with_datetime)
         return datetime.strptime(date_str, cls._get_date_format())
@@ -474,6 +506,7 @@ class Disk:
         self._name = name
 
     def decrypt(self) -> None:
+        """Open a LUKS-encrypted disk with `cryptsetup`. Silently pass if it was already decrypted."""
         if self._decrypted:
             raise ValueError(Messages.ALREADY_DECRYPTED, self._name)
         self._was_already_decrypted = Cryptsetup.decrypt(self._path, self._mapper_entry)
@@ -481,13 +514,16 @@ class Disk:
 
     @property
     def _path(self) -> Path:
+        """Unique path of the disk using /dev/disk/by-uuid."""
         return DISK_BY_UUID / str(self.uuid)
 
     @property
     def _mapper_entry(self) -> str:
+        """Name under which the decrypted disk will appear in /dev/mapper."""
         return f"crypt-{self._name}"
 
     def encrypt(self) -> None:
+        """Close a LUKS-encrypted container with `cryptsetup`."""
         if not self._decrypted:
             raise ValueError(Messages.NOT_DECRYPTED, self._name)
         if self._was_already_decrypted:
@@ -506,10 +542,12 @@ class Cryptsetup(CommandInterface):
 
     @classmethod
     def encrypt(cls, mapper_entry: str) -> None:
+        """Close a previously open LUKS container."""
         cls._run_command(cls._Subcommands.close, mapper_entry)
 
     @classmethod
     def _status(cls, name: str) -> str:
+        """Return `cryptsetup status` of a device under `/dev/mapper`"""
         return cls._get_output(cls._Subcommands.status, name)
 
     @classmethod
@@ -529,12 +567,17 @@ class Cryptsetup(CommandInterface):
 
     @classmethod
     def _get_device_by_uuid_from_status_output(cls, status: str) -> Path:
-        device = cls._get_device_name(status)
+        """Given the output of `cryptsetup status [device]`,
+        get the unique name of the disk as path in /dev/disk/by-uuid."""
+        device = cls._get_device_path(status)
         result = cls._resolve_uuid_of_device(device)
         return result
 
     @classmethod
     def _resolve_uuid_of_device(cls, device):
+        """Given a device like `/dev/sda`, resolve its entry under `/dev/disk/by-uuid`.
+
+        Exit if none can be found."""
         for item in DISK_BY_UUID.iterdir():
             if item.readlink() == device:
                 result = item
@@ -546,12 +589,14 @@ class Cryptsetup(CommandInterface):
         return result
 
     @classmethod
-    def _get_device_name(cls, status):
+    def _get_device_path(cls, status) -> Path:
+        """Given the output of `cryptsetup status`, return the path of the device."""
         for line in status.split("\n"):
             if line.startswith("  device:"):
                 device = Path(line.split(":")[-1].strip())
                 break
         else:
+            # TODO better error message
             print("panic")
             exit(EXIT_ERROR)
         # noinspection PyUnboundLocalVariable
@@ -559,6 +604,8 @@ class Cryptsetup(CommandInterface):
 
     @classmethod
     def _already_decrypted(cls, path: Path, mapper_entry: str) -> bool:
+        """Given a path in `/dev/disk/by-uuid` and the corresponding name under `/dev/mapper`,
+        check if the disk is already decrypted."""
         decrypted_disk_path = cls._get_device_by_uuid_from_status_output(
             cls._status(mapper_entry)
         )
@@ -566,6 +613,7 @@ class Cryptsetup(CommandInterface):
 
 
 class Pool:
+    """Represents an internal pool."""
     def __init__(self, name: str):
         self._name = name
         self._path: ZFSPath = ZFSPath.from_string(name)
@@ -580,12 +628,14 @@ class Pool:
 
 
 class ExternalPool(Pool):
+    """Represents an external pool."""
     def __init__(self, name: str, disk: Disk):
         super().__init__(name)
         self._imported: bool = False
         self._disk = disk
 
     def import_(self) -> None:
+        """Import a pool. Raise an error if the pool was previously imported."""
         if self._imported:
             raise ValueError(Messages.ALREADY_IMPORTED, self.name)
         self._disk.decrypt()
@@ -593,22 +643,21 @@ class ExternalPool(Pool):
         self._imported = True
 
     def export(self) -> None:
+        """Export a previously imported pool. Raise an error if the pool was not previously imported."""
         if not self._imported:
             raise ValueError(Messages.NOT_IMPORTED, self.name)
-        self._export_pool()
+        ZPool.export(self)
         self._disk.encrypt()
         self._imported = False
         print(f"Exported {self.name}. Disk {self._disk.uuid} can be removed.")
 
-    def _export_pool(self) -> None:
-        ZPool.export(self)
-
     def scrub(self, skip_if_recently_scrubbed: bool) -> None:
         print(f"Scrubbing {self._name}")
         ZPool.scrub(self, skip_if_recently_scrubbed=skip_if_recently_scrubbed)
 
     @classmethod
     def find_from_dict(cls, backup_pools: dict[str, UUID]) -> ExternalPool:
+        """Using the dict in config.py, find the backup pool."""
         present_uuids = [item.name for item in DISK_BY_UUID.iterdir()]
         for pool_name, disk_uuid in backup_pools.items():
             if str(disk_uuid) in present_uuids:
@@ -698,6 +747,7 @@ class Snapshot:
         return self._path.snapshot_name
 
     def matches_regex(self, regex: str) -> bool:
+        """Check if the qualified name of the snapshot matches a regular expression."""
         return bool(search(regex, str(self._path)))
 
     def destroy(self) -> None:
@@ -705,7 +755,7 @@ class Snapshot:
 
     @property
     def pool(self) -> str:
-        return self._path.pool
+        return self._path.pool_name
 
     @property
     def dataset(self) -> Dataset:
@@ -721,30 +771,35 @@ class CommandRunner:
 
     @staticmethod
     def _to_list(command: str | list[str]) -> list[str]:
+        """If the command is a str, return it as list. Otherwise return the command."""
         if isinstance(command, str):
             return [command]
         return command
 
     @classmethod
     def run(cls, cmdline: str | list[str]) -> None:
+        """Run the command, wait for it to return."""
         cmdline = cls._to_list(cmdline)
         cls._sanity_check(cmdline)
         call(cmdline)
 
     @classmethod
     def get_output(cls, cmdline: str | list[str]) -> str:
+        """Run the command, return the output of the command."""
         cmdline = cls._to_list(cmdline)
         cls._sanity_check(cmdline)
         return check_output(cmdline).strip().decode()
 
     @classmethod
     def receive_from_stream(cls, cmdline: str | list[str], stream: Popen) -> None:
+        """Run the command, receive data from a pipe to stdin."""
         cmdline = cls._to_list(cmdline)
         cls._sanity_check(cmdline)
         check_output(cmdline, stdin=stream.stdout)
 
     @classmethod
     def send_stream(cls, cmdline: str | list[str]) -> Popen:
+        """Run the command, redirect stdout to a pipe."""
         cmdline = cls._to_list(cmdline)
         cls._sanity_check(cmdline)
         return Popen(cmdline, stdout=PIPE)