Utility Operations

import subprocess
import json

def get_version_info(server_repo: str = None) -> str:
    """
    Get BorgBackup version information.
    
    Args:
        server_repo: Repository path to get server version (optional)
        
    Returns:
        Version information string
    """
    cmd = ['borg', 'version']
    if server_repo:
        cmd.append(server_repo)
    
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return result.stdout.strip()

import subprocess

# Get client version
result = subprocess.run(['borg', 'version'], capture_output=True, text=True, check=True)
client_version = result.stdout.strip()
print(f"Borg client version: {client_version}")

# Get server version (if using remote repository)
result = subprocess.run(['borg', 'version', 'user@server:/backup/repo'], 
                       capture_output=True, text=True, check=True)
server_version = result.stdout.strip()

def get_help(topic: str = None) -> str:
    """
    Get help information.
    
    Args:
        topic: Specific topic or command to get help for (optional)
        
    Returns:
        Help text
    """
    cmd = ['borg', 'help']
    if topic:
        cmd.append(topic)
    
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return result.stdout

import subprocess

# Get general help
result = subprocess.run(['borg', 'help'], capture_output=True, text=True, check=True)
general_help = result.stdout

# Get help for specific command
result = subprocess.run(['borg', 'help', 'create'], 
                       capture_output=True, text=True, check=True)
create_help = result.stdout

# Get help for specific topics
available_topics = ['patterns', 'placeholders', 'compression']
for topic in available_topics:
    result = subprocess.run(['borg', 'help', topic], 
                           capture_output=True, text=True, check=True)
    topic_help = result.stdout

def benchmark_crud(repo_path: str, path: str, file_size: int = 10485760,
                  file_count: int = 40, data_seed: int = None) -> None:
    """
    Benchmark CRUD (Create, Read, Update, Delete) operations.
    
    Args:
        repo_path: Path to repository for benchmarking
        path: Path to directory for benchmark files
        file_size: Size of each test file in bytes (default: 10MB)
        file_count: Number of files to create (default: 40)
        data_seed: Seed for random data generation
    """
    cmd = ['borg', 'benchmark', 'crud', repo_path, path]
    if file_size != 10485760:
        cmd.extend(['--file-size', str(file_size)])
    if file_count != 40:
        cmd.extend(['--file-count', str(file_count)])
    if data_seed:
        cmd.extend(['--data-seed', str(data_seed)])
    
    subprocess.run(cmd, check=True)

import subprocess

# Run standard benchmark
subprocess.run([
    'borg', 'benchmark', 'crud', '/benchmark/repo', '/tmp/benchmark'
], check=True)

# Run benchmark with custom parameters
subprocess.run([
    'borg', 'benchmark', 'crud', 
    '--file-size=1048576',  # 1MB files
    '--file-count=100',     # 100 files
    '/benchmark/repo', '/tmp/benchmark'
], check=True)

def with_lock(repo_path: str, command: list, wait: int = None) -> None:
    """
    Run command with repository lock held.
    
    Args:
        repo_path: Path to repository
        command: Command to run while holding lock
        wait: Maximum time to wait for lock in seconds
    """
    cmd = ['borg', 'with-lock']
    if wait:
        cmd.extend(['--lock-wait', str(wait)])
    cmd.append(repo_path)
    cmd.extend(command)
    
    subprocess.run(cmd, check=True)

import subprocess

# Run custom script with repository lock
subprocess.run([
    'borg', 'with-lock', '/backup/repo',
    'python', '/scripts/backup-maintenance.py'
], check=True)

# Run command with lock timeout
subprocess.run([
    'borg', 'with-lock', '--lock-wait=300',  # Wait up to 5 minutes
    '/backup/repo', 'rsync', '-av', '/data/', '/backup/staging/'
], check=True)

def serve_repository_advanced(restrict_paths: list = None, 
                            restrict_repos: list = None,
                            append_only: bool = False,
                            storage_quota: str = None,
                            umask: str = None) -> None:
    """
    Serve repository with advanced options (typically called via SSH).
    
    Args:
        restrict_paths: List of allowed path prefixes
        restrict_repos: List of allowed repository paths  
        append_only: Allow only append operations
        storage_quota: Set storage quota limit (e.g., '5G', '100M')
        umask: Set umask for created files (octal, e.g., '0077')
    """
    cmd = ['borg', 'serve']
    if restrict_paths:
        for path in restrict_paths:
            cmd.extend(['--restrict-to-path', path])
    if restrict_repos:
        for repo in restrict_repos:
            cmd.extend(['--restrict-to-repository', repo])
    if append_only:
        cmd.append('--append-only')
    if storage_quota:
        cmd.extend(['--storage-quota', storage_quota])
    if umask:
        cmd.extend(['--umask', umask])
    
    subprocess.run(cmd, check=True)

import subprocess

# Typical SSH authorized_keys entry for restricted serve
# command="borg serve --restrict-to-path /backup/repos --append-only" ssh-rsa AAAA...

# Serve with multiple restrictions
# command="borg serve --restrict-to-repository /backup/user1 --restrict-to-repository /backup/user2 --storage-quota 10G" ssh-rsa AAAA...

class VersionInfo:
    """Version information structure"""
    def __init__(self):
        self.client_version: str        # Client version string
        self.server_version: str        # Server version string (if applicable)
        self.python_version: str        # Python version
        self.platform: str              # Operating system platform
        
class BenchmarkResults:
    """Benchmark operation results"""
    def __init__(self):
        self.create_time: float         # Time for create operations
        self.extract_time: float        # Time for extract operations  
        self.update_time: float         # Time for update operations
        self.delete_time: float         # Time for delete operations
        self.throughput: dict           # Throughput statistics
        
class LockInfo:
    """Repository lock information"""
    def __init__(self):
        self.exclusive: bool            # Exclusive lock held
        self.lock_id: str              # Lock identifier
        self.timestamp: str            # Lock acquisition time
        self.hostname: str             # Host holding the lock
        self.pid: int                  # Process ID holding the lock