""" NHN Cloud API Base Module 공통 기능을 제공하는 베이스 클래스 """ import logging from typing import Optional, Any from dataclasses import dataclass from enum import Enum import requests logger = logging.getLogger(__name__) class Region(str, Enum): """NHN Cloud 리전""" KR1 = "kr1" # 판교 KR2 = "kr2" # 평촌 @dataclass class NHNCloudEndpoints: """NHN Cloud API Endpoints""" # Identity IDENTITY = "https://api-identity-infrastructure.nhncloudservice.com" # Compute COMPUTE_KR1 = "https://kr1-api-instance-infrastructure.nhncloudservice.com" COMPUTE_KR2 = "https://kr2-api-instance-infrastructure.nhncloudservice.com" # Image IMAGE_KR1 = "https://kr1-api-image-infrastructure.nhncloudservice.com" IMAGE_KR2 = "https://kr2-api-image-infrastructure.nhncloudservice.com" # Network (VPC) NETWORK_KR1 = "https://kr1-api-network-infrastructure.nhncloudservice.com" NETWORK_KR2 = "https://kr2-api-network-infrastructure.nhncloudservice.com" # Kubernetes (NKS) NKS_KR1 = "https://kr1-api-kubernetes-infrastructure.nhncloudservice.com" NKS_KR2 = "https://kr2-api-kubernetes-infrastructure.nhncloudservice.com" # Object Storage STORAGE_KR1 = "https://kr1-api-object-storage.nhncloudservice.com/v1" STORAGE_KR2 = "https://kr2-api-object-storage.nhncloudservice.com/v1" # DNS Plus (글로벌 서비스 - 리전 무관) DNSPLUS = "https://dnsplus.api.nhncloudservice.com" class NHNCloudAPIError(Exception): """NHN Cloud API 에러""" def __init__(self, message: str, code: Optional[int] = None, details: Optional[dict] = None): self.message = message self.code = code self.details = details or {} super().__init__(self.message) class BaseAPI: """NHN Cloud API 베이스 클래스""" DEFAULT_TIMEOUT = 30 def __init__(self, region: str, token: str): self.region = Region(region.lower()) if isinstance(region, str) else region self.token = token self._session = requests.Session() def _get_headers(self, extra_headers: Optional[dict] = None) -> dict: """기본 헤더 생성""" headers = { "X-Auth-Token": self.token, "Content-Type": "application/json", "Accept": "application/json", } if extra_headers: headers.update(extra_headers) return headers def _request( self, method: str, url: str, params: Optional[dict] = None, json_data: Optional[dict] = None, headers: Optional[dict] = None, timeout: Optional[int] = None, ) -> dict: """HTTP 요청 실행""" # 토큰 앞 8자리만 로깅 (보안) token_preview = self.token[:8] + "..." if self.token else "None" logger.info(f"[BaseAPI] 요청 시작 - method={method}, url={url}, token={token_preview}") if params: logger.info(f"[BaseAPI] 요청 파라미터 - params={params}") if json_data: logger.info(f"[BaseAPI] 요청 바디 - json={json_data}") try: response = self._session.request( method=method, url=url, params=params, json=json_data, headers=self._get_headers(headers), timeout=timeout or self.DEFAULT_TIMEOUT, ) logger.info(f"[BaseAPI] 응답 수신 - method={method}, url={url}, status_code={response.status_code}") if response.status_code >= 400: logger.error(f"[BaseAPI] 에러 응답 - status_code={response.status_code}, body={response.text[:500]}") self._handle_error(response) if response.text: return response.json() return {} except requests.exceptions.Timeout: logger.error(f"[BaseAPI] 타임아웃 - url={url}") raise NHNCloudAPIError("요청 시간이 초과되었습니다.", code=408) except requests.exceptions.ConnectionError as e: logger.error(f"[BaseAPI] 연결 오류 - url={url}, error={e}") raise NHNCloudAPIError("서버에 연결할 수 없습니다.", code=503) except requests.exceptions.RequestException as e: logger.error(f"[BaseAPI] 요청 오류 - url={url}, error={e}") raise NHNCloudAPIError(f"요청 중 오류가 발생했습니다: {e}") def _handle_error(self, response: requests.Response) -> None: """에러 응답 처리""" try: error_data = response.json() if "error" in error_data: error = error_data["error"] raise NHNCloudAPIError( message=error.get("message", "알 수 없는 오류"), code=error.get("code", response.status_code), details=error, ) raise NHNCloudAPIError( message=f"API 오류: {response.status_code}", code=response.status_code, details=error_data, ) except ValueError: raise NHNCloudAPIError( message=f"API 오류: {response.status_code} - {response.text}", code=response.status_code, ) def _get(self, url: str, params: Optional[dict] = None, **kwargs) -> dict: """GET 요청""" return self._request("GET", url, params=params, **kwargs) def _post(self, url: str, json_data: Optional[dict] = None, **kwargs) -> dict: """POST 요청""" return self._request("POST", url, json_data=json_data, **kwargs) def _put(self, url: str, json_data: Optional[dict] = None, **kwargs) -> dict: """PUT 요청""" return self._request("PUT", url, json_data=json_data, **kwargs) def _delete(self, url: str, **kwargs) -> dict: """DELETE 요청""" return self._request("DELETE", url, **kwargs) def _patch(self, url: str, json_data: Optional[dict] = None, **kwargs) -> dict: """PATCH 요청""" return self._request("PATCH", url, json_data=json_data, **kwargs)