backup-tool/app/specs/spec.py

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

        # Create destination if non-existent
        try:
            self.destination.mkdir(parents=True, exist_ok=True)

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