backup-tool/app/specs/volume.py

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


class VolumeSpec(Spec):
    """A spec for backing up a Docker volume."""

    _SKEL = {
        "volume": None,
        "image": "alpine:latest",
        "command": "tar -czf '/to/{filename}' .",
    }

    def __init__(
        self,
        name: str,
        volume: str,
        image: str,
        destination: Union[str, Path],
        limit: int,
        command: str,
        extension: str,
        notify=None,
    ):
        """Initialize a new VolumeSpec object.

        Args:
            name: name of the spec
            volume: Docker volume to back up
            image: base image to use to run backup command
            destination: where to store the backup files
            limit: max backup files to keep
            command: backup command to run within the base image
            extension: file extension of the backup files
            notify: Notifier object
        """
        super().__init__(name, destination, limit, extension, notify)

        self.volume = volume
        self.image = image
        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
        )

        base_cmd = "docker run --rm -v '{}:/from' -v '{}:/to' -w /from '{}' {}"

        process = subprocess.run(
            base_cmd.format(
                self.volume,
                self.destination,
                self.image,
                self.command.format(
                    filename=filename,
                ),
            ),
            shell=True,
        )

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