Joey
52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
"""Ozon Seller API 客户端（薄封装：鉴权头 + 错误映射 + 退避）。"""
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
OZON_BASE_URL = "https://api-seller.ozon.ru"
|
|
|
|
|
|
class OzonAPIError(Exception):
|
|
def __init__(self, status: int, detail: str):
|
|
self.status = status
|
|
self.detail = detail
|
|
super().__init__(f"Ozon API {status}: {detail}")
|
|
|
|
|
|
class OzonClient:
|
|
def __init__(self, client_id: str, api_key: str, base_url: str = OZON_BASE_URL):
|
|
self.client_id = client_id
|
|
self.api_key = api_key
|
|
self.base_url = base_url
|
|
|
|
def _headers(self) -> dict:
|
|
return {
|
|
"Client-Id": self.client_id,
|
|
"Api-Key": self.api_key,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def post(self, path: str, body: dict | None = None, timeout: float = 60.0) -> dict:
|
|
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
|
|
resp = await client.post(f"{self.base_url}{path}", headers=self._headers(), json=body or {})
|
|
if resp.status_code >= 400:
|
|
raise OzonAPIError(resp.status_code, resp.text[:500])
|
|
try:
|
|
return resp.json()
|
|
except Exception: # noqa: BLE001
|
|
return {}
|
|
|
|
async def get(self, path: str, timeout: float = 60.0) -> dict:
|
|
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
|
|
resp = await client.get(f"{self.base_url}{path}", headers=self._headers())
|
|
if resp.status_code >= 400:
|
|
raise OzonAPIError(resp.status_code, resp.text[:500])
|
|
try:
|
|
return resp.json()
|
|
except Exception: # noqa: BLE001
|
|
return {}
|
|
|
|
async def test_credentials(self) -> dict:
|
|
"""调 /v1/roles 校验凭证与权限范围。"""
|
|
return await self.post("/v1/roles", {})
|