#!/usr/bin/env python3 """ Docker Compose to Proxmox Backup Server Backup script that reads docker-compose.yaml, extracts service configuration, stops the service, creates backup on PBS, and restarts the service. """ import argparse import sys import os import subprocess import yaml import json from pathlib import Path from datetime import datetime from typing import Dict, List, Any, Optional class Docker2PBS: def __init__(self, compose_file: str, service_name: str, pbs_config: Dict[str, str], dry_run: bool = False): self.compose_file = Path(compose_file) self.service_name = service_name self.pbs_config = pbs_config self.dry_run = dry_run self.compose_data = None self.service_config = None self.volumes = [] self.networks = [] def load_compose_file(self) -> None: """Load and parse docker-compose.yaml file""" if not self.compose_file.exists(): raise FileNotFoundError(f"Docker compose file not found: {self.compose_file}") with open(self.compose_file, 'r') as f: self.compose_data = yaml.safe_load(f) if 'services' not in self.compose_data: raise ValueError("No 'services' section found in docker-compose.yaml") if self.service_name not in self.compose_data['services']: available_services = list(self.compose_data['services'].keys()) raise ValueError(f"Service '{self.service_name}' not found. Available services: {available_services}") self.service_config = self.compose_data['services'][self.service_name] def parse_volumes(self) -> List[Dict[str, Any]]: """Parse volumes configuration for the service""" volumes = [] # Get service volumes if 'volumes' in self.service_config: for volume in self.service_config['volumes']: volume_info = {'type': 'bind', 'external': False} if isinstance(volume, str): # Short syntax: "host_path:container_path" parts = volume.split(':') if len(parts) >= 2: volume_info.update({ 'source': parts[0], 'target': parts[1], 'mode': parts[2] if len(parts) > 2 else 'rw' }) elif isinstance(volume, dict): # Long syntax volume_info.update(volume) if volume_info.get('type') == 'volume': # Check if it's external volume volume_name = volume_info.get('source') if volume_name and self.is_external_volume(volume_name): volume_info['external'] = True volumes.append(volume_info) return volumes def is_external_volume(self, volume_name: str) -> bool: """Check if volume is external""" if 'volumes' in self.compose_data: volume_config = self.compose_data['volumes'].get(volume_name, {}) return volume_config.get('external', False) return False def parse_networks(self) -> List[Dict[str, Any]]: """Parse network configuration for the service""" networks = [] if 'networks' in self.service_config: service_networks = self.service_config['networks'] if isinstance(service_networks, list): # Short syntax for network in service_networks: networks.append({ 'name': network, 'external': self.is_external_network(network) }) elif isinstance(service_networks, dict): # Long syntax for network_name, config in service_networks.items(): network_info = { 'name': network_name, 'external': self.is_external_network(network_name) } if config: network_info.update(config) networks.append(network_info) return networks def is_external_network(self, network_name: str) -> bool: """Check if network is external""" if 'networks' in self.compose_data: network_config = self.compose_data['networks'].get(network_name, {}) return network_config.get('external', False) return False def stop_service(self) -> None: """Stop the Docker service""" print(f"Stopping service: {self.service_name}") cmd = ['docker-compose', '-f', str(self.compose_file), 'stop', self.service_name] if self.dry_run: print(f"[DRY-RUN] Would run: {' '.join(cmd)}") print(f"[DRY-RUN] Service {self.service_name} would be stopped") return result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError(f"Failed to stop service: {result.stderr}") print(f"Service {self.service_name} stopped successfully") def start_service(self) -> None: """Start the Docker service""" print(f"Starting service: {self.service_name}") cmd = ['docker-compose', '-f', str(self.compose_file), 'up', '-d', self.service_name] if self.dry_run: print(f"[DRY-RUN] Would run: {' '.join(cmd)}") print(f"[DRY-RUN] Service {self.service_name} would be started") return result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError(f"Failed to start service: {result.stderr}") print(f"Service {self.service_name} started successfully") def main(): parser = argparse.ArgumentParser(description='Backup Docker service to Proxmox Backup Server') parser.add_argument('compose_file', help='Path to docker-compose.yaml file') parser.add_argument('service_name', help='Name of the service to backup') parser.add_argument('--pbs-repository', required=True, help='PBS repository (user@host:datastore)') parser.add_argument('--pbs-username', help='PBS username') parser.add_argument('--pbs-password', help='PBS password') parser.add_argument('--pbs-fingerprint', help='PBS server fingerprint') parser.add_argument('--dry-run', action='store_true', help='Show what would be done without executing commands') args = parser.parse_args() pbs_config = { 'repository': args.pbs_repository } if args.pbs_username: pbs_config['username'] = args.pbs_username if args.pbs_password: pbs_config['password'] = args.pbs_password if args.pbs_fingerprint: pbs_config['fingerprint'] = args.pbs_fingerprint try: backup_tool = Docker2PBS(args.compose_file, args.service_name, pbs_config, args.dry_run) backup_tool.load_compose_file() print(f"Successfully loaded compose file and found service: {args.service_name}") except Exception as e: print(f"Error: {e}") sys.exit(1) if __name__ == '__main__': main()