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

"""
NHN Cloud VPC API Module

VPC, Subnet, Routing Table, Floating IP 관리
"""

import logging
from typing import Optional

from .base import BaseAPI, NHNCloudEndpoints, Region

logger = logging.getLogger(__name__)


class ApiVpc(BaseAPI):
    """NHN Cloud VPC API 클래스"""

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

        if self.region == Region.KR1:
            self.vpc_url = NHNCloudEndpoints.NETWORK_KR1
        else:
            self.vpc_url = NHNCloudEndpoints.NETWORK_KR2

    # ==================== VPC ====================

    def get_vpc_list(self) -> dict:
        """VPC 목록 조회"""
        url = f"{self.vpc_url}/v2.0/vpcs"
        logger.info(f"[NHN API] VPC 목록 조회 요청 - URL={url}")
        result = self._get(url)
        vpc_count = len(result.get("vpcs", []))
        logger.info(f"[NHN API] VPC 목록 조회 완료 - count={vpc_count}")
        return result

    def get_vpc_info(self, vpc_id: str) -> dict:
        """VPC 상세 조회"""
        url = f"{self.vpc_url}/v2.0/vpcs/{vpc_id}"
        logger.info(f"[NHN API] VPC 상세 조회 요청 - URL={url}, vpc_id={vpc_id}")
        result = self._get(url)
        logger.info(f"[NHN API] VPC 상세 조회 완료 - vpc_id={vpc_id}")
        return result

    def get_vpc_id_by_name(self, vpc_name: str) -> Optional[str]:
        """VPC 이름으로 ID 조회"""
        data = self.get_vpc_list()
        vpcs = data.get("vpcs", [])

        for vpc in vpcs:
            if vpc.get("name", "").startswith(vpc_name):
                return vpc.get("id")
        return None

    def create_vpc(self, name: str, cidr: str) -> dict:
        """
        VPC 생성

        Args:
            name: VPC 이름
            cidr: CIDR 블록 (예: 10.0.0.0/16)

        Returns:
            dict: 생성된 VPC 정보
        """
        url = f"{self.vpc_url}/v2.0/vpcs"
        payload = {"vpc": {"name": name, "cidrv4": cidr}}
        logger.info(f"VPC 생성 요청: {name} ({cidr})")
        return self._post(url, payload)

    def delete_vpc(self, vpc_id: str) -> dict:
        """VPC 삭제"""
        url = f"{self.vpc_url}/v2.0/vpcs/{vpc_id}"
        logger.info(f"VPC 삭제 요청: {vpc_id}")
        return self._delete(url)

    # ==================== Subnet ====================

    def get_subnet_list(self) -> dict:
        """서브넷 목록 조회"""
        url = f"{self.vpc_url}/v2.0/vpcsubnets"
        return self._get(url)

    def get_subnet_info(self, subnet_id: str) -> dict:
        """서브넷 상세 조회"""
        url = f"{self.vpc_url}/v2.0/vpcsubnets/{subnet_id}"
        return self._get(url)

    def get_subnet_id_by_name(self, subnet_name: str) -> Optional[str]:
        """서브넷 이름으로 ID 조회"""
        data = self.get_subnet_list()
        subnets = data.get("vpcsubnets", [])

        for subnet in subnets:
            name = subnet.get("name", "")
            if name.startswith(subnet_name) and "rt" not in name:
                return subnet.get("id")
        return None

    def create_subnet(self, vpc_id: str, cidr: str, name: str) -> dict:
        """
        서브넷 생성

        Args:
            vpc_id: VPC ID
            cidr: CIDR 블록 (예: 10.0.1.0/24)
            name: 서브넷 이름

        Returns:
            dict: 생성된 서브넷 정보
        """
        url = f"{self.vpc_url}/v2.0/vpcsubnets"
        payload = {"vpcsubnet": {"vpc_id": vpc_id, "cidr": cidr, "name": name}}
        logger.info(f"서브넷 생성 요청: {name} ({cidr})")
        return self._post(url, payload)

    def delete_subnet(self, subnet_id: str) -> dict:
        """서브넷 삭제"""
        url = f"{self.vpc_url}/v2.0/vpcsubnets/{subnet_id}"
        logger.info(f"서브넷 삭제 요청: {subnet_id}")
        return self._delete(url)

    # ==================== Routing Table ====================

    def get_routing_table_list(self, detail: bool = True) -> dict:
        """라우팅 테이블 목록 조회"""
        url = f"{self.vpc_url}/v2.0/routingtables"
        params = {"detail": "true"} if detail else None
        return self._get(url, params=params)

    def get_routing_table_info(self, routingtable_id: str) -> dict:
        """라우팅 테이블 상세 조회"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}"
        return self._get(url)

    def get_default_routing_table_id(self) -> Optional[str]:
        """기본 라우팅 테이블 ID 조회"""
        data = self.get_routing_table_list(detail=False)
        tables = data.get("routingtables", [])
        return tables[0].get("id") if tables else None

    def create_routing_table(self, name: str, vpc_id: str, distributed: bool = True) -> dict:
        """
        라우팅 테이블 생성

        Args:
            name: 라우팅 테이블 이름
            vpc_id: VPC ID
            distributed: 분산 라우팅 여부 (기본 True)

        Returns:
            dict: 생성된 라우팅 테이블 정보
        """
        url = f"{self.vpc_url}/v2.0/routingtables"
        payload = {
            "routingtable": {
                "name": name,
                "vpc_id": vpc_id,
                "distributed": str(distributed).lower(),
            }
        }
        logger.info(f"라우팅 테이블 생성 요청: {name}")
        return self._post(url, payload)

    def delete_routing_table(self, routingtable_id: str) -> dict:
        """라우팅 테이블 삭제"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}"
        logger.info(f"라우팅 테이블 삭제 요청: {routingtable_id}")
        return self._delete(url)

    def set_default_routing_table(self, routingtable_id: str) -> dict:
        """라우팅 테이블을 기본으로 설정"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}/set_as_default"
        return self._put(url)

    def attach_gateway_to_routing_table(self, routingtable_id: str, gateway_id: str) -> dict:
        """라우팅 테이블에 인터넷 게이트웨이 연결"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}/attach_gateway"
        payload = {"gateway_id": gateway_id}
        logger.info(f"게이트웨이 연결 요청: {routingtable_id} -> {gateway_id}")
        return self._put(url, payload)

    def detach_gateway_from_routing_table(self, routingtable_id: str) -> dict:
        """라우팅 테이블에서 인터넷 게이트웨이 분리"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}/detach_gateway"
        return self._put(url)

    def attach_routing_table_to_subnet(self, subnet_id: str, routingtable_id: str) -> dict:
        """서브넷에 라우팅 테이블 연결"""
        url = f"{self.vpc_url}/v2.0/vpcsubnets/{subnet_id}/attach_routingtable"
        payload = {"routingtable_id": routingtable_id}
        return self._put(url, payload)

    # ==================== Floating IP ====================

    def get_external_network_id(self) -> dict:
        """외부 네트워크 ID 조회"""
        url = f"{self.vpc_url}/v2.0/networks"
        params = {"router:external": "True"}
        return self._get(url, params=params)

    def get_floating_ip_list(self) -> dict:
        """Floating IP 목록 조회"""
        url = f"{self.vpc_url}/v2.0/floatingips"
        return self._get(url)

    def create_floating_ip(self, external_network_id: str) -> dict:
        """
        Floating IP 생성

        Args:
            external_network_id: 외부 네트워크 ID

        Returns:
            dict: 생성된 Floating IP 정보
        """
        url = f"{self.vpc_url}/v2.0/floatingips"
        payload = {"floatingip": {"floating_network_id": external_network_id}}
        logger.info("Floating IP 생성 요청")
        return self._post(url, payload)

    def delete_floating_ip(self, floating_ip_id: str) -> dict:
        """Floating IP 삭제"""
        url = f"{self.vpc_url}/v2.0/floatingips/{floating_ip_id}"
        logger.info(f"Floating IP 삭제 요청: {floating_ip_id}")
        return self._delete(url)

    def attach_floating_ip(self, floating_ip_id: str, port_id: str) -> dict:
        """Floating IP를 포트에 연결"""
        url = f"{self.vpc_url}/v2.0/floatingips/{floating_ip_id}"
        payload = {"floatingip": {"port_id": port_id}}
        return self._put(url, payload)

    def detach_floating_ip(self, floating_ip_id: str) -> dict:
        """Floating IP 연결 해제"""
        url = f"{self.vpc_url}/v2.0/floatingips/{floating_ip_id}"
        payload = {"floatingip": {"port_id": None}}
        return self._put(url, payload)

    # ==================== Security Group ====================

    def get_security_group_list(self) -> dict:
        """보안 그룹 목록 조회"""
        url = f"{self.vpc_url}/v2.0/security-groups"
        return self._get(url)

    def get_security_group_info(self, security_group_id: str) -> dict:
        """보안 그룹 상세 조회"""
        url = f"{self.vpc_url}/v2.0/security-groups/{security_group_id}"
        return self._get(url)

    # ==================== Port (NIC) ====================

    def get_port_list(self) -> dict:
        """포트 목록 조회"""
        url = f"{self.vpc_url}/v2.0/ports"
        return self._get(url)

    def get_port_id_by_device(self, device_id: str) -> Optional[str]:
        """디바이스 ID로 포트 ID 조회"""
        data = self.get_port_list()
        ports = data.get("ports", [])

        for port in ports:
            if port.get("device_id", "").startswith(device_id):
                return port.get("id")
        return None