pobsync/src/pobsync_backend/management/commands/run_pobsync_backup.py

from __future__ import annotations

from pathlib import Path
from typing import Any

from django.conf import settings
from django.core.management.base import BaseCommand, CommandError

from pobsync.paths import PobsyncPaths
from pobsync_backend.backup_runner import execute_backup_run
from pobsync_backend.models import BackupRun, HostConfig


class Command(BaseCommand):
    help = "Run a pobsync backup and record the result in Django."

    def add_arguments(self, parser) -> None:
        parser.add_argument("host", help="Host to back up")
        parser.add_argument("--prefix", default=settings.POBSYNC_HOME, help="Pobsync home directory")
        parser.add_argument("--dry-run", action="store_true", help="Run rsync --dry-run")
        parser.add_argument("--verbose-rsync", action="store_true", help="Write itemized rsync output to the run log")
        parser.add_argument("--prune", action="store_true", help="Apply retention after a successful run")
        parser.add_argument("--prune-max-delete", type=int, default=10)
        parser.add_argument("--prune-protect-bases", action="store_true")
        parser.add_argument("--manual", action="store_true", help="Record the run as manual instead of scheduled")

    def handle(self, *args: Any, **options: Any) -> None:
        host_name = options["host"]
        paths = PobsyncPaths(home=Path(options["prefix"]))
        try:
            host = HostConfig.objects.get(host=host_name, enabled=True)
        except HostConfig.DoesNotExist as exc:
            raise CommandError(f"Missing enabled HostConfig {host_name!r}") from exc

        run = BackupRun.objects.create(
            host=host,
            run_type=BackupRun.RunType.MANUAL if options["manual"] else BackupRun.RunType.SCHEDULED,
            status=BackupRun.Status.RUNNING,
            result={
                "requested": {
                    "dry_run": bool(options["dry_run"]),
                    "verbose_output": bool(options["dry_run"] or options["verbose_rsync"]),
                    "prune": bool(options["prune"]),
                    "prune_max_delete": int(options["prune_max_delete"]),
                    "prune_protect_bases": bool(options["prune_protect_bases"]),
                }
            },
        )
        execute_backup_run(
            run=run,
            prefix=paths.home,
            dry_run=bool(options["dry_run"]),
            verbose_output=bool(options["dry_run"] or options["verbose_rsync"]),
            prune=bool(options["prune"]),
            prune_max_delete=int(options["prune_max_delete"]),
            prune_protect_bases=bool(options["prune_protect_bases"]),
        )
        run.refresh_from_db()

        if run.status == BackupRun.Status.SUCCESS:
            self.stdout.write(self.style.SUCCESS(f"Backup completed for {host.host}."))
            return

        raise CommandError(f"Backup failed for {host.host}; run id={run.id}")