resticbackup/resticbackup/types.py

97 lines
2.9 KiB
Python

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,
}
@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)