borgmatic/borgmatic/hooks/data_source/dump.py

import fnmatch
import logging
import os
import shutil

logger = logging.getLogger(__name__)

IS_A_HOOK = False


def make_data_source_dump_path(borgmatic_runtime_directory, data_source_hook_name):
    '''
    Given a borgmatic runtime directory and a data source hook name, construct a data source dump
    path.
    '''
    return os.path.join(borgmatic_runtime_directory, data_source_hook_name)


def make_data_source_dump_filename(dump_path, name, hostname=None):
    '''
    Based on the given dump directory path, data source name, and hostname, return a filename to use
    for the data source dump. The hostname defaults to localhost.

    Raise ValueError if the data source name is invalid.
    '''
    if os.path.sep in name:
        raise ValueError(f'Invalid data source name {name}')

    return os.path.join(dump_path, hostname or 'localhost', name)


def create_parent_directory_for_dump(dump_path):
    '''
    Create a directory to contain the given dump path.
    '''
    os.makedirs(os.path.dirname(dump_path), mode=0o700, exist_ok=True)


def create_named_pipe_for_dump(dump_path):
    '''
    Create a named pipe at the given dump path.
    '''
    create_parent_directory_for_dump(dump_path)
    os.mkfifo(dump_path, mode=0o600)


def remove_data_source_dumps(dump_path, data_source_type_name, log_prefix, dry_run):
    '''
    Remove all data source dumps in the given dump directory path (including the directory itself).
    If this is a dry run, then don't actually remove anything.
    '''
    dry_run_label = ' (dry run; not actually removing anything)' if dry_run else ''

    logger.debug(f'{log_prefix}: Removing {data_source_type_name} data source dumps{dry_run_label}')

    if dry_run:
        return

    if os.path.exists(dump_path):
        shutil.rmtree(dump_path)


def convert_glob_patterns_to_borg_pattern(patterns):
    '''
    Convert a sequence of shell glob patterns like "/etc/*", "/tmp/*" to the corresponding Borg
    regular expression archive pattern as a single string like "re:etc/.*|tmp/.*".
    '''
    # Remove the "\Z" generated by fnmatch.translate() because we don't want the pattern to match
    # only at the end of a path, as directory format dumps require extracting files with paths
    # longer than the pattern. E.g., a pattern of "borgmatic/*/foo_databases/test" should also match
    # paths like "borgmatic/*/foo_databases/test/toc.dat"
    return 're:' + '|'.join(
        fnmatch.translate(pattern.lstrip('/')).replace('\\Z', '') for pattern in patterns
    )