Source code for twindb_backup.ls

# -*- coding: utf-8 -*-
"""
Module that works with list of backup copies
"""
from __future__ import print_function

from twindb_backup import INTERVALS, LOG, MEDIA_TYPES
from twindb_backup.destination.local import Local


def list_available_backups(twindb_config, copy_type=None):
    """
    Print known backup copies on a destination specified in the configuration.

    :param twindb_config: tool configuration
    :type twindb_config: TwinDBBackupConfig
    :param copy_type: Limit list to specific type of backups.
    :type copy_type: files|mysql
    """
    dsts = [twindb_config.destination()]
    if twindb_config.keep_local_path:
        dsts.insert(0, Local(twindb_config.keep_local_path))

    for dst in dsts:
        LOG.info("Destination %s", dst)
        for mtype in MEDIA_TYPES:
            if copy_type in [None, mtype]:
                globals()[f"_print_{mtype}"](dst)


def _print_files(dst):
    _print_media_type(dst, "files")


def _print_mysql(dst):
    _print_media_type(dst, "mysql")


def _print_media_type(dst, media_type):
    for run_type in INTERVALS:
        pattern = f"/{run_type}/{media_type}/"
        dst_files = dst.list_files(dst.remote_path, pattern=pattern, recursive=True, files_only=True)
        if dst_files:
            LOG.info("%s %s copies:", media_type, run_type)
            for copy in dst_files:
                print(copy)


def _print_binlog(dst):
    dst_files = dst.list_files(dst.remote_path, pattern="/binlog/", recursive=True, files_only=True)
    if dst_files:
        LOG.info("Binary logs:")
        for copy in dst_files:
            print(copy)