backup-tool/app/specs/spec.py

"""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()