zyc_data_rsync/common/utils.py

import json

import requests

from common.error import NoValidError
from common.logger import logger
from common import config
from util.login import LoginApi

cookies = {}

def call_api(endpoint, method="GET", params=None, data=None) -> dict:
    """
    通用的API调用方法。
    """
    # 记录调用的信息
    global cookies
    url = f"{config.BASE_URL}{endpoint}"
    for i in range(3):
        try:
            if method.upper() == "GET":
                response = requests.get(url, params=params, cookies=cookies)
            elif method.upper() == "POST":
                response = requests.post(url, json=data, params=params, cookies=cookies)
            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}")
            if response.status_code == 502:
                cookies = LoginApi().login_bmc(config.USERNAME, config.PASSWORD)
                raise NoValidError()
            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")


def save_json(data, filename):
    with open(filename, "w") as f:
        f.write(json.dumps(data))