#!/usr/bin/env python3 """ backup_compose_volumes.py ------------------------- Export (compress) every Docker *volume* that belongs to a Docker Compose project. DISCOVERY MODES - By label (default): looks for label com.docker.compose.project= - By name prefix (fallback/optional): looks for volume names starting with _ or - This helps when volumes were created with a Compose project like "suitecoffee_dev" or "suitecoffee_prod" and you're passing "SuiteCoffee" (capitalized) or when some volumes lack labels. Usage examples -------------- python3 backup_compose_volumes.py -p suitecoffee_dev python3 backup_compose_volumes.py -p suitecoffee_prod -o /backups/suitecoffee python3 backup_compose_volumes.py -p SuiteCoffee --discovery auto python3 backup_compose_volumes.py -p suitecoffee --discovery name # treat -p as a name prefix python3 backup_compose_volumes.py --list-only # just list what would be backed up Notes ----- - You generally want to pass the EXACT Compose project name (e.g., "suitecoffee_dev"). - Docker Compose sets project names in lowercase; labels are case-sensitive. - If zero volumes are found by label, this script tries lowercase and name-prefix fallback automatically. """ import argparse import datetime import json import os import pathlib import shlex import subprocess import sys from typing import List, Dict def run(cmd: List[str], check=True, capture_output=True, text=True) -> subprocess.CompletedProcess: return subprocess.run(cmd, check=check, capture_output=capture_output, text=text) def which(program: str) -> bool: from shutil import which as _which return _which(program) is not None def detect_project_name(args_project: str) -> str: if args_project: return args_project env_name = os.environ.get("COMPOSE_PROJECT_NAME") if env_name: return env_name return pathlib.Path.cwd().name.replace(" ", "_") def docker_volume_ls_json(filters: List[str]) -> List[Dict[str, str]]: cmd = ["docker", "volume", "ls", "--format", "{{json .}}"] for f in filters: cmd += ["--filter", f] cp = run(cmd) out = [] for line in cp.stdout.splitlines(): try: out.append(json.loads(line)) except json.JSONDecodeError: pass return out def list_by_label(project: str) -> List[Dict[str, str]]: return docker_volume_ls_json([f"label=com.docker.compose.project={project}"]) def list_by_name_prefix(prefix: str) -> List[Dict[str, str]]: # docker volume ls has no "name prefix" filter; we filter client-side. vols = docker_volume_ls_json([]) keep = [] for v in vols: name = v.get("Name") or v.get("Driver") # Name should be present if not name: continue if name.startswith(prefix + "_") or name.startswith(prefix + "-") or name == prefix: keep.append(v) return keep def ensure_alpine_image(): try: run(["docker", "image", "inspect", "alpine:latest"]) except subprocess.CalledProcessError: print("Pulling alpine:latest ...") run(["docker", "pull", "alpine:latest"], check=True, capture_output=False) def backup_volume(volume_name: str, out_dir: pathlib.Path, archive_name: str, dry_run: bool = False) -> int: out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / archive_name docker_cmd = [ "docker", "run", "--rm", "-v", f"{volume_name}:/volume:ro", "-v", f"{str(out_dir)}:/backup", "alpine:latest", "sh", "-lc", f"tar czf /backup/{shlex.quote(out_path.name)} -C /volume ." ] if dry_run: print("[DRY RUN] Would run:", " ".join(shlex.quote(c) for c in docker_cmd)) return 0 cp = subprocess.run(docker_cmd) return cp.returncode def main(): parser = argparse.ArgumentParser(description="Export (compress) every Docker volume of a Docker Compose project.") parser.add_argument("-p", "--project", help="Compose project or prefix (see --discovery).") parser.add_argument("-o", "--output", help="Output directory (default: ./docker-volumes-backup-).") parser.add_argument("--exclude", nargs="*", default=[], help="Volume names to exclude (space-separated).") parser.add_argument("--dry-run", action="store_true", help="Show what would be done without doing it.") parser.add_argument("--timestamp", default=datetime.datetime.now().strftime("%Y%m%d-%H%M%S"), help="Timestamp to embed into filenames (default: current time).") parser.add_argument("--discovery", choices=["auto","label","name"], default="auto", help="How to discover volumes: 'label' (strict), 'name' (prefix), or 'auto' (default).") parser.add_argument("--list-only", action="store_true", help="Only list volumes that would be backed up and exit.") args = parser.parse_args() if not which("docker"): print("ERROR: 'docker' is not on PATH. Please install Docker and/or add it to PATH.", file=sys.stderr) sys.exit(2) project_raw = detect_project_name(args.project) project_norm = project_raw.replace(" ", "_") project_lower = project_norm.lower() ts = args.timestamp out_dir = pathlib.Path(args.output) if args.output else pathlib.Path(f"./docker-volumes-backup-{ts}") # Ensure daemon available try: run(["docker", "version"], check=True, capture_output=True) except subprocess.CalledProcessError: print("ERROR: Unable to talk to the Docker daemon. Are you in the 'docker' group? Is the daemon running?", file=sys.stderr) sys.exit(2) selected = [] method_used = None # Discovery if args.discovery in ("auto","label"): vols = list_by_label(project_norm) if vols: selected = vols; method_used = f"label:{project_norm}" elif args.discovery == "auto": vols2 = list_by_label(project_lower) if vols2: selected = vols2; method_used = f"label:{project_lower}" if not selected and args.discovery in ("auto","name"): # Treat project as a prefix # Try exact, then lowercase by_name = list_by_name_prefix(project_norm) if by_name: selected = by_name; method_used = f"name-prefix:{project_norm}" else: by_name2 = list_by_name_prefix(project_lower) if by_name2: selected = by_name2; method_used = f"name-prefix:{project_lower}" if not selected: print(f"No volumes found for project/prefix '{project_raw}'. Tried methods:") print(f" - label:{project_norm}") print(f" - label:{project_lower}") print(f" - name-prefix:{project_norm} (prefix_*, prefix-*)") print(f" - name-prefix:{project_lower} (prefix_*, prefix-*)") sys.exit(0) exclude_set = set(args.exclude or []) selected = [v for v in selected if v.get("Name") not in exclude_set] print(f"Discovery method: {method_used}") print(f"Volumes discovered: {len(selected)}") for v in selected: print(" -", v.get("Name")) if args.list_only: return if not args.dry_run: ensure_alpine_image() failures = [] for v in selected: vname = v.get("Name") if not vname: continue # Determine a 'project token' for filename: take the leading prefix before first '_' or '-' prefix = project_lower archive = f"{prefix}-{vname}-{ts}.tar.gz" print(f"Backing up volume: {vname} -> {archive}") rc = backup_volume(vname, out_dir, archive, dry_run=args.dry_run) if rc != 0: print(f" ERROR: backup failed for volume '{vname}' (exit code {rc})", file=sys.stderr) failures.append(vname) if failures: print("\nCompleted with errors. Failed volumes:", ", ".join(failures)) sys.exit(1) else: print("\nAll done. Archives written to:", str(out_dir.resolve())) if __name__ == "__main__": main()