sync_amz_data/sync_amz_data/public/asj_client.py

import requests
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
import logging
from typing import Union, List

logger = logging.getLogger(__name__)


class AsjHttpClient:
    """广告后端API客户端"""

    def __init__(self):
        retry_strategy = Retry(
            total=3,  # 重试次数
            # 强制重试的状态码，在method_whitelist中的请求方法才会重试
            allowed_methods=["GET", "POST"],
            status_forcelist=[429, 500, 502, 503, 504],
            raise_on_status=False,  # 在status_forcelist中的状态码达到重试次数后是否抛出异常
            # backoff_factor * (2 ** (retry_time-1))， 即间隔1s, 2s, 4s, 8s, ...
            backoff_factor=1,
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session = requests.session()
        self.session.mount("http://", adapter)

    def _request(self, method, url, *, params=None, body=None):
        headers = {"X-Token": "da4ab6bc5cbf1dfa"}
        resp = self.session.request(method, url, params=params, headers=headers, json=body)
        if resp.status_code >= 400:
            raise Exception(resp.text)
        js = resp.json()
        if js.get("code") != 2000:
            raise Exception(js)

        return js

    def create(self, url: str, data: [dict, List[dict]]):
        return self._request("POST", url, body=data)


asj_api = AsjHttpClient()