backup-tool/app/specs/directory.py

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


class DirectorySpec(Spec):
    """A spec for backing up a local directory."""

    _SKEL = {
        "source": None,
        "command": "tar -czf '{destination}/{filename}' .",
    }

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

        Args:
            name: name of the spec
            source: what directory to back up
            destination: where to store the backup
            limit: how many backups to keep
            command: what command to use to create the backup
            extension: extension of the backup files
            notify: a Notifier object that handles sending notifications
        """
        super().__init__(name, destination, limit, extension, notify)

        self.source = source if type(source) == Path else Path(source)

        # Check existence of source directory
        if not self.source.exists() or not self.source.is_dir():
            raise NotADirectoryError(
                "{} doesn't exist or isn't a directory.".format(self.source)
            )

        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
        )

        # TODO add logging stuff
        process = subprocess.run(
            self.command.format(
                destination=self.destination,
                filename=filename,
            ),
            cwd=self.source,
            shell=True,
        )

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