zyc_data_rsync/common/util.py

import requests

from common import config
from common.logger import logger


def req_api(endpoint, method="GET", params=None, data=None, headers=None, cookies=None) -> dict:
    """
    通用的API调用方法。
    """
    for i in range(3):
        try:
            if method.upper() == "GET":
                response = requests.get(endpoint, params=params, cookies=cookies, verify=False, headers=headers)
            elif method.upper() == "POST":
                response = requests.post(endpoint, json=data, params=params, cookies=cookies, verify=False, headers=headers)
            elif method.upper() == "PUT":
                response = requests.put(endpoint, json=data, params=params, cookies=cookies, verify=False, headers=headers)
            else:
                raise ValueError("Unsupported HTTP method")
            logger.info(f"Calling API: {endpoint}, Method: {method}, Params: {params}, Data: {data}, ResCode:{response.status_code}, ResData: {response.text}")
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"Error calling API {endpoint}: {e}")
            logger.warning(f"Calling API: {endpoint}, Method: {method}, Params: {params}, Data: {data}, Error:{e}")
    logger.error(f"Failed to call API 3 times: {endpoint}, Method: {method}, Params: {params}, Data: {data}")
    raise Exception("Failed to call API 3 times")