yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

implement backup cli for borgbackup

Browse Source
This commit is contained in:
lassulus
2023-12-04 17:05:37 +01:00
parent c9e640a74d
commit 83dd430d03
10 changed files with 125 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkgs/clan-cli/clan_cli/backups/__init__.py
View File

@@ -1,8 +1,8 @@
# !/usr/bin/env python3
import argparse
from .list import register_list_parser
from .create import register_create_parser
from .list import register_list_parser
from .restore import register_restore_parser

47
pkgs/clan-cli/clan_cli/backups/create.py
View File

@@ -1,34 +1,45 @@
import argparse
import pprint
from pathlib import Path
import json
import subprocess
from typing import Optional
from ..errors import ClanError
from ..machines.machines import Machine
def create_backup(flake_dir: Path, machine: Optional[str] = None, provider: Optional[str] = None) -> None:
if machine is None:
# TODO get all machines here
machines = [ "machine1", "machine2" ]
else:
machines = [ machine ]
def create_backup(machine: Machine, provider: Optional[str] = None) -> None:
backup_scripts = json.loads(
machine.eval_nix(f"nixosConfigurations.{machine.name}.config.clanCore.backups")
)
if provider is None:
# TODO get all providers here
providers = [ "provider1", "provider2" ]
for provider in backup_scripts["providers"]:
proc = subprocess.run(
["bash", "-c", backup_scripts["providers"][provider]["start"]],
)
if proc.returncode != 0:
raise ClanError("failed to start backup")
else:
print("successfully started backup")
else:
providers = [ provider ]
print("would create backups for machines: ", machines, " with providers: ", providers)
if provider not in backup_scripts["providers"]:
raise ClanError(f"provider {provider} not found")
proc = subprocess.run(
["bash", "-c", backup_scripts["providers"][provider]["start"]],
)
if proc.returncode != 0:
raise ClanError("failed to start backup")
else:
print("successfully started backup")
def create_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
create_backup(Path(args.flake), machine=args.machine, provider=args.provider)
machine = Machine(name=args.machine, flake_dir=args.flake)
create_backup(machine=machine, provider=args.provider)
def register_create_parser(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--machine", type=str, help="machine in the flake to create backups of")
parser.add_argument(
"machine", type=str, help="machine in the flake to create backups of"
)
parser.add_argument("--provider", type=str, help="backup provider to use")
parser.set_defaults(func=create_command)

33
pkgs/clan-cli/clan_cli/backups/list.py
View File

@@ -6,25 +6,27 @@ from typing import Optional
from ..errors import ClanError
def list_backups(flake_dir: Path, machine: Optional[str] = None, provider: Optional[str] = None) -> dict[str, dict[str, list[str]]]:
def list_backups(
flake_dir: Path, machine: str, provider: Optional[str] = None
) -> dict[str, dict[str, list[dict[str, str]]]]:
dummy_data = {
"testhostname": {
"borgbackup": [
"2021-01-01T00:00:00Z",
"2022-01-01T00:00:00Z",
"2023-01-01T00:00:00Z",
{"date": "2021-01-01T00:00:00Z", "id": "1"},
{"date": "2022-01-01T00:00:00Z", "id": "2"},
{"date": "2023-01-01T00:00:00Z", "id": "3"},
],
"restic" : [
"2021-01-01T00:00:00Z",
"2022-01-01T00:00:00Z",
"2023-01-01T00:00:00Z",
"restic": [
{"date": "2021-01-01T00:00:00Z", "id": "1"},
{"date": "2022-01-01T00:00:00Z", "id": "2"},
{"date": "2023-01-01T00:00:00Z", "id": "3"},
],
},
"another host": {
"borgbackup": [
"2021-01-01T00:00:00Z",
"2022-01-01T00:00:00Z",
"2023-01-01T00:00:00Z",
{"date": "2021-01-01T00:00:00Z", "id": "1"},
{"date": "2022-01-01T00:00:00Z", "id": "2"},
{"date": "2023-01-01T00:00:00Z", "id": "3"},
],
},
}
@@ -41,16 +43,21 @@ def list_backups(flake_dir: Path, machine: Optional[str] = None, provider: Optio
else:
return {machine: dummy_data[machine]}
def list_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
backups = list_backups(Path(args.flake), machine=args.machine, provider=args.provider)
backups = list_backups(
Path(args.flake), machine=args.machine, provider=args.provider
)
if len(backups) > 0:
pp = pprint.PrettyPrinter(depth=4)
pp.pprint(backups)
def register_list_parser(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--machine", type=str, help="machine in the flake to show backups of")
parser.add_argument(
"machine", type=str, help="machine in the flake to show backups of"
)
parser.add_argument("--provider", type=str, help="backup provider to filter by")
parser.set_defaults(func=list_command)

27
pkgs/clan-cli/clan_cli/backups/restore.py
View File

@@ -1,26 +1,41 @@
import argparse
import pprint
from pathlib import Path
from typing import Optional
from ..errors import ClanError
def restore_backup(flake_dir: Path, machine: str, provider: str, backup_id: str, service: Optional[str] = None) -> None:
def restore_backup(
flake_dir: Path,
machine: str,
provider: str,
backup_id: str,
service: Optional[str] = None,
) -> None:
if service is None:
print("would restore backup", machine, provider, backup_id)
print("would restore backup", machine, provider, backup_id)
else:
print("would restore backup", machine, provider, backup_id, "of service:", service)
print(
"would restore backup", machine, provider, backup_id, "of service:", service
)
def restore_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
restore_backup(Path(args.flake), machine=args.machine, provider=args.provider, backup_id=args.backup_id, service=args.service)
restore_backup(
Path(args.flake),
machine=args.machine,
provider=args.provider,
backup_id=args.backup_id,
service=args.service,
)
def register_restore_parser(parser: argparse.ArgumentParser) -> None:
parser.add_argument("machine", type=str, help="machine in the flake to create backups of")
parser.add_argument(
"machine", type=str, help="machine in the flake to create backups of"
)
parser.add_argument("provider", type=str, help="backup provider to use")
parser.add_argument("backup_id", type=str, help="id of the backup to restore")
parser.add_argument("--service", type=str, help="name of the service to restore")