resticbackup/resticbackup/types.py

import os
import json
import subprocess
from urllib.parse import urlparse
from dataclasses import dataclass, field
from typing import List

from resticbackup import CFG_DIR, RESTIC_BIN


def list_configs():
    config_names = []
    for fname in os.listdir(CFG_DIR):
        name = fname.split(".")[0]
        if name == "main":
            continue
        config_names.append(name)
    config_names.sort()
    return config_names


def load_base_config():
    main_path = os.path.join(CFG_DIR, "main.json")
    with open(main_path) as f:
        return json.load(f)


def load_config(name):
    # load configs from /etc/resticbackup.d/
    # we have special handling for /etc/resticbackup.d/main.json
    # return (main_config, dict(config_name=>config))
    cfg_path = os.path.join(CFG_DIR, "{}.json".format(name))

    with open(cfg_path) as f:
        backup_config = json.load(f)

    return ClientConfig.load(load_base_config(), backup_config)


@dataclass
class BackupConfig:
    path: str
    repo: str
    schedule: dict = field(default_factory=dict)
    exclude: List[str] = field(default_factory=list)
    backup_preexec: List[str] = field(default_factory=list)
    backup_postexec: List[str] = field(default_factory=list)
    restore_preexec: List[str] = field(default_factory=list)
    restore_postexec: List[str] = field(default_factory=list)

    @staticmethod
    def load(data: dict) -> 'BackupConfig':
        return BackupConfig(
            path=data['path'],
            repo=data['repo'],
            schedule=data.get('schedule', {}),
            exclude=data.get('exclude', []),
            backup_preexec=data.get('backup_preexec', []),
            backup_postexec=data.get('backup_postexec', []),
            restore_preexec=data.get('restore_preexec', []),
            restore_postexec=data.get('restore_postexec', []),
        )


@dataclass
class ClientConfig:
    server_type: str
    uri: str
    secret: str
    backup: BackupConfig

    @property
    def repo(self):  # port ignored
        return "{}:{}://{}/{}".format(self.server_type, self.uri.scheme, self.uri.hostname, self.backup.repo)

    @property
    def env(self):
        return {
            "AWS_ACCESS_KEY_ID": self.uri.username,
            "AWS_SECRET_ACCESS_KEY": self.uri.password,
            "RESTIC_PASSWORD": self.secret,
            "RESTIC_REPOSITORY": self.repo,
            "GOGC": "20",
        }

    @staticmethod
    def load(main, backup) -> "ClientConfig":
        backup = BackupConfig.load(backup)
        server_type, server = main["server"].split(":", 1)
        if server_type != "s3":
            raise Exception("unsupported server type: {}".format(server_type))
        return ClientConfig(server_type=server_type, uri=urlparse(server), secret=main["secret"], backup=backup)

    def run(self, args, **kwargs):
        env = dict(os.environ)
        env.update(**self.env)
        return subprocess.Popen([RESTIC_BIN] + args, env=env, **kwargs)