"""This module contains the base Spec class.""" 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, ): """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 notifier: notifier object """ 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()