backup-tool/app/specs/spec.py

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