self-hosting/backups/app/specs/specs.py

from pathlib import Path
from datetime import datetime
import requests
import os


class Spec:
    def __init__(self, name, destination, limit, title, events=None):
        self.name        = name
        self.destination = Path(destination)
        self.limit       = limit
        self.title       = title
        self.events      = [] if events is None else events

    def to_dict(self):
        return {
            "name": self.name,
            "destination": str(self.destination),
            "limit": self.limit,
            "notify": {
                "title": self.title,
                "events": self.events
            }
        }

    def backup(self):
        raise NotImplementedError()

    def remove_redundant(self):
        tarballs = sorted(self.destination.glob('*.tar.gz'),
                          key=os.path.getmtime, reverse=True)

        if len(tarballs) >= self.limit:
            for path in tarballs[self.limit - 1:]:
                path.unlink()

    def notify(self, status_code):
        if status_code:
            if "failure" not in self.events:
                return

            message = "backup for {} failed.".format(self.name)

        else:
            if "success" not in self.events:
                return

            message = "backup for {} succeeded.".format(self.name)

        # Read API key from env vars
        try:
            key = os.environ["IFTTT_API_KEY"]

        # Don't send notification if there's not API key defined
        except KeyError:
            return

        url = "https://maker.ifttt.com/trigger/{}/with/key/{}".format(
            "phone_notifications",
            key
        )

        data = {
            "value1": self.title,
            "value2": message
        }

        requests.post(url, data=data)

    def get_filename(self):
        return '{}_{}.tar.gz'.format(
            self.name,
            datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
        )

    @staticmethod
    def from_dict(name, data) -> "Specification":
        if data.get("volume", False):
            return VolumeSpec.from_dict(name, data)

        return DirSpec.from_dict(name, data)

    @staticmethod
    def from_file(path: str):
        with open(path, 'r') as yaml_file:
            data = yaml.load(yaml_file, Loader=yaml.Loader)

        return [Spec.from_dict(name, info)
                for name, info in data["specs"].items()]


class DirSpec(Spec):
    def __init__(self, name, source, destination, limit, title, events=None):
        super().__init__(name, destination, limit, title, events)

        self.source = Path(source)

    def backup(self):
        self.remove_redundant()

        status_code = os.system(
            "tar -C '{}' -czf '{}' -- .".format(
                self.source,
                self.destination / self.get_filename()
            )
        )

        self.notify(status_code)

    @staticmethod
    def from_dict(name, data):
        return DirSpec(
            name,
            data["source"],
            data["destination"],
            data["limit"],
            data["notify"]["title"],
            data["notify"]["events"]
        )

class VolumeSpec(Spec):
    def __init__(self, name, volume, destination, limit, title, events=None):
        super().__init__(name, destination, limit, title, events)

        self.volume = volume

    def backup(self):
        status_code = os.system(
            "docker run --rm -v '{}:/from' -v '{}:/to' alpine:latest "
            "tar -C /from -czf '/to/{}' -- .".format(
                self.volume,
                self.destination,
                self.get_filename()
            )
        )

    @staticmethod
    def from_dict(name, data):
        return VolumeSpec(
            name,
            data["source"],
            data["destination"],
            data["limit"],
            data["notify"]["title"],
            data["notify"]["events"]
        )