from __future__ import annotations from pathlib import Path from typing import Union, Dict import skeleton import os class Spec: """ Base class for all other spec types. """ __SKEL = { "name": None, "destination": None, "limit": None, "notifier": None, "extension": "tar.gz", } def __init__( self, name: str, destination: Union[Path, str], limit: int, extension: str, notifier=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 self.extension = extension 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 = cls.__SKEL for default in defaults: skel = skeleton.combine(defaults, skel) # Then, combine actual values with new skeleton obj = skeleton.combine(obj, skel) return cls(name, **obj)