docker2pbs/docker2pbs.py

#!/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 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()