selfprivacy-rest-api/selfprivacy_api/backup/__init__.py

from typing import List

from selfprivacy_api.models.backup.snapshot import Snapshot

from selfprivacy_api.utils.singleton_metaclass import SingletonMetaclass
from selfprivacy_api.utils import ReadUserData

from selfprivacy_api.services import get_service_by_id
from selfprivacy_api.services.service import Service

from selfprivacy_api.backup.providers.provider import AbstractBackupProvider
from selfprivacy_api.backup.providers import get_provider
from selfprivacy_api.graphql.queries.providers import BackupProvider


# Singleton has a property of being persistent between tests.
# I don't know what to do with this yet
# class Backups(metaclass=SingletonMetaclass):
class Backups:
    """A singleton controller for backups"""

    provider: AbstractBackupProvider

    def __init__(self, test_repo_file: str = ""):
        if test_repo_file != "":
            self.set_localfile_repo(test_repo_file)
        else:
            self.provider = self.lookup_provider()

    def set_localfile_repo(self, file_path: str):
        ProviderClass = get_provider(BackupProvider.FILE)
        provider = ProviderClass(file_path)
        self.provider = provider

    @staticmethod
    def construct_provider(kind: str, login: str, key: str):
        provider_class = get_provider(BackupProvider[kind])
        return provider_class(login=login, key=key)

    def lookup_provider(self) -> AbstractBackupProvider:
        redis_provider = Backups.load_provider_redis()
        if redis_provider is not None:
            return redis_provider

        json_provider = Backups.load_provider_json()
        if json_provider is not None:
            return json_provider

        return Backups.construct_provider("MEMORY", login="", key="")

    @staticmethod
    def load_provider_redis() -> AbstractBackupProvider:
        pass

    @staticmethod
    def load_provider_json() -> AbstractBackupProvider:
        with ReadUserData() as user_data:
            account = ""
            key = ""

            if "backup" not in user_data.keys():
                if "backblaze" in user_data.keys():
                    account = user_data["backblaze"]["accountId"]
                    key = user_data["backblaze"]["accountKey"]
                    provider_string = "BACKBLAZE"
                    return Backups.construct_provider(
                        kind=provider_string, login=account, key=key
                    )
                return None

            account = user_data["backup"]["accountId"]
            key = user_data["backup"]["accountKey"]
            provider_string = user_data["backup"]["provider"]
            return Backups.construct_provider(
                kind=provider_string, login=account, key=key
            )

    def back_up(self, service: Service):
        folder = service.get_location()
        repo_name = service.get_id()

        service.pre_backup()
        self.provider.backuper.start_backup(folder, repo_name)
        service.post_restore()

    def init_repo(self, service: Service):
        repo_name = service.get_id()
        self.provider.backuper.init(repo_name)

    def get_snapshots(self, service: Service) -> List[Snapshot]:
        repo_name = service.get_id()

        return self.provider.backuper.get_snapshots(repo_name)

    def restore_service_from_snapshot(self, service: Service, snapshot_id: str):
        repo_name = service.get_id()
        folder = service.get_location()

        self.provider.backuper.restore_from_backup(repo_name, snapshot_id, folder)

    # Our dummy service is not yet globally registered so this is not testable yet
    def restore_snapshot(self, snapshot: Snapshot):
        self.restore_service_from_snapshot(
            get_service_by_id(snapshot.service_name), snapshot.id
        )

    def service_snapshot_size(self, service: Service, snapshot_id: str) -> float:
        repo_name = service.get_id()
        return self.provider.backuper.restored_size(repo_name, snapshot_id)

    # Our dummy service is not yet globally registered so this is not testable yet
    def snapshot_restored_size(self, snapshot: Snapshot) -> float:
        return self.service_snapshot_size(
            get_service_by_id(snapshot.service_name), snapshot.id
        )