123 lines
3.4 KiB
Python
123 lines
3.4 KiB
Python
"""This module contains the base Spec class."""
|
|
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,
|
|
):
|
|
"""Initialize a new Spec object.
|
|
|
|
This initializer usually gets called by a subclass's init instead of
|
|
directly.
|
|
|
|
Args:
|
|
name: name of the spec
|
|
destination: directory where the backups shall reside
|
|
limit: max amount of backups
|
|
extension: file extension of the backup files
|
|
notify: notifier object to send IFTT notifications
|
|
"""
|
|
self.name = name
|
|
self.destination = Path(destination)
|
|
|
|
# Create destination if non-existent
|
|
try:
|
|
self.destination.mkdir(parents=True, exist_ok=True)
|
|
|
|
# TODO just make this some checks in advance
|
|
except FileExistsError:
|
|
raise NotADirectoryError(
|
|
"{} already exists, but 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: "Spec") -> Dict:
|
|
"""Return the skeleton for the given class.
|
|
|
|
It works by inspecting the inheritance tree and merging the skeleton
|
|
for each of the parents.
|
|
|
|
Args:
|
|
cls: the class to get the skeleton for
|
|
|
|
Returns:
|
|
a dictionary containing the skeleton
|
|
"""
|
|
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):
|
|
"""Create a new backup.
|
|
|
|
This function should be implemented by the subclasses.
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
def restore(self):
|
|
"""Restore a given backup (NOT IMPLEMENTED).
|
|
|
|
This function should be implemented by the subclasses.
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
@classmethod
|
|
def from_dict(cls, name, obj: Dict, defaults: Dict) -> "Spec":
|
|
"""Create the class given a dictionary (e.g. from a config)."""
|
|
# 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):
|
|
"""Export the class as a dictionary.
|
|
|
|
This function should be imnplemented by the subclasses.
|
|
"""
|
|
raise NotImplementedError()
|