""" NHN Cloud DNS Plus API Module DNS Zone, 레코드 세트, GSLB, Pool, Health Check 관리 """ import logging from typing import Optional, List import requests from .base import NHNCloudEndpoints, NHNCloudAPIError logger = logging.getLogger(__name__) class ApiDnsPlus: """NHN Cloud DNS Plus API 클래스""" DEFAULT_TIMEOUT = 30 def __init__(self, appkey: str): """ Args: appkey: NHN Cloud 앱키 """ self.appkey = appkey self.base_url = f"{NHNCloudEndpoints.DNSPLUS}/dnsplus/v1.0/appkeys/{appkey}" self._session = requests.Session() def _get_headers(self) -> dict: """DNS Plus API 헤더""" return { "Content-Type": "application/json", "Accept": "application/json", } def _request( self, method: str, url: str, params: Optional[dict] = None, json_data: Optional[dict] = None, timeout: Optional[int] = None, ) -> dict: """HTTP 요청 실행""" logger.info(f"[DnsPlus] 요청 시작 - method={method}, url={url}") if params: logger.info(f"[DnsPlus] 요청 파라미터 - params={params}") if json_data: logger.info(f"[DnsPlus] 요청 바디 - json={json_data}") try: response = self._session.request( method=method, url=url, params=params, json=json_data, headers=self._get_headers(), timeout=timeout or self.DEFAULT_TIMEOUT, ) logger.info(f"[DnsPlus] 응답 수신 - status_code={response.status_code}") if response.text: result = response.json() # DNS Plus API는 header.isSuccessful로 성공 여부 판단 if "header" in result and not result["header"].get("isSuccessful", True): error_msg = result["header"].get("resultMessage", "알 수 없는 오류") error_code = result["header"].get("resultCode", response.status_code) logger.error(f"[DnsPlus] API 오류 - code={error_code}, message={error_msg}") raise NHNCloudAPIError(message=error_msg, code=error_code, details=result) return result return {} except requests.exceptions.Timeout: logger.error(f"[DnsPlus] 타임아웃 - url={url}") raise NHNCloudAPIError("요청 시간이 초과되었습니다.", code=408) except requests.exceptions.ConnectionError as e: logger.error(f"[DnsPlus] 연결 오류 - url={url}, error={e}") raise NHNCloudAPIError("서버에 연결할 수 없습니다.", code=503) except requests.exceptions.RequestException as e: logger.error(f"[DnsPlus] 요청 오류 - url={url}, error={e}") raise NHNCloudAPIError(f"요청 중 오류가 발생했습니다: {e}") def _get(self, url: str, params: Optional[dict] = None) -> dict: return self._request("GET", url, params=params) def _post(self, url: str, json_data: Optional[dict] = None) -> dict: return self._request("POST", url, json_data=json_data) def _put(self, url: str, json_data: Optional[dict] = None) -> dict: return self._request("PUT", url, json_data=json_data) def _delete(self, url: str, params: Optional[dict] = None) -> dict: return self._request("DELETE", url, params=params) # ==================== DNS Zone ==================== def get_zone_list( self, zone_id_list: Optional[List[str]] = None, zone_status_list: Optional[List[str]] = None, search_zone_name: Optional[str] = None, page: int = 1, limit: int = 50, ) -> dict: """ DNS Zone 목록 조회 Args: zone_id_list: Zone ID 목록 zone_status_list: Zone 상태 목록 (USE, DISUSE) search_zone_name: 검색할 Zone 이름 page: 페이지 번호 limit: 페이지당 항목 수 Returns: dict: Zone 목록 """ url = f"{self.base_url}/zones" params = {"page": page, "limit": limit} if zone_id_list: params["zoneIdList"] = ",".join(zone_id_list) if zone_status_list: params["zoneStatusList"] = ",".join(zone_status_list) if search_zone_name: params["searchZoneName"] = search_zone_name return self._get(url, params) def get_zone(self, zone_id: str) -> dict: """ DNS Zone 상세 조회 Args: zone_id: Zone ID Returns: dict: Zone 상세 정보 """ url = f"{self.base_url}/zones" params = {"zoneIdList": zone_id} result = self._get(url, params) zones = result.get("zoneList", []) if zones: return {"zone": zones[0]} raise NHNCloudAPIError(f"Zone을 찾을 수 없습니다: {zone_id}", code=404) def create_zone(self, zone_name: str, description: Optional[str] = None) -> dict: """ DNS Zone 생성 Args: zone_name: Zone 이름 (도메인, 예: example.com.) description: 설명 Returns: dict: 생성된 Zone 정보 """ url = f"{self.base_url}/zones" payload = { "zone": { "zoneName": zone_name, } } if description: payload["zone"]["description"] = description logger.info(f"DNS Zone 생성 요청: {zone_name}") return self._post(url, payload) def update_zone(self, zone_id: str, description: str) -> dict: """ DNS Zone 수정 Args: zone_id: Zone ID description: 설명 Returns: dict: 수정된 Zone 정보 """ url = f"{self.base_url}/zones/{zone_id}" payload = { "zone": { "description": description, } } logger.info(f"DNS Zone 수정 요청: {zone_id}") return self._put(url, payload) def delete_zones(self, zone_id_list: List[str]) -> dict: """ DNS Zone 삭제 (비동기) Args: zone_id_list: 삭제할 Zone ID 목록 Returns: dict: 삭제 결과 """ url = f"{self.base_url}/zones/async" params = {"zoneIdList": ",".join(zone_id_list)} logger.info(f"DNS Zone 삭제 요청: {zone_id_list}") return self._delete(url, params) # ==================== Record Set ==================== def get_recordset_list( self, zone_id: str, recordset_id_list: Optional[List[str]] = None, recordset_type_list: Optional[List[str]] = None, search_recordset_name: Optional[str] = None, page: int = 1, limit: int = 50, ) -> dict: """ 레코드 세트 목록 조회 Args: zone_id: Zone ID recordset_id_list: 레코드 세트 ID 목록 recordset_type_list: 레코드 타입 목록 (A, AAAA, CNAME, MX, TXT, NS, SRV, etc.) search_recordset_name: 검색할 레코드 세트 이름 page: 페이지 번호 limit: 페이지당 항목 수 Returns: dict: 레코드 세트 목록 """ url = f"{self.base_url}/zones/{zone_id}/recordsets" params = {"page": page, "limit": limit} if recordset_id_list: params["recordsetIdList"] = ",".join(recordset_id_list) if recordset_type_list: params["recordsetTypeList"] = ",".join(recordset_type_list) if search_recordset_name: params["searchRecordsetName"] = search_recordset_name return self._get(url, params) def get_recordset(self, zone_id: str, recordset_id: str) -> dict: """ 레코드 세트 상세 조회 Args: zone_id: Zone ID recordset_id: 레코드 세트 ID Returns: dict: 레코드 세트 상세 정보 """ url = f"{self.base_url}/zones/{zone_id}/recordsets" params = {"recordsetIdList": recordset_id} result = self._get(url, params) recordsets = result.get("recordsetList", []) if recordsets: return {"recordset": recordsets[0]} raise NHNCloudAPIError(f"레코드 세트를 찾을 수 없습니다: {recordset_id}", code=404) def create_recordset( self, zone_id: str, recordset_name: str, recordset_type: str, recordset_ttl: int, record_list: List[dict], ) -> dict: """ 레코드 세트 생성 Args: zone_id: Zone ID recordset_name: 레코드 세트 이름 recordset_type: 레코드 타입 (A, AAAA, CNAME, MX, TXT, NS, SRV, etc.) recordset_ttl: TTL (초) record_list: 레코드 목록 [{"recordContent": "...", "recordDisabled": False}] Returns: dict: 생성된 레코드 세트 정보 """ url = f"{self.base_url}/zones/{zone_id}/recordsets" payload = { "recordset": { "recordsetName": recordset_name, "recordsetType": recordset_type, "recordsetTtl": recordset_ttl, "recordList": record_list, } } logger.info(f"레코드 세트 생성 요청: {recordset_name} ({recordset_type})") return self._post(url, payload) def create_recordset_bulk( self, zone_id: str, recordset_list: List[dict], ) -> dict: """ 레코드 세트 대량 생성 Args: zone_id: Zone ID recordset_list: 레코드 세트 목록 (최대 2,000개) Returns: dict: 생성된 레코드 세트 정보 """ url = f"{self.base_url}/zones/{zone_id}/recordsets/list" payload = {"recordsetList": recordset_list} logger.info(f"레코드 세트 대량 생성 요청: {len(recordset_list)}개") return self._post(url, payload) def update_recordset( self, zone_id: str, recordset_id: str, recordset_type: str, recordset_ttl: int, record_list: List[dict], ) -> dict: """ 레코드 세트 수정 Args: zone_id: Zone ID recordset_id: 레코드 세트 ID recordset_type: 레코드 타입 recordset_ttl: TTL (초) record_list: 레코드 목록 Returns: dict: 수정된 레코드 세트 정보 """ url = f"{self.base_url}/zones/{zone_id}/recordsets/{recordset_id}" payload = { "recordset": { "recordsetType": recordset_type, "recordsetTtl": recordset_ttl, "recordList": record_list, } } logger.info(f"레코드 세트 수정 요청: {recordset_id}") return self._put(url, payload) def delete_recordsets(self, zone_id: str, recordset_id_list: List[str]) -> dict: """ 레코드 세트 삭제 Args: zone_id: Zone ID recordset_id_list: 삭제할 레코드 세트 ID 목록 Returns: dict: 삭제 결과 """ url = f"{self.base_url}/zones/{zone_id}/recordsets" params = {"recordsetIdList": ",".join(recordset_id_list)} logger.info(f"레코드 세트 삭제 요청: {recordset_id_list}") return self._delete(url, params) # ==================== GSLB ==================== def get_gslb_list( self, gslb_id_list: Optional[List[str]] = None, search_gslb_name: Optional[str] = None, gslb_domain: Optional[str] = None, page: int = 1, limit: int = 50, ) -> dict: """ GSLB 목록 조회 Args: gslb_id_list: GSLB ID 목록 search_gslb_name: 검색할 GSLB 이름 gslb_domain: GSLB 도메인 page: 페이지 번호 limit: 페이지당 항목 수 Returns: dict: GSLB 목록 """ url = f"{self.base_url}/gslbs" params = {"page": page, "limit": limit} if gslb_id_list: params["gslbIdList"] = ",".join(gslb_id_list) if search_gslb_name: params["searchGslbName"] = search_gslb_name if gslb_domain: params["gslbDomain"] = gslb_domain return self._get(url, params) def get_gslb(self, gslb_id: str) -> dict: """ GSLB 상세 조회 Args: gslb_id: GSLB ID Returns: dict: GSLB 상세 정보 """ url = f"{self.base_url}/gslbs" params = {"gslbIdList": gslb_id} result = self._get(url, params) gslbs = result.get("gslbList", []) if gslbs: return {"gslb": gslbs[0]} raise NHNCloudAPIError(f"GSLB를 찾을 수 없습니다: {gslb_id}", code=404) def create_gslb( self, gslb_name: str, gslb_ttl: int, gslb_routing_rule: str, gslb_disabled: bool = False, connected_pool_list: Optional[List[dict]] = None, ) -> dict: """ GSLB 생성 Args: gslb_name: GSLB 이름 gslb_ttl: TTL (초) gslb_routing_rule: 라우팅 규칙 (FAILOVER, RANDOM, GEOLOCATION) gslb_disabled: 비활성화 여부 connected_pool_list: 연결된 Pool 목록 Returns: dict: 생성된 GSLB 정보 """ url = f"{self.base_url}/gslbs" payload = { "gslb": { "gslbName": gslb_name, "gslbTtl": gslb_ttl, "gslbRoutingRule": gslb_routing_rule, "gslbDisabled": gslb_disabled, } } if connected_pool_list: payload["gslb"]["connectedPoolList"] = connected_pool_list logger.info(f"GSLB 생성 요청: {gslb_name}") return self._post(url, payload) def update_gslb( self, gslb_id: str, gslb_name: Optional[str] = None, gslb_ttl: Optional[int] = None, gslb_routing_rule: Optional[str] = None, gslb_disabled: Optional[bool] = None, connected_pool_list: Optional[List[dict]] = None, ) -> dict: """ GSLB 수정 Args: gslb_id: GSLB ID gslb_name: GSLB 이름 gslb_ttl: TTL (초) gslb_routing_rule: 라우팅 규칙 gslb_disabled: 비활성화 여부 connected_pool_list: 연결된 Pool 목록 Returns: dict: 수정된 GSLB 정보 """ url = f"{self.base_url}/gslbs/{gslb_id}" gslb_data = {} if gslb_name is not None: gslb_data["gslbName"] = gslb_name if gslb_ttl is not None: gslb_data["gslbTtl"] = gslb_ttl if gslb_routing_rule is not None: gslb_data["gslbRoutingRule"] = gslb_routing_rule if gslb_disabled is not None: gslb_data["gslbDisabled"] = gslb_disabled if connected_pool_list is not None: gslb_data["connectedPoolList"] = connected_pool_list payload = {"gslb": gslb_data} logger.info(f"GSLB 수정 요청: {gslb_id}") return self._put(url, payload) def delete_gslbs(self, gslb_id_list: List[str]) -> dict: """ GSLB 삭제 Args: gslb_id_list: 삭제할 GSLB ID 목록 Returns: dict: 삭제 결과 """ url = f"{self.base_url}/gslbs" params = {"gslbIdList": ",".join(gslb_id_list)} logger.info(f"GSLB 삭제 요청: {gslb_id_list}") return self._delete(url, params) def connect_pool( self, gslb_id: str, pool_id: str, connected_pool_order: int, connected_pool_region_content: Optional[str] = None, ) -> dict: """ GSLB에 Pool 연결 Args: gslb_id: GSLB ID pool_id: Pool ID connected_pool_order: 연결 순서 (우선순위) connected_pool_region_content: 지역 콘텐츠 (GEOLOCATION 규칙 사용 시) Returns: dict: 연결 결과 """ url = f"{self.base_url}/gslbs/{gslb_id}/connected-pools/{pool_id}" payload = { "connectedPool": { "connectedPoolOrder": connected_pool_order, } } if connected_pool_region_content: payload["connectedPool"]["connectedPoolRegionContent"] = connected_pool_region_content logger.info(f"Pool 연결 요청: GSLB={gslb_id}, Pool={pool_id}") return self._post(url, payload) def update_connected_pool( self, gslb_id: str, pool_id: str, connected_pool_order: int, connected_pool_region_content: Optional[str] = None, ) -> dict: """ 연결된 Pool 수정 Args: gslb_id: GSLB ID pool_id: Pool ID connected_pool_order: 연결 순서 (우선순위) connected_pool_region_content: 지역 콘텐츠 Returns: dict: 수정 결과 """ url = f"{self.base_url}/gslbs/{gslb_id}/connected-pools/{pool_id}" payload = { "connectedPool": { "connectedPoolOrder": connected_pool_order, } } if connected_pool_region_content: payload["connectedPool"]["connectedPoolRegionContent"] = connected_pool_region_content logger.info(f"연결된 Pool 수정 요청: GSLB={gslb_id}, Pool={pool_id}") return self._put(url, payload) def disconnect_pools(self, gslb_id: str, pool_id_list: List[str]) -> dict: """ GSLB에서 Pool 연결 해제 Args: gslb_id: GSLB ID pool_id_list: 연결 해제할 Pool ID 목록 Returns: dict: 연결 해제 결과 """ url = f"{self.base_url}/gslbs/{gslb_id}/connected-pools" params = {"poolIdList": ",".join(pool_id_list)} logger.info(f"Pool 연결 해제 요청: GSLB={gslb_id}, Pools={pool_id_list}") return self._delete(url, params) # ==================== Pool ==================== def get_pool_list( self, pool_id_list: Optional[List[str]] = None, search_pool_name: Optional[str] = None, health_check_id: Optional[str] = None, page: int = 1, limit: int = 50, ) -> dict: """ Pool 목록 조회 Args: pool_id_list: Pool ID 목록 search_pool_name: 검색할 Pool 이름 health_check_id: Health Check ID page: 페이지 번호 limit: 페이지당 항목 수 Returns: dict: Pool 목록 """ url = f"{self.base_url}/pools" params = {"page": page, "limit": limit} if pool_id_list: params["poolIdList"] = ",".join(pool_id_list) if search_pool_name: params["searchPoolName"] = search_pool_name if health_check_id: params["healthCheckId"] = health_check_id return self._get(url, params) def get_pool(self, pool_id: str) -> dict: """ Pool 상세 조회 Args: pool_id: Pool ID Returns: dict: Pool 상세 정보 """ url = f"{self.base_url}/pools" params = {"poolIdList": pool_id} result = self._get(url, params) pools = result.get("poolList", []) if pools: return {"pool": pools[0]} raise NHNCloudAPIError(f"Pool을 찾을 수 없습니다: {pool_id}", code=404) def create_pool( self, pool_name: str, endpoint_list: List[dict], pool_disabled: bool = False, health_check_id: Optional[str] = None, ) -> dict: """ Pool 생성 Args: pool_name: Pool 이름 endpoint_list: 엔드포인트 목록 [{"endpointAddress": "1.1.1.1", "endpointWeight": 1.0, "endpointDisabled": False}] pool_disabled: 비활성화 여부 health_check_id: Health Check ID Returns: dict: 생성된 Pool 정보 """ url = f"{self.base_url}/pools" payload = { "pool": { "poolName": pool_name, "poolDisabled": pool_disabled, "endpointList": endpoint_list, } } if health_check_id: payload["pool"]["healthCheckId"] = health_check_id logger.info(f"Pool 생성 요청: {pool_name}") return self._post(url, payload) def update_pool( self, pool_id: str, pool_name: Optional[str] = None, endpoint_list: Optional[List[dict]] = None, pool_disabled: Optional[bool] = None, health_check_id: Optional[str] = None, ) -> dict: """ Pool 수정 Args: pool_id: Pool ID pool_name: Pool 이름 endpoint_list: 엔드포인트 목록 pool_disabled: 비활성화 여부 health_check_id: Health Check ID Returns: dict: 수정된 Pool 정보 """ url = f"{self.base_url}/pools/{pool_id}" pool_data = {} if pool_name is not None: pool_data["poolName"] = pool_name if endpoint_list is not None: pool_data["endpointList"] = endpoint_list if pool_disabled is not None: pool_data["poolDisabled"] = pool_disabled if health_check_id is not None: pool_data["healthCheckId"] = health_check_id payload = {"pool": pool_data} logger.info(f"Pool 수정 요청: {pool_id}") return self._put(url, payload) def delete_pools(self, pool_id_list: List[str]) -> dict: """ Pool 삭제 Args: pool_id_list: 삭제할 Pool ID 목록 Returns: dict: 삭제 결과 """ url = f"{self.base_url}/pools" params = {"poolIdList": ",".join(pool_id_list)} logger.info(f"Pool 삭제 요청: {pool_id_list}") return self._delete(url, params) # ==================== Health Check ==================== def get_health_check_list( self, health_check_id_list: Optional[List[str]] = None, search_health_check_name: Optional[str] = None, page: int = 1, limit: int = 50, ) -> dict: """ Health Check 목록 조회 Args: health_check_id_list: Health Check ID 목록 search_health_check_name: 검색할 Health Check 이름 page: 페이지 번호 limit: 페이지당 항목 수 Returns: dict: Health Check 목록 """ url = f"{self.base_url}/health-checks" params = {"page": page, "limit": limit} if health_check_id_list: params["healthCheckIdList"] = ",".join(health_check_id_list) if search_health_check_name: params["searchHealthCheckName"] = search_health_check_name return self._get(url, params) def get_health_check(self, health_check_id: str) -> dict: """ Health Check 상세 조회 Args: health_check_id: Health Check ID Returns: dict: Health Check 상세 정보 """ url = f"{self.base_url}/health-checks" params = {"healthCheckIdList": health_check_id} result = self._get(url, params) health_checks = result.get("healthCheckList", []) if health_checks: return {"healthCheck": health_checks[0]} raise NHNCloudAPIError(f"Health Check를 찾을 수 없습니다: {health_check_id}", code=404) def create_health_check( self, health_check_name: str, health_check_protocol: str, health_check_port: int, health_check_interval: int = 30, health_check_timeout: int = 5, health_check_retries: int = 2, health_check_path: Optional[str] = None, health_check_expected_codes: Optional[str] = None, health_check_expected_body: Optional[str] = None, health_check_request_header_list: Optional[List[dict]] = None, ) -> dict: """ Health Check 생성 Args: health_check_name: Health Check 이름 health_check_protocol: 프로토콜 (HTTPS, HTTP, TCP) health_check_port: 포트 health_check_interval: 체크 간격 (초) health_check_timeout: 타임아웃 (초) health_check_retries: 재시도 횟수 health_check_path: 경로 (HTTP/HTTPS 전용) health_check_expected_codes: 예상 응답 코드 (HTTP/HTTPS 전용) health_check_expected_body: 예상 응답 본문 (HTTP/HTTPS 전용) health_check_request_header_list: 요청 헤더 목록 (HTTP/HTTPS 전용) Returns: dict: 생성된 Health Check 정보 """ url = f"{self.base_url}/health-checks" payload = { "healthCheck": { "healthCheckName": health_check_name, "healthCheckProtocol": health_check_protocol, "healthCheckPort": health_check_port, "healthCheckInterval": health_check_interval, "healthCheckTimeout": health_check_timeout, "healthCheckRetries": health_check_retries, } } # HTTP/HTTPS 전용 옵션 if health_check_path: payload["healthCheck"]["healthCheckPath"] = health_check_path if health_check_expected_codes: payload["healthCheck"]["healthCheckExpectedCodes"] = health_check_expected_codes if health_check_expected_body: payload["healthCheck"]["healthCheckExpectedBody"] = health_check_expected_body if health_check_request_header_list: payload["healthCheck"]["healthCheckRequestHeaderList"] = health_check_request_header_list logger.info(f"Health Check 생성 요청: {health_check_name}") return self._post(url, payload) def update_health_check( self, health_check_id: str, health_check_name: Optional[str] = None, health_check_protocol: Optional[str] = None, health_check_port: Optional[int] = None, health_check_interval: Optional[int] = None, health_check_timeout: Optional[int] = None, health_check_retries: Optional[int] = None, health_check_path: Optional[str] = None, health_check_expected_codes: Optional[str] = None, health_check_expected_body: Optional[str] = None, health_check_request_header_list: Optional[List[dict]] = None, ) -> dict: """ Health Check 수정 Args: health_check_id: Health Check ID (나머지 파라미터는 create_health_check과 동일) Returns: dict: 수정된 Health Check 정보 """ url = f"{self.base_url}/health-checks/{health_check_id}" health_check_data = {} if health_check_name is not None: health_check_data["healthCheckName"] = health_check_name if health_check_protocol is not None: health_check_data["healthCheckProtocol"] = health_check_protocol if health_check_port is not None: health_check_data["healthCheckPort"] = health_check_port if health_check_interval is not None: health_check_data["healthCheckInterval"] = health_check_interval if health_check_timeout is not None: health_check_data["healthCheckTimeout"] = health_check_timeout if health_check_retries is not None: health_check_data["healthCheckRetries"] = health_check_retries if health_check_path is not None: health_check_data["healthCheckPath"] = health_check_path if health_check_expected_codes is not None: health_check_data["healthCheckExpectedCodes"] = health_check_expected_codes if health_check_expected_body is not None: health_check_data["healthCheckExpectedBody"] = health_check_expected_body if health_check_request_header_list is not None: health_check_data["healthCheckRequestHeaderList"] = health_check_request_header_list payload = {"healthCheck": health_check_data} logger.info(f"Health Check 수정 요청: {health_check_id}") return self._put(url, payload) def delete_health_checks(self, health_check_id_list: List[str]) -> dict: """ Health Check 삭제 Args: health_check_id_list: 삭제할 Health Check ID 목록 Returns: dict: 삭제 결과 """ url = f"{self.base_url}/health-checks" params = {"healthCheckIdList": ",".join(health_check_id_list)} logger.info(f"Health Check 삭제 요청: {health_check_id_list}") return self._delete(url, params)