backup-tool/app/specs/container.py

"""Module defining a container-based spec."""
from .spec import Spec
from typing import Union
from pathlib import Path
from datetime import datetime
import subprocess


class ContainerSpec(Spec):
    """Spec for backing up via a container."""

    _SKEL = {"container": None, "command": None, "mountpoint": "/from"}
    """The skeleton for the ContainerSpec config."""

    def __init__(
        self,
        name: str,
        container: str,
        destination: Union[str, Path],
        limit: int,
        command: str,
        extension: str,
        mountpoint: str,
        notify=None,
    ):
        """Create a new ContainerSpec object.

        Args:
            name: name of the spec (used as an identifier)
            container: the Docker container to back up
            destination: where to store the backups (gets created if
                non-existent)
            limit: max amount of backups to keep
            command: command to run inside the container. This command should
                perform a specified backup and output this data to stdout. This
                output then gets piped to a backup file.
            extension: the extension of the backup files.
            mountpoint: I don't actually know, this never gets used
            notify: notifier object (may be None)
        """
        super().__init__(name, destination, limit, extension, notify)

        self.container = container
        self.mountpoint = mountpoint
        self.command = command

    def backup(self):
        """Create a new backup."""
        # Remove excess backups
        self.remove_backups()

        # Run actual backup command
        filename = "{}.{}".format(
            datetime.now().strftime("%Y-%m-%d_%H-%M-%S"), self.extension
        )

        process = subprocess.run(
            "docker exec -t '{}' {} > '{}/{}'".format(
                self.container, self.command, self.destination, filename
            ),
            shell=True,
        )

        if self.notifier:
            self.notifier.notify("backup", self.name, process.returncode)