Add NHN Cloud API integration with async task support
Some checks failed
Build and Push Docker Image / build (push) Has been cancelled

- NHN Cloud API packages: token, vpc, compute, nks, storage
- REST API endpoints with Swagger documentation
- Async task processing for long-running operations
- CORS configuration for frontend integration
- Enhanced logging for debugging API calls

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-14 01:29:21 +09:00
parent 256fed485e
commit 8c7739ffad
32 changed files with 4059 additions and 0 deletions

165
nhn/packages/base.py Normal file
View File

@ -0,0 +1,165 @@
"""
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"
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)