68 lines
2.7 KiB
Python
68 lines
2.7 KiB
Python
import os
|
|
import requests
|
|
import time
|
|
from typing import Dict, Any, Optional
|
|
from .exceptions import WBAPIError, WBAuthError, WBRateLimitError, WBBadRequestError
|
|
|
|
class WBClient:
|
|
"""Base client for interacting with Wildberries API."""
|
|
|
|
# WB domains
|
|
CONTENT_API = "https://content-api.wildberries.ru"
|
|
MARKETPLACE_API = "https://marketplace-api.wildberries.ru"
|
|
PRICES_API = "https://discounts-prices-api.wildberries.ru"
|
|
STATISTICS_API = "https://statistics-api.wildberries.ru"
|
|
|
|
def __init__(self, api_token: Optional[str] = None):
|
|
# Fetch from env if not provided
|
|
self.api_token = api_token or os.getenv('WB_API_TOKEN')
|
|
|
|
if not self.api_token:
|
|
raise ValueError("WB_API_TOKEN must be provided")
|
|
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
"Authorization": self.api_token,
|
|
"Content-Type": "application/json"
|
|
})
|
|
|
|
def request(self, method: str, base_url: 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 WB API with automatic retries."""
|
|
url = f"{base_url}{path}"
|
|
|
|
for attempt in range(retries):
|
|
try:
|
|
response = self.session.request(method, url, json=data, params=params)
|
|
|
|
if response.status_code == 429:
|
|
if attempt < retries - 1:
|
|
time.sleep(2 ** attempt) # Exponential backoff
|
|
continue
|
|
raise WBRateLimitError("Rate limit exceeded")
|
|
|
|
if response.status_code in (401, 403):
|
|
raise WBAuthError(f"Authentication failed: {response.text}")
|
|
|
|
if response.status_code == 400:
|
|
raise WBBadRequestError(f"Bad Request: {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 WBAPIError(f"Request failed: {str(e)}")
|
|
|
|
raise WBAPIError("Request failed after max retries")
|
|
|
|
def post(self, base_url: str, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
return self.request("POST", base_url, path, data=data)
|
|
|
|
def get(self, base_url: str, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
return self.request("GET", base_url, path, params=params)
|