80 lines
3.4 KiB
Python
80 lines
3.4 KiB
Python
import os
|
||
import requests
|
||
import time
|
||
from typing import Dict, Any, Optional
|
||
from .exceptions import MegaAPIError, MegaAuthError, MegaRateLimitError, MegaBadRequestError, MegaNotFoundError
|
||
|
||
class MegaClient:
|
||
"""Base client for interacting with MegaMarket API."""
|
||
|
||
BASE_URL = "https://partner.goodsteam.tech/api/market/v1"
|
||
|
||
def __init__(self, api_token: Optional[str] = None):
|
||
# Fetch from env if not provided
|
||
self.api_token = api_token or os.getenv('MEGA_API_TOKEN')
|
||
|
||
if not self.api_token:
|
||
raise ValueError("MEGA_API_TOKEN must be provided")
|
||
|
||
self.session = requests.Session()
|
||
# Мегамаркет может требовать токен как Bearer, либо внутри тела JSON в "meta".
|
||
# Будем использовать Bearer, как наиболее распространенный стандарт для V1.
|
||
self.session.headers.update({
|
||
"Authorization": f"Bearer {self.api_token}",
|
||
"Content-Type": "application/json"
|
||
})
|
||
|
||
def request(self, method: str, path: str, data: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None, retries: int = 3) -> Dict[str, Any]:
|
||
"""Perform HTTP request to MegaMarket API with automatic retries."""
|
||
url = f"{self.BASE_URL}{path}"
|
||
|
||
for attempt in range(retries):
|
||
try:
|
||
response = self.session.request(method, url, json=data, params=params)
|
||
|
||
# Check rate limiting
|
||
if response.status_code == 429:
|
||
if attempt < retries - 1:
|
||
time.sleep(2 ** attempt) # Exponential backoff
|
||
continue
|
||
raise MegaRateLimitError("Rate limit exceeded")
|
||
|
||
if response.status_code in (401, 403):
|
||
raise MegaAuthError(f"Authentication failed: {response.text}")
|
||
|
||
if response.status_code == 400:
|
||
raise MegaBadRequestError(f"Bad Request: {response.text}")
|
||
|
||
if response.status_code == 404:
|
||
raise MegaNotFoundError(f"Not Found: {response.text}")
|
||
|
||
response.raise_for_status()
|
||
|
||
if response.content:
|
||
return response.json()
|
||
return {}
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
if attempt < retries - 1:
|
||
time.sleep(2 ** attempt)
|
||
continue
|
||
raise MegaAPIError(f"Request failed: {str(e)}")
|
||
|
||
raise MegaAPIError("Request failed after max retries")
|
||
|
||
def post(self, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
# Мегамаркет во многих методах требует структуру {"meta": {}, "data": {}}
|
||
# Если data не обернута, обернем ее
|
||
if data and "data" not in data and "meta" not in data:
|
||
payload = {
|
||
"meta": {},
|
||
"data": data
|
||
}
|
||
else:
|
||
payload = data
|
||
|
||
return self.request("POST", path, data=payload)
|
||
|
||
def get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
return self.request("GET", path, params=params)
|