49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
"""NAS stock 백엔드 trade-alert 계약 — X-WebAI-Key + retry."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_MAX_ATTEMPTS = 3
|
|
_RETRY_STATUSES = {429, 500, 502, 503, 504}
|
|
|
|
|
|
class NASClient:
|
|
def __init__(self, base_url: str, api_key: str, timeout: float = 10.0):
|
|
self._base_url = base_url.rstrip("/")
|
|
self._api_key = api_key
|
|
self._client = httpx.AsyncClient(timeout=timeout)
|
|
|
|
async def close(self) -> None:
|
|
await self._client.aclose()
|
|
|
|
async def get_monitor_set(self) -> dict:
|
|
return await self._request("GET", "/api/webai/trade-alert/monitor-set")
|
|
|
|
async def post_report(self, as_of: str, firing: list[dict]) -> dict:
|
|
return await self._request(
|
|
"POST", "/api/webai/trade-alert/report",
|
|
json={"as_of": as_of, "firing": firing})
|
|
|
|
async def _request(self, method: str, path: str, **kwargs) -> dict:
|
|
url = f"{self._base_url}{path}"
|
|
headers = {"X-WebAI-Key": self._api_key}
|
|
for attempt in range(_MAX_ATTEMPTS):
|
|
try:
|
|
resp = await self._client.request(method, url, headers=headers, **kwargs)
|
|
if resp.status_code in _RETRY_STATUSES and attempt < _MAX_ATTEMPTS - 1:
|
|
await asyncio.sleep(2 ** attempt)
|
|
continue
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except httpx.TimeoutException:
|
|
if attempt < _MAX_ATTEMPTS - 1:
|
|
await asyncio.sleep(2 ** attempt)
|
|
continue
|
|
raise
|
|
raise RuntimeError("retry exhausted")
|