""" 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