tessl/pypi-borgbackup

Deduplicated, encrypted, authenticated and compressed backups

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Pruning and Maintenance

Name: tessl/pypi-borgbackup
Author: tessl

Archive pruning, repository maintenance, and cleanup operations for managing backup retention policies and optimizing repository storage in BorgBackup.

import subprocess

Capabilities

Archive Pruning

Automatically remove old archives based on retention policies to manage storage space and backup history.

def prune_archives(repo_path: str, keep_within: str = None, keep_daily: int = None,
                  keep_weekly: int = None, keep_monthly: int = None, 
                  keep_yearly: int = None, dry_run: bool = False,
                  stats: bool = False, list_archives: bool = False) -> None:
    """
    Prune archives based on retention policy.
    
    Args:
        repo_path: Path to repository
        keep_within: Keep archives within time period (e.g., '1d', '7d', '1m', '1y')
        keep_daily: Number of daily archives to keep
        keep_weekly: Number of weekly archives to keep  
        keep_monthly: Number of monthly archives to keep
        keep_yearly: Number of yearly archives to keep
        dry_run: Show what would be pruned without pruning
        stats: Show pruning statistics
        list_archives: List archives that would be kept/pruned
    """
    cmd = ['borg', 'prune']
    if keep_within:
        cmd.extend(['--keep-within', keep_within])
    if keep_daily:
        cmd.extend(['--keep-daily', str(keep_daily)])
    if keep_weekly:
        cmd.extend(['--keep-weekly', str(keep_weekly)])
    if keep_monthly:
        cmd.extend(['--keep-monthly', str(keep_monthly)])
    if keep_yearly:
        cmd.extend(['--keep-yearly', str(keep_yearly)])
    if dry_run:
        cmd.append('--dry-run')
    if stats:
        cmd.append('--stats')
    if list_archives:
        cmd.append('--list')
    
    cmd.append(repo_path)
    subprocess.run(cmd, check=True)

Usage example:

import subprocess

# Prune with common retention policy
subprocess.run([
    'borg', 'prune', '--stats', 
    '--keep-daily=7',    # Keep 7 daily backups
    '--keep-weekly=4',   # Keep 4 weekly backups
    '--keep-monthly=6',  # Keep 6 monthly backups
    '--keep-yearly=2',   # Keep 2 yearly backups
    '/backup/repo'
], check=True)

# Prune keeping archives within last 30 days
subprocess.run(['borg', 'prune', '--keep-within=30d', '/backup/repo'], check=True)

# Dry run to see what would be pruned
subprocess.run(['borg', 'prune', '--dry-run', '--list', '--keep-daily=7', '/backup/repo'], check=True)

Repository Compaction

Reclaim space by removing deleted data and optimizing repository storage.

def compact_repository(repo_path: str, threshold: float = None, 
                      dry_run: bool = False, progress: bool = False) -> None:
    """
    Compact repository to reclaim space.
    
    Args:
        repo_path: Path to repository
        threshold: Minimum saved space ratio to trigger compaction (0.0-1.0)
        dry_run: Show compaction statistics without compacting
        progress: Show progress during compaction
    """
    cmd = ['borg', 'compact']
    if threshold:
        cmd.extend(['--threshold', str(threshold)])
    if dry_run:
        cmd.append('--dry-run')
    if progress:
        cmd.append('--progress')
    
    cmd.append(repo_path)
    subprocess.run(cmd, check=True)

Usage example:

import subprocess

# Compact repository with progress
subprocess.run(['borg', 'compact', '--progress', '/backup/repo'], check=True)

# Compact only if at least 20% space can be saved
subprocess.run(['borg', 'compact', '--threshold=0.2', '/backup/repo'], check=True)

# Dry run to see potential space savings
subprocess.run(['borg', 'compact', '--dry-run', '/backup/repo'], check=True)

Cache Management

Manage local cache for improved performance, including cache deletion and recreation.

def delete_cache(repo_path: str, cache_only: bool = False, 
                force: bool = False) -> None:
    """
    Delete repository cache.
    
    Args:
        repo_path: Path to repository
        cache_only: Delete only cache files, not security info
        force: Force deletion without confirmation
    """
    cmd = ['borg', 'delete']
    if cache_only:
        cmd.append('--cache-only')
    if force:
        cmd.append('--force')
    
    cmd.append(repo_path)
    subprocess.run(cmd, check=True)

def recreate_cache(repo_path: str, progress: bool = False) -> None:
    """
    Recreate repository cache by reading all archives.
    
    Args:
        repo_path: Path to repository  
        progress: Show progress during cache recreation
    """
    # Cache is recreated automatically on next access
    # Force recreation by listing archives
    cmd = ['borg', 'list']
    if progress:
        cmd.append('--progress')
    cmd.append(repo_path)
    subprocess.run(cmd, check=True)

Usage example:

import subprocess

# Delete cache (will be recreated on next access)
subprocess.run(['borg', 'delete', '--cache-only', '/backup/repo'], check=True)

# Recreate cache by listing archives
subprocess.run(['borg', 'list', '/backup/repo'], check=True)

Archive Recreation

Recreate archives with different settings, such as changing compression or excluding/including different files.

def recreate_archive(repo_path: str, archive_name: str = None, 
                    target_name: str = None, excludes: list = None,
                    compression: str = None, recompress: bool = False,
                    dry_run: bool = False, stats: bool = False) -> None:
    """
    Recreate archive with modified settings.
    
    Args:
        repo_path: Path to repository
        archive_name: Archive to recreate (or pattern)
        target_name: New archive name
        excludes: List of exclusion patterns
        compression: New compression algorithm
        recompress: Recompress all chunks
        dry_run: Show what would be recreated
        stats: Show recreation statistics
    """
    cmd = ['borg', 'recreate']
    if target_name:
        cmd.extend(['--target', target_name])
    if excludes:
        for exclude in excludes:
            cmd.extend(['--exclude', exclude])
    if compression:
        cmd.extend(['--compression', compression])
    if recompress:
        cmd.append('--recompress')
    if dry_run:
        cmd.append('--dry-run')
    if stats:
        cmd.append('--stats')
    
    if archive_name:
        cmd.append(f'{repo_path}::{archive_name}')
    else:
        cmd.append(repo_path)
    
    subprocess.run(cmd, check=True)

Usage example:

import subprocess

# Recreate archive with better compression
subprocess.run([
    'borg', 'recreate', '--compression=zstd,9', '--recompress',
    '/backup/repo::old-archive'
], check=True)

# Recreate excluding additional patterns
subprocess.run([
    'borg', 'recreate', '--exclude=*.log', '--exclude=temp/*',
    '/backup/repo::archive-name'
], check=True)

# Dry run to see what would change
subprocess.run([
    'borg', 'recreate', '--dry-run', '--stats', '--compression=lzma,6',
    '/backup/repo::test-archive'
], check=True)

Repository Upgrading

Upgrade repository format to newer versions for improved features and performance.

def upgrade_repository(repo_path: str, dry_run: bool = False, 
                      force: bool = False, tam: bool = False) -> None:
    """
    Upgrade repository format.
    
    Args:
        repo_path: Path to repository
        dry_run: Show what would be upgraded
        force: Force upgrade without confirmation
        tam: Enable/disable TAM (Tamper Authentication Mode)
    """
    cmd = ['borg', 'upgrade']
    if dry_run:
        cmd.append('--dry-run')
    if force:
        cmd.append('--force')
    if tam:
        cmd.append('--tam')
    
    cmd.append(repo_path)
    subprocess.run(cmd, check=True)

Usage example:

import subprocess

# Upgrade repository (dry run first)
subprocess.run(['borg', 'upgrade', '--dry-run', '/backup/repo'], check=True)

# Perform actual upgrade
subprocess.run(['borg', 'upgrade', '/backup/repo'], check=True)

Repository Serving

Serve repository over SSH for remote access with security restrictions.

def serve_repository(restrict_to_paths: list = None, restrict_to_repositories: list = None,
                    append_only: bool = False, storage_quota: str = None) -> None:
    """
    Serve repository for remote access (typically called via SSH).
    
    Args:
        restrict_to_paths: Restrict access to specific paths
        restrict_to_repositories: Restrict access to specific repositories
        append_only: Allow only append operations
        storage_quota: Set storage quota limit
    """
    cmd = ['borg', 'serve']
    if restrict_to_paths:
        cmd.extend(['--restrict-to-path'] + restrict_to_paths)
    if restrict_to_repositories:
        cmd.extend(['--restrict-to-repository'] + restrict_to_repositories)
    if append_only:
        cmd.append('--append-only')
    if storage_quota:
        cmd.extend(['--storage-quota', storage_quota])
    
    subprocess.run(cmd, check=True)

Usage example:

import subprocess

# Serve with path restrictions (typically in SSH authorized_keys)
# command="borg serve --restrict-to-path /backup/repos" ssh-rsa AAAA...

# Serve in append-only mode
# command="borg serve --append-only --restrict-to-path /backup/repo" ssh-rsa AAAA...

Lock Management

Break repository and cache locks when operations are interrupted or stuck.

def break_repository_lock(repo_path: str) -> None:
    """
    Break repository and cache locks.
    
    Args:
        repo_path: Path to repository
        
    Warning: Only use when no borg process is running
    """
    cmd = ['borg', 'break-lock', repo_path]
    subprocess.run(cmd, check=True)

Usage example:

import subprocess

# Break locks when repository appears stuck
# WARNING: Ensure no other borg processes are running first!
subprocess.run(['borg', 'break-lock', '/backup/repo'], check=True)

Types

class PruneStats:
    """Pruning operation statistics"""
    def __init__(self):
        self.kept_archives: int         # Number of archives kept
        self.pruned_archives: int       # Number of archives pruned
        self.deleted_data: int          # Bytes of data deleted
        self.freed_space: int           # Space freed in repository
        
class CompactionStats:
    """Repository compaction statistics"""
    def __init__(self):
        self.segments_before: int       # Segments before compaction
        self.segments_after: int        # Segments after compaction
        self.size_before: int           # Size before compaction
        self.size_after: int            # Size after compaction
        self.freed_space: int           # Space freed by compaction
        self.time_elapsed: float        # Compaction time in seconds

class UpgradeInfo:
    """Repository upgrade information"""
    def __init__(self):
        self.version_before: str        # Version before upgrade
        self.version_after: str         # Version after upgrade
        self.tam_enabled: bool          # TAM authentication enabled
        self.changes_made: list         # List of changes made

Install with Tessl CLI