msa-django-nhn/nhn/packages/compute.py

"""
NHN Cloud Compute API Module

인스턴스, 이미지, Flavor, Keypair 관리
"""

import logging
from typing import Optional, List

from .base import BaseAPI, NHNCloudEndpoints, Region

logger = logging.getLogger(__name__)


class ApiCompute(BaseAPI):
    """NHN Cloud Compute API 클래스"""

    def __init__(self, region: str, tenant_id: str, token: str):
        """
        Args:
            region: 리전 (kr1: 판교, kr2: 평촌)
            tenant_id: 테넌트 ID
            token: API 인증 토큰
        """
        super().__init__(region, token)
        self.tenant_id = tenant_id

        if self.region == Region.KR1:
            self.compute_url = NHNCloudEndpoints.COMPUTE_KR1
            self.image_url = NHNCloudEndpoints.IMAGE_KR1
        else:
            self.compute_url = NHNCloudEndpoints.COMPUTE_KR2
            self.image_url = NHNCloudEndpoints.IMAGE_KR2

    # ==================== Flavor ====================

    def get_flavor_list(self) -> dict:
        """Flavor 목록 조회"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/flavors"
        return self._get(url)

    def get_flavor_detail(self, flavor_id: str) -> dict:
        """Flavor 상세 조회"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/flavors/{flavor_id}"
        return self._get(url)

    def get_flavor_id_by_name(self, flavor_name: str) -> Optional[str]:
        """Flavor 이름으로 ID 조회"""
        data = self.get_flavor_list()
        flavors = data.get("flavors", [])

        for flavor in flavors:
            if flavor_name in flavor.get("name", ""):
                return flavor.get("id")
        return None

    # ==================== Keypair ====================

    def get_keypair_list(self) -> dict:
        """Keypair 목록 조회"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/os-keypairs"
        return self._get(url)

    def get_keypair_info(self, keypair_name: str) -> dict:
        """Keypair 상세 조회"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/os-keypairs/{keypair_name}"
        return self._get(url)

    # ==================== Instance ====================

    def get_instance_list(self) -> dict:
        """인스턴스 목록 조회"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/servers"
        return self._get(url)

    def get_instance_list_detail(self) -> dict:
        """인스턴스 상세 목록 조회"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/servers/detail"
        return self._get(url)

    def get_instance_info(self, server_id: str) -> dict:
        """인스턴스 상세 정보 조회"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/servers/{server_id}"
        return self._get(url)

    def get_instance_status(self, server_id: str) -> str:
        """인스턴스 상태 조회"""
        data = self.get_instance_info(server_id)
        return data.get("server", {}).get("status", "UNKNOWN")

    def get_instance_id_by_name(self, instance_name: str) -> Optional[str]:
        """인스턴스 이름으로 ID 조회"""
        data = self.get_instance_list()
        servers = data.get("servers", [])

        for server in servers:
            if server.get("name") == instance_name:
                return server.get("id")
        return None

    def create_instance(
        self,
        name: str,
        image_id: str,
        flavor_id: str,
        subnet_id: str,
        keypair_name: str,
        volume_size: int = 50,
        volume_type: str = "General SSD",
        security_groups: Optional[List[str]] = None,
        availability_zone: Optional[str] = None,
    ) -> dict:
        """
        인스턴스 생성

        Args:
            name: 인스턴스 이름
            image_id: 이미지 ID
            flavor_id: Flavor ID
            subnet_id: 서브넷 ID
            keypair_name: Keypair 이름
            volume_size: 볼륨 크기 (GB, 기본 50)
            volume_type: 볼륨 타입 (General SSD, General HDD)
            security_groups: 보안 그룹 목록 (기본 ["default"])
            availability_zone: 가용 영역

        Returns:
            dict: 생성된 인스턴스 정보
        """
        url = f"{self.compute_url}/v2/{self.tenant_id}/servers"

        security_groups = security_groups or ["default"]
        sg_list = [{"name": sg} for sg in security_groups]

        payload = {
            "server": {
                "name": name,
                "imageRef": image_id,
                "flavorRef": flavor_id,
                "networks": [{"subnet": subnet_id}],
                "key_name": keypair_name,
                "max_count": 1,
                "min_count": 1,
                "block_device_mapping_v2": [
                    {
                        "uuid": image_id,
                        "boot_index": 0,
                        "volume_size": volume_size,
                        "volume_type": volume_type,
                        "device_name": "vda",
                        "source_type": "image",
                        "destination_type": "volume",
                        "delete_on_termination": True,
                    }
                ],
                "security_groups": sg_list,
            }
        }

        if availability_zone:
            payload["server"]["availability_zone"] = availability_zone

        logger.info(f"인스턴스 생성 요청: {name}")
        return self._post(url, payload)

    def delete_instance(self, server_id: str) -> dict:
        """인스턴스 삭제"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/servers/{server_id}"
        logger.info(f"인스턴스 삭제 요청: {server_id}")
        return self._delete(url)

    def start_instance(self, server_id: str) -> dict:
        """인스턴스 시작"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/servers/{server_id}/action"
        return self._post(url, {"os-start": None})

    def stop_instance(self, server_id: str) -> dict:
        """인스턴스 정지"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/servers/{server_id}/action"
        return self._post(url, {"os-stop": None})

    def reboot_instance(self, server_id: str, hard: bool = False) -> dict:
        """인스턴스 재부팅"""
        url = f"{self.compute_url}/v2/{self.tenant_id}/servers/{server_id}/action"
        reboot_type = "HARD" if hard else "SOFT"
        return self._post(url, {"reboot": {"type": reboot_type}})

    # ==================== Image ====================

    def get_image_list(self) -> dict:
        """이미지 목록 조회"""
        url = f"{self.image_url}/v2/images"
        return self._get(url)

    def get_image_info(self, image_id: str) -> dict:
        """이미지 상세 조회"""
        url = f"{self.image_url}/v2/images/{image_id}"
        return self._get(url)

    def get_image_id_by_name(self, image_name: str, exclude_container: bool = True) -> Optional[str]:
        """
        이미지 이름으로 ID 조회

        Args:
            image_name: 이미지 이름 (부분 일치)
            exclude_container: 컨테이너 이미지 제외 여부

        Returns:
            str: 이미지 ID 또는 None
        """
        data = self.get_image_list()
        images = data.get("images", [])

        for image in images:
            name = image.get("name", "")
            if name.startswith(image_name):
                if exclude_container and "Container" in name:
                    continue
                return image.get("id")
        return None