Add WebDAV backup push to backup_database

PROJECT.org

@@ -649,3 +649,22 @@ scrobbled are identical. Run it dry (default) to report duplicates, or with
 There should be a tool that can look at the track associated with a scrobble and
 the actual raw_data from mopidy or jellyfin, and flag all tracks where the
 artist or track name or both differ.
+** DONE [#B] Push database backups to WebDAV from the backup management command :backups:webdav:
+:PROPERTIES:
+:ID:       01d35a28-aa62-464f-9730-80bd025c0381
+:END:
+
+*** Description
+
+The `backup_database` management command / celery task now uploads the nightly
+pg_dump + gzip archive to a WebDAV destination in addition to the existing local
+and optional SSH (scp) targets.
+
+Configured via server-level env vars:
+- `VROBBLER_DB_BACKUP_WEBDAV_URL`, `VROBBLER_DB_BACKUP_WEBDAV_USER`,
+  `VROBBLER_DB_BACKUP_WEBDAV_PASS`
+- `VROBBLER_DB_BACKUP_WEBDAV_PATH` (default `var/vrobbler-backups/`)
+
+Old backups on WebDAV are pruned with the same retention policy as the SSH
+remote (`_retention_files_to_delete`), and the local copy in
+`DB_BACKUP_LOCAL_DIR` is kept.

envrc.sample

@@ -2,3 +2,7 @@ export ENV_PATH=$(poetry env info --path)
 source "${ENV_PATH}/bin/activate"
 
 #export PYPI_PASSWORD="$(pass personal/apikey/pypi)"
+#export VROBBLER_DB_BACKUP_WEBDAV_URL="https://webdav.example.com/dav/"
+#export VROBBLER_DB_BACKUP_WEBDAV_USER="backups"
+#export VROBBLER_DB_BACKUP_WEBDAV_PASS="$(pass personal/webdav/backups)"
+#export VROBBLER_DB_BACKUP_WEBDAV_PATH="var/vrobbler-backups/"

tests/scrobbles_tests/test_backup_database.py

@@ -0,0 +1,92 @@
+from datetime import datetime
+
+import pytest
+import time_machine
+from scrobbles import tasks as scrobbles_tasks
+
+
+class FakeWebDAVClient:
+    def __init__(self, files=None):
+        self.files = files if files is not None else {}
+        self.uploaded = []
+
+    def mkdir(self, path, recursive=False):
+        self.files.setdefault(path, None)
+
+    def upload_sync(self, remote_path, local_path):
+        with open(local_path, "rb") as f:
+            self.files[remote_path] = f.read()
+        self.uploaded.append(remote_path)
+
+    def list(self, remote_path, get_info=False):
+        names = [
+            k for k in self.files if k.startswith(remote_path) and k != remote_path
+        ]
+        if get_info:
+            return [{"path": n} for n in names]
+        return names
+
+    def clean(self, remote_path):
+        self.files.pop(remote_path, None)
+
+
+@pytest.fixture
+def fake_webdav_client():
+    return FakeWebDAVClient()
+
+
+def test_retention_files_to_delete_keeps_recent_and_monthly():
+    now = datetime(2026, 8, 10)
+    files = [
+        "vrobbler-backup-2026_08_10.sql.gz",
+        "vrobbler-backup-2026_08_09.sql.gz",
+        "vrobbler-backup-2026_08_01.sql.gz",
+        "vrobbler-backup-2026_07_31.sql.gz",
+        "vrobbler-backup-2026_07_01.sql.gz",
+        "vrobbler-backup-2026_06_15.sql.gz",
+        "vrobbler-backup-2026_06_02.sql.gz",
+        "not-a-backup.txt",
+    ]
+    to_delete = scrobbles_tasks._retention_files_to_delete(files, now)
+
+    assert "vrobbler-backup-2026_08_10.sql.gz" not in to_delete
+    assert "vrobbler-backup-2026_08_09.sql.gz" not in to_delete
+    assert "vrobbler-backup-2026_08_01.sql.gz" not in to_delete
+    assert "vrobbler-backup-2026_07_31.sql.gz" not in to_delete
+    assert "vrobbler-backup-2026_06_15.sql.gz" not in to_delete
+    assert "not-a-backup.txt" not in to_delete
+
+    assert "vrobbler-backup-2026_07_01.sql.gz" in to_delete
+    assert "vrobbler-backup-2026_06_02.sql.gz" in to_delete
+
+
+@time_machine.travel(datetime(2026, 8, 10))
+def test_run_webdav_cleanup_prunes_old_backups(fake_webdav_client):
+    path = "var/vrobbler-backups/"
+    fake_webdav_client.files = {
+        f"{path}vrobbler-backup-2026_08_10.sql.gz": b"new",
+        f"{path}vrobbler-backup-2026_07_31.sql.gz": b"monthly",
+        f"{path}vrobbler-backup-2026_07_01.sql.gz": b"old",
+    }
+
+    summary = scrobbles_tasks._run_webdav_cleanup(fake_webdav_client, path)
+
+    assert summary == "pruned 1 old webdav backup(s)"
+    assert f"{path}vrobbler-backup-2026_08_10.sql.gz" in fake_webdav_client.files
+    assert f"{path}vrobbler-backup-2026_07_31.sql.gz" in fake_webdav_client.files
+    assert f"{path}vrobbler-backup-2026_07_01.sql.gz" not in fake_webdav_client.files
+
+
+def test_run_webdav_cleanup_no_files(fake_webdav_client):
+    assert (
+        scrobbles_tasks._run_webdav_cleanup(fake_webdav_client, "var/vrobbler-backups/")
+        is None
+    )
+
+
+def test_run_webdav_cleanup_ignores_unrelated_files(fake_webdav_client):
+    path = "var/vrobbler-backups/"
+    fake_webdav_client.files = {
+        f"{path}notes.txt": b"x",
+    }
+    assert scrobbles_tasks._run_webdav_cleanup(fake_webdav_client, path) is None

vrobbler/apps/scrobbles/tasks.py

@@ -619,9 +619,46 @@ def _run_remote_cleanup(ssh_key, ssh_host, remote_path):
     return f"pruned {len(to_delete)} old remote backup(s)"
 
 
+def _run_webdav_cleanup(webdav_client, remote_path):
+    """List backup files on WebDAV, delete those outside retention.
+
+    Returns a summary string (e.g. "pruned 3 old backup(s)") or None.
+    """
+    import os
+    from datetime import datetime
+
+    try:
+        files = webdav_client.list(remote_path, get_info=True)
+    except Exception as e:
+        logger.warning("backup_database: could not list webdav files (%s)", e)
+        return None
+
+    names = [os.path.basename(f.get("path") or f.get("name") or "") for f in files]
+    names = [n for n in names if n]
+    if not names:
+        return None
+
+    now = datetime.now()
+    to_delete = _retention_files_to_delete(names, now)
+    if not to_delete:
+        logger.info("backup_database: no webdav files to prune")
+        return None
+
+    for fname in to_delete:
+        try:
+            webdav_client.clean(f"{remote_path}{fname}")
+        except Exception as e:
+            logger.warning(
+                "backup_database: could not delete webdav file %s (%s)", fname, e
+            )
+
+    logger.info("backup_database: pruned %d webdav backup(s)", len(to_delete))
+    return f"pruned {len(to_delete)} old webdav backup(s)"
+
+
 @shared_task
 def backup_database():
-    """pg_dump + gzip, scp to remote, retention cleanup, ntfy notification."""
+    """pg_dump + gzip, scp/WebDAV to remote, retention cleanup, ntfy."""
     import os
     import re
     import subprocess
@@ -711,12 +748,44 @@ def backup_database():
                 backup_path,
             )
 
+        webdav_url = getattr(settings, "DB_BACKUP_WEBDAV_URL", "")
+        webdav_cleanup_summary = None
+        if webdav_url:
+            try:
+                from webdav3.client import Client
+
+                webdav_client = Client(
+                    {
+                        "webdav_hostname": webdav_url,
+                        "webdav_login": getattr(settings, "DB_BACKUP_WEBDAV_USER", ""),
+                        "webdav_password": getattr(
+                            settings, "DB_BACKUP_WEBDAV_PASS", ""
+                        ),
+                    }
+                )
+                webdav_path = getattr(
+                    settings, "DB_BACKUP_WEBDAV_PATH", "var/vrobbler-backups/"
+                )
+                webdav_client.mkdir(webdav_path, recursive=True)
+                webdav_dest = f"{webdav_path}{backup_path.name}"
+                logger.info("backup_database: uploading to %s", webdav_dest)
+                webdav_client.upload_sync(webdav_dest, str(backup_path))
+                locations.append(webdav_dest)
+                logger.info("backup_database: uploaded to %s", webdav_dest)
+                webdav_cleanup_summary = _run_webdav_cleanup(webdav_client, webdav_path)
+            except Exception as exc:
+                logger.error("backup_database: webdav upload failed: %s", exc)
+                _cleanup_failed_backup(backup_path)
+                raise
+
         msg = (
             f"✅ Vrobbler backup complete — {size_mb:.1f} MB\n"
             f"Stored at: {', '.join(locations)}"
         )
         if cleanup_summary:
             msg += f"\nRemote: {cleanup_summary}"
+        if webdav_cleanup_summary:
+            msg += f"\nWebDAV: {webdav_cleanup_summary}"
 
         ntfy_url = getattr(
             settings, "DB_BACKUP_NTFY_URL", "https://ntfy.unbl.ink/backups"

vrobbler/settings.py

@@ -87,6 +87,12 @@ BGG_ACCESS_TOKEN = os.getenv("VROBBLER_BGG_ACCESS_TOKEN", "")
 DB_BACKUP_SSH_KEY = os.getenv("VROBBLER_DB_BACKUP_SSH_KEY", "")
 DB_BACKUP_SSH_DEST = os.getenv("VROBBLER_DB_BACKUP_SSH_DEST", "")
 DB_BACKUP_LOCAL_DIR = os.getenv("VROBBLER_DB_BACKUP_LOCAL_DIR", "/var/backups/")
+DB_BACKUP_WEBDAV_URL = os.getenv("VROBBLER_DB_BACKUP_WEBDAV_URL", "")
+DB_BACKUP_WEBDAV_USER = os.getenv("VROBBLER_DB_BACKUP_WEBDAV_USER", "")
+DB_BACKUP_WEBDAV_PASS = os.getenv("VROBBLER_DB_BACKUP_WEBDAV_PASS", "")
+DB_BACKUP_WEBDAV_PATH = os.getenv(
+    "VROBBLER_DB_BACKUP_WEBDAV_PATH", "var/vrobbler-backups/"
+)
 DB_BACKUP_NTFY_URL = os.getenv(
     "VROBBLER_DB_BACKUP_NTFY_URL", "https://ntfy.unbl.ink/backups"
 )