fix pyright lints
This commit is contained in:
+8
-8
@@ -38,8 +38,6 @@ def main():
|
|||||||
class Manager:
|
class Manager:
|
||||||
"""Controls the overall backup process."""
|
"""Controls the overall backup process."""
|
||||||
|
|
||||||
_map_type: dict[Dataset, Dataset | None]
|
|
||||||
|
|
||||||
def __init__(self, local_pool: Pool, external_pool: ExternalPool):
|
def __init__(self, local_pool: Pool, external_pool: ExternalPool):
|
||||||
self._local_pool = local_pool
|
self._local_pool = local_pool
|
||||||
self._external_pool = external_pool
|
self._external_pool = external_pool
|
||||||
@@ -70,11 +68,11 @@ class Manager:
|
|||||||
continue
|
continue
|
||||||
ZFS.receive(stream, self._external_pool)
|
ZFS.receive(stream, self._external_pool)
|
||||||
|
|
||||||
def _get_user_filtered_dataset_backup_map(self) -> _map_type:
|
def _get_user_filtered_dataset_backup_map(self) -> dict[Dataset, Dataset | None]:
|
||||||
"""Get the dataset map, let user confirm new datasets."""
|
"""Get the dataset map, let user confirm new datasets."""
|
||||||
return self._user_confirm_new_datasets(self._get_raw_dataset_map())
|
return self._user_confirm_new_datasets(self._get_raw_dataset_map())
|
||||||
|
|
||||||
def _user_confirm_new_datasets(self, dataset_map: _map_type) -> _map_type:
|
def _user_confirm_new_datasets(self, dataset_map: dict[Dataset, Dataset | None]) -> dict[Dataset, Dataset | None]:
|
||||||
"""For datasets that exist only locally, ask the user if he wants to back it up.
|
"""For datasets that exist only locally, ask the user if he wants to back it up.
|
||||||
|
|
||||||
If he doesn't, remove it from the dict.
|
If he doesn't, remove it from the dict.
|
||||||
@@ -85,7 +83,7 @@ class Manager:
|
|||||||
if remote_dataset is not None or self._ask_user_confirmation(local_dataset)
|
if remote_dataset is not None or self._ask_user_confirmation(local_dataset)
|
||||||
}
|
}
|
||||||
|
|
||||||
def _get_raw_dataset_map(self) -> _map_type:
|
def _get_raw_dataset_map(self) -> dict[Dataset, Dataset | None]:
|
||||||
"""Return map from local to remote datasets.
|
"""Return map from local to remote datasets.
|
||||||
|
|
||||||
If the local dataset has no remote match, set the remote dataset to None.
|
If the local dataset has no remote match, set the remote dataset to None.
|
||||||
@@ -171,7 +169,7 @@ class Manager:
|
|||||||
return common_snapshots[-1]
|
return common_snapshots[-1]
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _ask_user_confirmation(dataset: Dataset) -> bool:
|
def _ask_user_confirmation(dataset: Dataset) -> bool: # type: ignore (false positive)
|
||||||
msg = f"Dataset {dataset.qualified_name} does not exist in backup pool. Backup? [y/n] "
|
msg = f"Dataset {dataset.qualified_name} does not exist in backup pool. Backup? [y/n] "
|
||||||
reply = ""
|
reply = ""
|
||||||
while not reply:
|
while not reply:
|
||||||
@@ -256,6 +254,7 @@ class ZFSPath:
|
|||||||
"""Return the `zfs` path, but without the pool name."""
|
"""Return the `zfs` path, but without the pool name."""
|
||||||
name = "/".join(self._elements[1:])
|
name = "/".join(self._elements[1:])
|
||||||
if self.is_snapshot:
|
if self.is_snapshot:
|
||||||
|
assert self.snapshot_name
|
||||||
name = "@".join([name, self.snapshot_name])
|
name = "@".join([name, self.snapshot_name])
|
||||||
return name
|
return name
|
||||||
|
|
||||||
@@ -751,8 +750,8 @@ class Snapshot:
|
|||||||
def from_string(cls, qualified_name: str) -> Snapshot:
|
def from_string(cls, qualified_name: str) -> Snapshot:
|
||||||
return cls(ZFSPath.from_string(qualified_name))
|
return cls(ZFSPath.from_string(qualified_name))
|
||||||
|
|
||||||
def __lt__(self, other: Snapshot) -> bool:
|
def __gt__(self, other: Snapshot) -> bool:
|
||||||
return self.snapshot_name < other.snapshot_name
|
return self.snapshot_name > other.snapshot_name
|
||||||
|
|
||||||
def __repr__(self):
|
def __repr__(self):
|
||||||
return str(self._path)
|
return str(self._path)
|
||||||
@@ -765,6 +764,7 @@ class Snapshot:
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def snapshot_name(self) -> str:
|
def snapshot_name(self) -> str:
|
||||||
|
assert self._path.snapshot_name
|
||||||
return self._path.snapshot_name
|
return self._path.snapshot_name
|
||||||
|
|
||||||
def matches_regex(self, regex: str) -> bool:
|
def matches_regex(self, regex: str) -> bool:
|
||||||
|
|||||||
Reference in New Issue
Block a user