Created
February 25, 2026 14:24
-
-
Save maqnius/98b8c788c3e3b497838229b4c4c62890 to your computer and use it in GitHub Desktop.
execute restic profile if possible and not done yet
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """ | |
| Execute resticprofiles if possible and at most once daily. | |
| Can and should be polled until successful e.g.: | |
| # try every 10 minutes to backup if not done yet | |
| */10 * * * * daily.py nextcloud >> /var/log/backup_nextcloud.log 2>&1 | |
| Last backup ist tracked with a lock file next to daily.py for each repository. | |
| Run with DEBUG=1 for more verbosity | |
| DEBUG=1 daily.py nextcloud | |
| """ | |
| import logging | |
| import sys | |
| import os | |
| import subprocess | |
| from pathlib import Path | |
| from datetime import datetime, date | |
| logging.basicConfig( | |
| level=logging.DEBUG if os.getenv("DEBUG", "0") == "1" else logging.INFO, | |
| format="%(asctime)s [%(levelname)s] %(message)s", | |
| ) | |
| def run(cmd: list[str]) -> str: | |
| try: | |
| result = subprocess.run( | |
| cmd, | |
| capture_output=True, | |
| text=True, | |
| check=True, | |
| ) | |
| if result.stdout: | |
| logging.debug(result.stdout.strip()) | |
| except subprocess.CalledProcessError as e: | |
| return e.stdout | |
| return "" | |
| def lockfile(repo: str) -> Path: | |
| return Path(__file__).parent / f"{repo}-last-run.lock" | |
| def available(repo: str) -> bool: | |
| err = run(["resticprofile", f"{repo}.snapshots", "--latest", "1"]) | |
| if err: | |
| logging.debug("err: %s", err) | |
| return not err | |
| def already_done(repo: str) -> bool: | |
| lf = lockfile(repo) | |
| if not lf.exists(): | |
| return False | |
| mtime = datetime.fromtimestamp(lf.stat().st_mtime).date() | |
| return mtime == date.today() | |
| def backup(repo: str) -> str: | |
| return run(["resticprofile", f"{repo}.backup"]) | |
| def mark_done(repo: str): | |
| """touch lock file""" | |
| lockfile(repo).touch(exist_ok=True) | |
| if __name__ == "__main__": | |
| try: | |
| repo = sys.argv[1] | |
| except IndexError: | |
| repo = "default" | |
| if already_done(repo): | |
| logging.debug("backup for %s already done for today", repo) | |
| sys.exit(0) | |
| if not available(repo): | |
| logging.debug("repository %s not available", repo) | |
| sys.exit(0) | |
| logging.info("start backup for %s", repo) | |
| err = backup(repo) | |
| if err: | |
| logging.error("backup failed for %s:\n%s", repo, err) | |
| sys.exit(1) | |
| logging.info("backup for %s successful", repo) | |
| mark_done(repo) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment