2023-09-30 10:10:10 -07:00

249 lines
8.5 KiB

import os
import argparse
import subprocess
import traceback
import time
from resticbackup import __version__
from resticbackup.types import ClientConfig, load_config, load_base_config, list_configs
from resticbackup.lib import fprint, ExecWrapper, init_ok, die, pdie, checkp, runp, get_retention_args, \
get_newest_snapshot, update_statefile
def cmd_backup(args, parser):
config = load_config(
# run the backup
backoff(run_backup,, config)
# run retention
if backoff(run_retention,, config):
# run prune (optional)
if args.prune:
backoff(run_prune, config)
# update statefile if configured
if config.statefile:
path = config.statefile.format(
update_statefile(path, True)
if config.statefile:
path = config.statefile.format(
update_statefile(path, False)
def run_backup(name, config):
# these tags will be added to the backup.
# They need to match retention_tags in run_rentention, below
backup_tags = {
"name": name,
# perform pre/post-exec
# post-exec is executed if the backup fails OR if any pre-exec commands fail
# pre-exec commands failing will stop the backup from being executed
with ExecWrapper(post=config.backup.backup_postexec):
with ExecWrapper(pre=config.backup.backup_preexec):
proc =["init"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout, _ = proc.communicate()
stdout = stdout.decode()
if not init_ok(stdout):
raise Exception(stdout)
# run backup
backup_cmd = ["backup", "./"]
for k, v in backup_tags.items():
backup_cmd.extend(["--tag", "{}={}".format(k, v)])
for entry in config.backup.exclude:
if entry.startswith("/"):
entry = os.path.join(config.backup.path, entry[1:])
backup_cmd.extend(["--exclude", entry])
def backoff(func, *args, **kwargs):
run the retention function, but retry if it fails. Restic must lock the entire repo during this process, so it is
possible that it fails due to already being in use by another client
backoff = [1, 2, 3, 5, 10, 15, 30, 60]
backoffs = 0
while True:
return func(*args, **kwargs)
if backoffs >= len(backoff):
delay = backoff[backoffs]
backoffs += 1
fprint("...caught, and will be retried in {}m".format(delay))
time.sleep(delay * 60)
def run_retention(name, config):
run retention, which untags old backups. Returns true if retention is configured and was executed, false otherwise
# the tags in retention_tags will be used when pruning backups.
# They need to match backup_tags in run_backup, above
retention_tags = {
"name": name,
# perform retention
retention_args = get_retention_args(config.backup.schedule)
if retention_args:
forget_cmd = ["forget", "--group-by", "tags"] + retention_args
for k, v in retention_tags.items():
forget_cmd.extend(["--tag", "{}={}".format(k, v)])
return True
return False
def run_prune(config):
# perform junk deletion
def cmd_restore(args, parser):
config = load_config(
if os.listdir(config.backup.path):
return die("output dir has files present - refusing to restore")
except FileNotFoundError:
return die("output dir '{}' does not exist".format(config.backup.path))
snapshot = get_newest_snapshot(, config)
if not snapshot:
if args.no_snapshots_ok:
fprint("no snapshots found, but exiting OK")
return die("no snapshots found")
with ExecWrapper(post=config.backup.restore_postexec):
with ExecWrapper(pre=config.backup.restore_preexec):
return pdie(["restore", snapshot["id"], "-t", "./"]))
def cmd_retrieve(args, parser):
# this is the same as cmd_restore except for how the config is formed and no execs
config = ClientConfig.load(load_base_config(), {"repo": args.repo, "path": args.outdir})
if os.listdir(config.backup.path):
return die("output dir has files present - refusing to restore")
except FileNotFoundError:
return die("output dir does not exist")
snapshot = get_newest_snapshot(, config)
if not snapshot:
return die("no snapshots found")
return pdie(["restore", snapshot["id"], "-t", args.outdir]))
def cmd_list(args, parser):
config = load_config(
cmd = ["snapshots", "--group-by", "tags"]
retention_tags = {
for k, v in retention_tags.items():
cmd.extend(["--tag", "{}={}".format(k, v)])
return pdie(
for entry in list_configs():
def cmd_exec(args, parser):
def cmd_prune(args, parser):
if is None:
names = list_configs()
names = []
configs = [load_config(n) for n in names]
# we only need to prune each repo once, so use a dict to group configs by repo address
configs = {c.repo: c for c in configs}
for config in configs.values():
fprint("pruning:", config.repo)
def cmd_version(args, parser):
def main():
parser = argparse.ArgumentParser(description="restic wrapper")
# parser.add_argument('-n', '--no-exec', action='store_true', help='do not run pre/post-exec commands') # TODO
# parser.add_argument('-b', '--no-pre-exec', action='store_true', help='do not run pre-exec commands') # TODO
# parser.add_argument('-m', '--no-post-exec', action='store_true', help='do not run post-exec commands') # TODO
sp_action = parser.add_subparsers(dest="action", help="action to take")
p_backup = sp_action.add_parser("backup", help="take a backup")
p_backup.add_argument("name", help="name of backup to execute")
p_backup.add_argument('--prune', action='store_true', help='run the prune operation after backup')
# p_backup.add_argument("-c", "--print", action="store_true", help="just print the restic commands instead") # TODO
# p_backup.add_argument("-e", "--env", action="store_true", help="just print the environment variables instead") # TODO
p_restore = sp_action.add_parser("restore", help="restore a backup")
p_restore.add_argument("name", help="name of backup to restore")
# p_restore.add_argument("snapshot", nargs="?", help="id of snapshot to restore") # TODO
p_restore.add_argument('--no-snapshots-ok', action='store_true',
help='don\'t return an error if no snapshots are available to be restored')
p_exec = sp_action.add_parser("exec", help="execute a restic command")
p_exec.add_argument("name", help="name of backup to configure restic for")
p_exec.add_argument("args", nargs="+", help="command arguments")
p_list = sp_action.add_parser("list", help="list configs or snapshots")
p_list.add_argument("name", nargs="?", help="name of config to list snapshots for")
p_retrieve = sp_action.add_parser("retrieve", help="download arbitrary snapshots")
p_retrieve.add_argument("repo", help="repo path")
p_retrieve.add_argument("name", help="backup name")
p_retrieve.add_argument("outdir", help="file path to write to")
p_prune = sp_action.add_parser("prune", help="run the prune function")
p_prune.add_argument("name", nargs="?", help="name of backup to run prune in repo for. Omit for all repos")
p_version = sp_action.add_parser("version", help="display program version")
args = parser.parse_args()
args.func(args, parser)
if __name__ == '__main__':