timeshifter/zfs-backup
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

chg: refactoring

Browse Source
This commit is contained in:
timeshifter
2021-12-02 14:44:04 +01:00
parent 9d89463219
commit d16b972021
1 changed files with 17 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zfs-backup.py
+16 -12
View File
@@ -96,16 +96,13 @@ class Manager:
@staticmethod @staticmethod
def _find_snapshots_with_backup_tag(dataset: Dataset) -> list[Snapshot]: def _find_snapshots_with_backup_tag(dataset: Dataset) -> list[Snapshot]:
regex = Manager._get_regex_matching_backup_tags() regex = _get_regex_matching_snapshots_with_tags(
[config.snapshot_tag, config.snapshot_interval]
)
return [ return [
snapshot for snapshot in dataset.snapshots if snapshot.matches_regex(regex) snapshot for snapshot in dataset.snapshots if snapshot.matches_regex(regex)
] ]
@staticmethod
# TODO make function
def _get_regex_matching_backup_tags():
return "@" + ".*".join([config.snapshot_tag, config.snapshot_interval])
@staticmethod @staticmethod
def _search_matching_dataset_in_remote_pool(dataset: Dataset, pool: Pool): def _search_matching_dataset_in_remote_pool(dataset: Dataset, pool: Pool):
dataset_to_search = dataset.replace_pool(pool.name) dataset_to_search = dataset.replace_pool(pool.name)
@@ -125,6 +122,10 @@ class Manager:
return common_snapshots[-1] return common_snapshots[-1]
def _get_regex_matching_snapshots_with_tags(tags: list[str]):
return "@" + ".*".join(tags)
class ZFSPath: class ZFSPath:
def __init__(self, elements: list[str], snapname: Optional[str] = None): def __init__(self, elements: list[str], snapname: Optional[str] = None):
self._elements = elements self._elements = elements
@@ -433,16 +434,19 @@ class Dataset:
self._clean_old_snapshots_for_interval(interval, number) self._clean_old_snapshots_for_interval(interval, number)
def _clean_old_snapshots_for_interval(self, interval: str, max_count: int): def _clean_old_snapshots_for_interval(self, interval: str, max_count: int):
# TODO this is somewhat similar to the regex method in Manager regex = _get_regex_matching_snapshots_with_tags(
regex = ".*".join(["", config.snapshot_tag, interval, ""]) [config.snapshot_tag, interval]
)
snapshots = [ snapshots = [
snapshot for snapshot in self.snapshots if snapshot.matches_regex(regex) snapshot for snapshot in self.snapshots if snapshot.matches_regex(regex)
] ]
# TODO function _get_n_oldest_snapshots old_snapshots = self._get_n_oldest_snapshots(snapshots, max_count)
if len(snapshots) > max_count: for snapshot in old_snapshots:
snapshot.destroy()
def _get_n_oldest_snapshots(self, snapshots: list[Snapshot], count: int) -> list[Snapshot]:
snapshots.sort() snapshots.sort()
for snap in snapshots[:-max_count]: return snapshots[:count]
snap.destroy()
class Snapshot: class Snapshot: