openstack_storage_manager.py

from typing import Optional

from openstack.exceptions import ResourceNotFound

from interface_openstack.interface.storage.storage_manager import StorageManager
from interface_openstack.interface.storage.volume import Volume
from interface_openstack.implementation.utils.connection import OpenStackConnection


class OpenstackStorageManager(StorageManager, OpenStackConnection):

    @staticmethod
    def os_volume_to_rc_volume(os_volume) -> Volume:
        return Volume(
            os_volume.id,
            os_volume.image_id,
            os_volume.size,
            os_volume.is_bootable,
            os_volume.status,
            os_volume.created_at
        )

    def create(self, size) -> Volume:
        os_volume = self.openstack.block_storage.create_volume(
            size=size
        )

        return self.os_volume_to_rc_volume(os_volume)

    def create_from_image(self, id, size, bootable) -> Volume:
        os_volume = self.openstack.block_storage.create_volume(
            image_id=id,
            size=size,
            bootable=bootable
        )

        return self.os_volume_to_rc_volume(os_volume)

    def create_from_snapshot(self, id) -> Volume:
        os_volume = self.openstack.block_storage.create_volume(
            snapshot_id=id
        )

        return self.os_volume_to_rc_volume(os_volume)

    def get(self, id) -> Optional[Volume]:
        try:
            os_volume = self.openstack.block_storage.get_volume(id)
        except ResourceNotFound:
            return None

        return self.os_volume_to_rc_volume(os_volume)

    def delete(self, id):
        try:
            self.openstack.block_storage.delete_volume(id)
        except ResourceNotFound:
            return False

        return True

    def extend(self, id, size) -> bool:
        try:
            self.openstack.block_storage.extend_volume(id, size)
        except ResourceNotFound:
            return False

        return True

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

        for os_volume in self.openstack.block_storage.volumes():
            volumes.append(self.os_volume_to_rc_volume(os_volume))

        return volumes