openstack_snapshot_manager.py

from typing import Optional

from openstack.exceptions import ResourceNotFound

from interface_openstack.interface.storage.snapshot import Snapshot
from interface_openstack.interface.storage.snapshot_manager import SnapshotManager
from interface_openstack.implementation.utils.connection import OpenStackConnection


class OpenstackSnapshotManager(SnapshotManager, OpenStackConnection):

    @staticmethod
    def os_snapshot_to_rc_snapshot(os_snapshot) -> Snapshot:
        return Snapshot(
            os_snapshot.id,
            os_snapshot.volume_id,
            os_snapshot.size,
            os_snapshot.status,
            os_snapshot.created_at
        )

    def create_from_volume(self, id) -> Snapshot:
        os_snapshot = self.openstack.block_storage.create_snapshot(volume_id=id)

        return self.os_snapshot_to_rc_snapshot(os_snapshot)

    def get(self, id) -> Optional[Snapshot]:
        try:
            os_snapshot = self.openstack.block_storage.get_snapshot(id)
        except ResourceNotFound:
            return None

        return self.os_snapshot_to_rc_snapshot(os_snapshot)

    def list(self) -> []:
        snapshots = []

        for os_snapshot in self.openstack.block_storage.snapshots():
            snapshots.append(self.os_snapshot_to_rc_snapshot(os_snapshot))

        return snapshots

    def delete(self, id) -> bool:
        try:
            self.openstack.block_storage.delete_snapshot(id)
        except ResourceNotFound:
            return False

        return True