Files
icurfer 8c7739ffad
Some checks failed
Build and Push Docker Image / build (push) Has been cancelled
Add NHN Cloud API integration with async task support
- 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>
2026-01-14 01:29:21 +09:00

124 lines
3.8 KiB
Python

"""
NHN Cloud Token Management Module
NHN Cloud API 인증 토큰 관리
"""
import logging
from typing import Optional
from dataclasses import dataclass
import requests
from .base import NHNCloudEndpoints, NHNCloudAPIError
logger = logging.getLogger(__name__)
@dataclass
class TokenResponse:
"""토큰 응답 데이터"""
token: str
tenant_id: str
expires_at: Optional[str] = None
class NHNCloudToken:
"""NHN Cloud 토큰 관리 클래스"""
TOKEN_URL = f"{NHNCloudEndpoints.IDENTITY}/v2.0/tokens"
def __init__(self, tenant_id: str, username: str, password: str):
"""
Args:
tenant_id: NHN Cloud 테넌트 ID
username: NHN Cloud 사용자 이메일
password: NHN Cloud API 비밀번호
"""
self.tenant_id = tenant_id
self.username = username
self.password = password
self._token: Optional[str] = None
def create_token(self) -> TokenResponse:
"""
새 토큰 생성
Returns:
TokenResponse: 생성된 토큰 정보
Raises:
NHNCloudAPIError: 토큰 생성 실패 시
"""
payload = {
"auth": {
"tenantId": self.tenant_id,
"passwordCredentials": {
"username": self.username,
"password": self.password,
},
}
}
logger.info(f"[NHN API] 토큰 생성 요청 - URL={self.TOKEN_URL}, tenant_id={self.tenant_id}, username={self.username}")
try:
response = requests.post(
self.TOKEN_URL,
json=payload,
headers={"Content-Type": "application/json"},
timeout=30,
)
logger.info(f"[NHN API] 토큰 생성 응답 - status_code={response.status_code}")
data = response.json()
if "error" in data:
error = data["error"]
logger.error(f"[NHN API] 토큰 생성 실패 - code={error.get('code')}, message={error.get('message')}, details={error}")
raise NHNCloudAPIError(
message=error.get("message", "토큰 생성 실패"),
code=error.get("code"),
details=error,
)
access = data.get("access", {})
token_info = access.get("token", {})
self._token = token_info.get("id")
token_preview = self._token[:8] + "..." if self._token else "None"
logger.info(f"[NHN API] 토큰 생성 성공 - tenant_id={self.tenant_id}, token={token_preview}, expires={token_info.get('expires')}")
return TokenResponse(
token=self._token,
tenant_id=self.tenant_id,
expires_at=token_info.get("expires"),
)
except requests.exceptions.Timeout as e:
logger.error(f"[NHN API] 토큰 생성 타임아웃 - URL={self.TOKEN_URL}, error={e}")
raise NHNCloudAPIError(f"토큰 생성 요청 타임아웃: {e}")
except requests.exceptions.ConnectionError as e:
logger.error(f"[NHN API] 토큰 생성 연결 실패 - URL={self.TOKEN_URL}, error={e}")
raise NHNCloudAPIError(f"토큰 생성 서버 연결 실패: {e}")
except requests.exceptions.RequestException as e:
logger.error(f"[NHN API] 토큰 생성 요청 실패 - URL={self.TOKEN_URL}, error={e}")
raise NHNCloudAPIError(f"토큰 생성 요청 실패: {e}")
@property
def token(self) -> Optional[str]:
"""현재 토큰 반환"""
return self._token
def get_token(self) -> str:
"""
토큰 반환 (없으면 생성)
Returns:
str: API 토큰
"""
if not self._token:
self.create_token()
return self._token