"""Cyphers API Client - Neople Cyphers Open API 래핑 레이어"""
import uuid
from typing import Any
from urllib.parse import quote
import httpx
from .cache import CacheManager, generate_cache_key
from .rate_limiter import TokenBucketRateLimiter
from .types import (
APIResponse,
CacheTTL,
CyphersAPIConfig,
CyphersAPIError,
RateLimitConfig,
)
class CyphersAPIClient:
"""Neople Cyphers Open API 클라이언트"""
def __init__(self, config: CyphersAPIConfig):
"""
Args:
config: API 클라이언트 설정
"""
self._api_key = config.api_key
self._base_url = config.base_url
self._timeout = config.timeout
self._retries = config.retries
# HTTP 클라이언트
self._client = httpx.AsyncClient(
base_url=self._base_url,
timeout=httpx.Timeout(config.timeout),
headers={"Content-Type": "application/json"},
)
# 캐시 및 레이트 리밋
self._cache = CacheManager()
self._rate_limiter = TokenBucketRateLimiter(RateLimitConfig())
async def close(self) -> None:
"""클라이언트 종료"""
await self._client.aclose()
async def _request(
self,
endpoint: str,
params: dict[str, Any] | None = None,
ttl: int | None = None,
) -> APIResponse:
"""
API 요청 실행 (캐싱 및 레이트 리밋 적용)
Args:
endpoint: API 엔드포인트
params: 쿼리 파라미터
ttl: 캐시 TTL (초)
Returns:
APIResponse: API 응답
"""
request_id = str(uuid.uuid4())
params = params or {}
# 캐시 확인
cache_key = generate_cache_key(endpoint, params)
cached_data = self._cache.get(cache_key)
if cached_data is not None:
rate_limit_status = self._rate_limiter.get_status()
return APIResponse(
data=cached_data,
cached=True,
request_id=request_id,
rate_limit={
"remaining": rate_limit_status.available,
"reset_at": rate_limit_status.reset_at,
},
)
# 레이트 리밋 확인
rate_limit_result = await self._rate_limiter.acquire()
if not rate_limit_result.allowed:
raise CyphersAPIError(
message=f"Rate limit exceeded. Reset at {rate_limit_result.reset_at}",
status_code=429,
error_code="RATE_LIMITED",
request_id=request_id,
)
# API Key 추가
query_params = {**params, "apikey": self._api_key}
# 요청 실행 (재시도 포함)
last_error: Exception | None = None
for attempt in range(self._retries + 1):
try:
response = await self._client.get(endpoint, params=query_params)
response.raise_for_status()
data = response.json()
# 캐시 저장
if ttl:
self._cache.set(cache_key, data, ttl)
return APIResponse(
data=data,
cached=False,
request_id=request_id,
rate_limit={
"remaining": rate_limit_result.remaining,
"reset_at": rate_limit_result.reset_at,
},
)
except httpx.HTTPStatusError as e:
last_error = self._handle_http_error(e, request_id)
# 4xx 에러는 재시도하지 않음
if e.response.status_code < 500:
raise last_error
except httpx.RequestError as e:
last_error = CyphersAPIError(
message=f"Request failed: {str(e)}",
error_code="UPSTREAM_TIMEOUT",
request_id=request_id,
)
# 모든 재시도 실패
raise last_error or CyphersAPIError(
message="Unknown error",
error_code="UNKNOWN_ERROR",
request_id=request_id,
)
def _handle_http_error(
self,
error: httpx.HTTPStatusError,
request_id: str,
) -> CyphersAPIError:
"""HTTP 에러 처리 및 표준화"""
status = error.response.status_code
try:
data = error.response.json()
error_info = data.get("error", {})
message = error_info.get("message", f"API Error: {status}")
error_code = error_info.get("code", "UNKNOWN_ERROR")
except Exception:
message = f"API Error: {status}"
error_code = "UNKNOWN_ERROR"
# HTTP 상태코드에 따른 에러 코드 매핑
if status == 401 or "API003" in error_code:
error_code = "AUTH_INVALID"
message = "Invalid API Key"
elif status == 429:
error_code = "RATE_LIMITED"
message = "Rate limit exceeded"
elif status == 404 or "CY002" in error_code:
error_code = "NOT_FOUND"
message = "Resource not found"
elif status >= 500:
error_code = "UPSTREAM_ERROR"
message = "Neople API server error"
return CyphersAPIError(
message=message,
status_code=status,
error_code=error_code,
request_id=request_id,
)
# ========== Player API ==========
async def search_players(
self,
nickname: str,
word_type: str | None = None,
limit: int | None = None,
) -> APIResponse:
"""
플레이어 검색
Args:
nickname: 검색할 닉네임
word_type: 검색 타입 (match/full)
limit: 결과 개수 제한
"""
params: dict[str, Any] = {"nickname": nickname}
if word_type:
params["wordType"] = word_type
if limit:
params["limit"] = limit
return await self._request("/cy/players", params, CacheTTL.SEARCH)
async def get_player(self, player_id: str) -> APIResponse:
"""
플레이어 정보 조회
Args:
player_id: 플레이어 ID
"""
return await self._request(f"/cy/players/{player_id}", ttl=CacheTTL.PLAYER_INFO)
async def get_player_matches(
self,
player_id: str,
game_type_id: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
next_token: str | None = None,
limit: int | None = None,
) -> APIResponse:
"""
플레이어 매칭 기록 조회
Args:
player_id: 플레이어 ID
game_type_id: 게임 타입 ID
start_date: 시작 날짜 (YYYYMMDD)
end_date: 종료 날짜 (YYYYMMDD)
next_token: 페이지네이션 토큰
limit: 결과 개수 제한
"""
params: dict[str, Any] = {}
if game_type_id:
params["gameTypeId"] = game_type_id
if start_date:
params["startDate"] = start_date
if end_date:
params["endDate"] = end_date
if next_token:
params["next"] = next_token
if limit:
params["limit"] = limit
return await self._request(
f"/cy/players/{player_id}/matches",
params,
CacheTTL.PLAYER_MATCHES,
)
# ========== Match API ==========
async def get_match(self, match_id: str) -> APIResponse:
"""
매치 상세 조회
Args:
match_id: 매치 ID
"""
return await self._request(f"/cy/matches/{match_id}", ttl=CacheTTL.MATCH_DETAIL)
# ========== Ranking API ==========
async def get_ranking_rating_point(
self,
player_id: str | None = None,
nickname: str | None = None,
offset: int | None = None,
limit: int | None = None,
) -> APIResponse:
"""
통합 랭킹 조회
Args:
player_id: 플레이어 ID
nickname: 플레이어 닉네임
offset: 오프셋
limit: 결과 개수 제한
"""
params: dict[str, Any] = {}
if player_id:
params["playerId"] = player_id
if nickname:
params["nickname"] = nickname
if offset is not None:
params["offset"] = offset
if limit:
params["limit"] = limit
return await self._request("/cy/ranking/ratingpoint", params, CacheTTL.RANKING)
async def get_ranking_characters(
self,
character_id: str,
ranking_type: str,
offset: int | None = None,
limit: int | None = None,
) -> APIResponse:
"""
캐릭터 랭킹 조회
Args:
character_id: 캐릭터 ID
ranking_type: 랭킹 타입
offset: 오프셋
limit: 결과 개수 제한
"""
params: dict[str, Any] = {}
if offset is not None:
params["offset"] = offset
if limit:
params["limit"] = limit
return await self._request(
f"/cy/ranking/characters/{character_id}/{ranking_type}",
params,
CacheTTL.RANKING,
)
async def get_ranking_tsj(
self,
tsj_type: str,
offset: int | None = None,
limit: int | None = None,
) -> APIResponse:
"""
투신전 랭킹 조회
Args:
tsj_type: 투신전 타입
offset: 오프셋
limit: 결과 개수 제한
"""
params: dict[str, Any] = {}
if offset is not None:
params["offset"] = offset
if limit:
params["limit"] = limit
return await self._request(f"/cy/ranking/tsj/{tsj_type}", params, CacheTTL.RANKING)
# ========== Item API ==========
async def search_items(
self,
item_name: str | None = None,
character_id: str | None = None,
slot_code: str | None = None,
rarity_code: str | None = None,
season_code: str | None = None,
limit: int | None = None,
) -> APIResponse:
"""
아이템 검색
Args:
item_name: 아이템 이름
character_id: 캐릭터 ID
slot_code: 슬롯 코드
rarity_code: 레어리티 코드
season_code: 시즌 코드
limit: 결과 개수 제한
"""
params: dict[str, Any] = {}
if item_name:
params["itemName"] = item_name
if character_id:
params["q[characterId]"] = character_id
if slot_code:
params["q[slotCode]"] = slot_code
if rarity_code:
params["q[rarityCode]"] = rarity_code
if season_code:
params["q[seasonCode]"] = season_code
if limit:
params["limit"] = limit
return await self._request("/cy/battleitems", params, CacheTTL.SEARCH)
async def get_item(self, item_id: str) -> APIResponse:
"""
아이템 상세 조회
Args:
item_id: 아이템 ID
"""
return await self._request(f"/cy/battleitems/{item_id}", ttl=CacheTTL.ITEM_DETAIL)
async def get_multi_items(self, item_ids: list[str]) -> APIResponse:
"""
다중 아이템 상세 조회
Args:
item_ids: 아이템 ID 목록 (최대 30개)
"""
if len(item_ids) > 30:
raise CyphersAPIError(
message="Maximum 30 items allowed",
status_code=400,
error_code="INVALID_INPUT",
request_id=str(uuid.uuid4()),
)
params = {"itemIds": ",".join(item_ids)}
return await self._request("/cy/multi/battleitems", params, CacheTTL.ITEM_DETAIL)
# ========== Character API ==========
async def get_characters(self) -> APIResponse:
"""캐릭터 목록 조회"""
return await self._request("/cy/characters", ttl=CacheTTL.CHARACTERS)
# ========== Utility ==========
def get_cache_stats(self):
"""캐시 통계 조회"""
return self._cache.get_stats()
def clear_cache(self) -> None:
"""캐시 초기화"""
self._cache.clear()
def get_rate_limit_status(self):
"""레이트 리밋 상태 조회"""
return self._rate_limiter.get_status()
