home-backup/home_backup/user_service/backups.py

import asyncio
from . import remotes
from .. import utils


class Backup():
    name:str
    btype:str
    periode:int
    blocked:set

    to_backup:list
    remote:remotes.Remote

    def __init__(self, name:str, btype:str="borgbackup", periode:int=None, blocked:list=[], to_backup:list=[], remote:remotes.Remote=None):
        # Check args
        utils.valid_name_check(name)
        if periode is not None and not isinstance(periode, int):
            raise TypeError("Periode have to be an integer or null.")
        if periode is not None and periode < 0:
            raise ValueError("periode can't be negetive.")
        blocked = set(blocked)
        for i in blocked:
            utils.valid_name_check(i)
        to_backup = set(to_backup)
        for i in to_backup:
            if not isinstance(i, str):
                raise TypeError("to_backup have to be a string.")
            if ":" in i:
                raise ValueError(": isn't allowed a char.")
        if not isinstance(btype, str):
            raise TypeError("btype has to be a string.")
        if btype not in ("borgbackup",):
            raise ValueError("%s is an unknown backup tool." % repr(btype))
        if remote is not None and not isinstance(remote, remotes.Remote):
            raise TypeError("remote has to be an remote object.")

        # Check type
        if btype == "borgbackup":
            if remote is None:
                raise ValueError("remote is required for borg backup.")
            if remote.rtype != "borgbackup":
                raise ValueError("remote has to be an borg backup target.")
        else:
            raise NotImplementedError("%s isn't an implemented type." % btype)

        # Set values
        self.name = name
        self.periode = periode
        self.blocked = blocked
        self.to_backup = to_backup
        self.btype = btype
        self.remote = remote
    
    def get_next_scedule(self, latest, zero):
        if self.periode is not None:
            tmp = (latest - zero) // self.periode
            return zero + self.periode * (tmp + 1)
        else:
            raise NotImplementedError("No implemented types.")

    def dump_config(self):
        result = {}
        result["type"] = self.btype
        if self.blocked:
            result["blocked"] = ",".join(self.blocked)
        if self.periode is not None:
            result["periode"] = str(self.periode)
        if self.to_backup:
            result["to_backup"] = ":".join(self.to_backup)
        if self.remote:
            result["remote"] = self.remote.name
        return result

    @staticmethod
    def load_backup(name:str, conf, remotes):
        # Load informations
        conf = dict(conf.items())

        btype = conf["type"]
        del conf["type"]

        periode = None
        if "periode" in conf:
            periode = int(conf["periode"])
            del conf["periode"]

        blocked = []
        if "blocked" in conf:
            blocked = conf["blocked"].split(",")
            del conf["blocked"]
        
        to_backup = []
        if "to_backup" in conf:
            to_backup = conf["to_backup"].split(":")
            del conf["to_backup"]
        
        remote = None
        if "remote" in conf:
            remote = conf["remote"]
            if remote not in remotes.keys():
                raise ValueError("Remote %s doesn't exists but is required." & remote)
            remote = remotes[remote]
            del conf["remote"]

        # Generate backup
        utils.check_empty_data_dict(conf)

        return Backup(name=name, btype=btype, periode=periode, blocked=blocked, to_backup=to_backup, remote=remote)
    
    async def run_backup(self, subvolumes:list):
        print("Subvolumes: %s" % repr(subvolumes))

# RPC implementations
async def add_backup(data):
    # Import config
    from . import config

    # Load base values
    name = data["name"]
    del data["name"]

    info = data["info"]
    if not isinstance(info, dict):
        raise TypeError("info has to be an object.")
    del data["info"]

    btype = data["type"]
    del data["type"]
    utils.check_empty_data_dict(data)

    # Load info
    periode = None
    if "periode" in info:
        periode = int(info["periode"])
        del info["periode"]

    blocked = []
    if "blocked" in info:
        blocked = info["blocked"]
        del info["blocked"]

    to_backup = []
    if "to_backup" in info:
        to_backup = info["to_backup"]
        del info["to_backup"]

    remote = None
    if "remote" in info:
        remote = info["remote"]
        del info["remote"]

    utils.check_empty_data_dict(info)

    # Add backup
    async with config.config_lock:
        # Check if backup exists
        for _ in filter(lambda x: x.name == name, config.backups):
            return {"status": "fail-already-exists"}

        # Search for remote
        if remote is not None:
            if remote not in config.remotes:
                return {"status": "fail-remote-missing"}
            remote = config.remotes[remote]

        # Add backup
        backup = Backup(name=name, btype=btype, periode=periode, blocked=blocked, to_backup=to_backup, remote=remote)
        config.backups.append(backup)
        await config.save_config()
    return {"status": "success"}