witten_borgmatic/borgmatic/hooks/data_source/dump.py
2024-12-25 21:59:10 -08:00

76 lines
2.5 KiB
Python

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, port=None):
'''
Based on the given dump directory path, data source name, hostname, and port, 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') + ('' if port is None else f':{port}'), 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
)