p4u1/backup-bot-two
0
0
Fork 0
forked from coop-cloud/backup-bot-two
Code Pull Requests Activity

chore: formatting

Browse Source
This commit is contained in:
Moritz 2023-10-12 12:50:10 +02:00
parent 359140781e
commit 79d19e7ac5
1 changed files with 14 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
backupbot.py
View File

@ -57,7 +57,7 @@ def export_secrets():
if env.endswith('FILE') and not "COMPOSE_FILE" in env: if env.endswith('FILE') and not "COMPOSE_FILE" in env:
logging.debug(f"exported secret: {env}") logging.debug(f"exported secret: {env}")
with open(os.environ[env]) as file: with open(os.environ[env]) as file:
secret = file.read() secret = file.read()
os.environ[env.removesuffix('_FILE')] = secret os.environ[env.removesuffix('_FILE')] = secret
# logging.debug(f"Read secret value: {secret}") # logging.debug(f"Read secret value: {secret}")
@ -166,9 +166,11 @@ def restore(snapshot, target, noninteractive):
if not noninteractive: if not noninteractive:
snapshot_date = datetime.fromisoformat(snapshots[0]['time']) snapshot_date = datetime.fromisoformat(snapshots[0]['time'])
delta = datetime.now(tz=timezone.utc) - snapshot_date delta = datetime.now(tz=timezone.utc) - snapshot_date
print(f"You are going to restore Snapshot {snapshot} of {service_paths} at {target}") print(
f"You are going to restore Snapshot {snapshot} of {service_paths} at {target}")
print(f"This snapshot is {delta} old") print(f"This snapshot is {delta} old")
print(f"THIS COMMAND WILL IRREVERSIBLY OVERWRITES {target}{service_paths.removeprefix('/')}") print(
f"THIS COMMAND WILL IRREVERSIBLY OVERWRITES {target}{service_paths.removeprefix('/')}")
prompt = input("Type YES (uppercase) to continue: ") prompt = input("Type YES (uppercase) to continue: ")
if prompt != 'YES': if prompt != 'YES':
logging.error("Restore aborted") logging.error("Restore aborted")
@ -221,7 +223,7 @@ def list_files(snapshot, path):
logging.error(err_msg) logging.error(err_msg)
exit(1) exit(1)
else: else:
raise error raise error
output = output.replace('}\n{', '}|{') output = output.replace('}\n{', '}|{')
results = list(map(json.loads, output.split('|'))) results = list(map(json.loads, output.split('|')))
return results return results
@ -234,7 +236,8 @@ def list_files(snapshot, path):
@click.option('secrets', '--secrets', '-c', is_flag=True) @click.option('secrets', '--secrets', '-c', is_flag=True)
def download(snapshot, path, volumes, secrets): def download(snapshot, path, volumes, secrets):
if sum(map(bool, [path, volumes, secrets])) != 1: if sum(map(bool, [path, volumes, secrets])) != 1:
logging.error("Please specify exactly one of '--path', '--volumes', '--secrets'") logging.error(
"Please specify exactly one of '--path', '--volumes', '--secrets'")
exit(1) exit(1)
if path: if path:
path = path.removesuffix('/') path = path.removesuffix('/')
@ -254,7 +257,7 @@ def download(snapshot, path, volumes, secrets):
filename = f"/tmp/{SERVICE}.tar" filename = f"/tmp/{SERVICE}.tar"
files = list_files(snapshot, VOLUME_PATH) files = list_files(snapshot, VOLUME_PATH)
for f in files[1:]: for f in files[1:]:
path = f[ 'path' ] path = f['path']
if SERVICE in path and f['type'] == 'dir': if SERVICE in path and f['type'] == 'dir':
content = dump(snapshot, path) content = dump(snapshot, path)
# Concatenate tar files (extract with tar -xi) # Concatenate tar files (extract with tar -xi)
@ -268,7 +271,7 @@ def download(snapshot, path, volumes, secrets):
files = list_files(snapshot, SECRET_PATH) files = list_files(snapshot, SECRET_PATH)
secrets = {} secrets = {}
for f in files[1:]: for f in files[1:]:
path = f[ 'path' ] path = f['path']
if SERVICE in path and f['type'] == 'file': if SERVICE in path and f['type'] == 'file':
secret = dump(snapshot, path).decode() secret = dump(snapshot, path).decode()
secret_name = path.removeprefix(f'{SECRET_PATH}{SERVICE}_') secret_name = path.removeprefix(f'{SECRET_PATH}{SERVICE}_')
@ -277,15 +280,17 @@ def download(snapshot, path, volumes, secrets):
json.dump(secrets, file) json.dump(secrets, file)
print(filename) print(filename)
def dump(snapshot, path): def dump(snapshot, path):
cmd = restic.cat.base_command() + ['dump'] cmd = restic.cat.base_command() + ['dump']
if SERVICE: if SERVICE:
cmd = cmd + ['--tag', SERVICE] cmd = cmd + ['--tag', SERVICE]
cmd = cmd +[snapshot, path] cmd = cmd + [snapshot, path]
logging.debug(f"Dumping {path} from snapshot '{snapshot}'") logging.debug(f"Dumping {path} from snapshot '{snapshot}'")
output = subprocess.run(cmd, capture_output=True) output = subprocess.run(cmd, capture_output=True)
if output.returncode: if output.returncode:
logging.error(f"error while dumping {path} from snapshot '{snapshot}': {output.stderr}") logging.error(
f"error while dumping {path} from snapshot '{snapshot}': {output.stderr}")
exit(1) exit(1)
return output.stdout return output.stdout