backup-tool/app/specs/spec.py

99 lines
2.4 KiB
Python

from __future__ import annotations
from pathlib import Path
from typing import Union, Dict
import skeleton
import os
from notifier import Notifier
import inspect
class Spec:
"""
Base class for all other spec types.
"""
_SKEL = {
"destination": None,
"limit": None,
"notify": {
"title": "Backup Notification",
"events": ["backup_sucess"],
"endpoint": None,
"api_key": "",
},
"extension": "tar.gz",
}
def __init__(
self,
name: str,
destination: Union[Path, str],
limit: int,
extension: str,
notify=None,
):
"""
Args:
name: name of the spec
destination: directory where the backups shall reside
limit: max amount of backups
notifier: notifier object
"""
self.name = name
self.destination = (
destination if type(destination) == Path else Path(destination)
)
# Check existence of destination folder
if not self.destination.exists() or not self.destination.is_dir():
raise NotADirectoryError(
"{} doesn't exist or isn't a directory.".format(
self.destination
)
)
self.limit = limit
self.notifier = Notifier(*notify) if notify else None
self.extension = extension
@classmethod
def skeleton(cls):
return skeleton.merge(
*[val._SKEL for val in reversed(inspect.getmro(cls)[:-1])]
)
def remove_backups(self):
"""
Remove all backups exceeding the limit
"""
files = sorted(
self.destination.glob("*." + self.extension),
key=os.path.getmtime,
reverse=True,
)
if len(files) >= self.limit:
for path in files[self.limit - 1 :]:
path.unlink()
def backup(self):
raise NotImplementedError()
def restore(self):
raise NotImplementedError()
@classmethod
def from_dict(cls, name, obj: Dict, defaults: Dict) -> Spec:
# Combine defaults with skeleton, creating new skeleton
skel = skeleton.merge(cls.skeleton(), defaults)
# Then, combine actual values with new skeleton
obj = skeleton.merge_with_skeleton(obj, skel)
return cls(name, **obj)
def to_dict(self):
raise NotImplementedError()