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

from __future__ import annotations

from typing import Any

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

from pobsync.config.retention import parse_retention
from pobsync.config.defaults import DEFAULT_EXCLUDES, DEFAULT_RSYNC_ARGS
from pobsync.util import is_absolute_non_root
from pobsync_backend.models import GlobalConfig


class Command(BaseCommand):
    help = "Create or update the SQL-backed global pobsync configuration."

    def add_arguments(self, parser) -> None:
        parser.add_argument("--name", default="default")
        parser.add_argument("--backup-root", required=True)
        parser.add_argument("--ssh-user", default="root")
        parser.add_argument("--ssh-port", type=int, default=22)
        parser.add_argument("--source-root", default="/")
        parser.add_argument("--retention", default="daily=14,weekly=8,monthly=12,yearly=0")
        parser.add_argument("--force", action="store_true", help="Update existing config")

    def handle(self, *args: Any, **options: Any) -> None:
        backup_root = options["backup_root"]
        if not is_absolute_non_root(backup_root):
            raise CommandError("--backup-root must be an absolute path and must not be '/'")

        retention = parse_retention(options["retention"])
        defaults = {
            "backup_root": backup_root,
            "ssh_user": options["ssh_user"],
            "ssh_port": options["ssh_port"],
            "ssh_options": ["-oBatchMode=yes", "-oStrictHostKeyChecking=accept-new"],
            "rsync_binary": "rsync",
            "rsync_args": DEFAULT_RSYNC_ARGS,
            "rsync_extra_args": [],
            "rsync_timeout_seconds": 0,
            "rsync_bwlimit_kbps": 0,
            "default_source_root": options["source_root"],
            "default_destination_subdir": "",
            "excludes_default": DEFAULT_EXCLUDES,
            "retention_daily": retention["daily"],
            "retention_weekly": retention["weekly"],
            "retention_monthly": retention["monthly"],
            "retention_yearly": retention["yearly"],
        }

        if GlobalConfig.objects.filter(name=options["name"]).exists() and not options["force"]:
            raise CommandError(f"GlobalConfig {options['name']!r} already exists; use --force to update")

        _obj, created = GlobalConfig.objects.update_or_create(name=options["name"], defaults=defaults)
        action = "Created" if created else "Updated"
        self.stdout.write(self.style.SUCCESS(f"{action} GlobalConfig {options['name']!r}."))