kg_person/bin/utils/apiMethod.py

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
"""
定义不同的请求方式
Author: songjian at <songjian12@baidu.com>
Created: 2020/4/21
"""

import sys
import requests
import os
import logging
import json

reload(sys)
sys.setdefaultencoding('utf-8')


def post(url, header, request_parameter_type='raw', data=None, filePath=None, file_key='file'):
    """
    post请求
    """
    if request_parameter_type == 'form-data':
        files = {
            file_key: open(filePath, "rb")
        }
        try:
            response = requests.post(url=url, data=data, headers=header, files=files, timeout=(15, 300))
        except requests.exceptions.RequestException as e:
            logging.error(e)
        logging.info('url:%s,params:%s,headers:%s,files:%s' % (url, data, header, os.path.basename(filePath)))
    else:
        if header and 'Content-Type' not in header.keys():
            header['Content-Type'] = 'application/json'
        elif not header:
            header = {"Content-Type": "application/json"}
        try:
            response = requests.post(url=url, data=data, headers=header, timeout=(15, 300))
        except requests.exceptions.RequestException as e:
            logging.error(e)
        logging.info('url:%s,params:%s,headers:%s' % (url, data, header))
    try:
        logging.info('response:%s %s' % (response.status_code, json.dumps(response.json(), ensure_ascii=False)))
        return response.status_code, response.json()
    except Exception as e:
        logging.error(e)
        logging.error('response:%s %s' % (response.status_code, response.text))
        return response.status_code, response.text


def get(url, header, data='{}'):
    """
    get请求
    """
    if header and 'Content-Type' not in header.keys():
        header['Content-Type'] = 'application/json'
    elif not header:
        header = {"Content-Type": "application/json"}
    try:
        response = requests.get(url=url, params=json.loads(data), headers=header, timeout=(15, 300))
    except requests.exceptions.RequestException as e:
        logging.error(e)
    logging.info('url:%s,params:%s,headers:%s' % (url, data, header))
    try:
        logging.info('response:%s %s' % (response.status_code, json.dumps(response.json(), ensure_ascii=False)))
        return response.status_code, response.json()
    except Exception as e:
        logging.error(e)
        logging.error('response:%s %s' % (response.status_code, response.text))
        return response.status_code, response.text

def put(url, header, data=None):
    """
    put请求
    """
    if header and 'Content-Type' not in header.keys():
        header['Content-Type'] = 'application/json'
    elif not header:
        header = {"Content-Type": "application/json"}
    try:
        response = requests.put(url=url, data=data, headers=header, timeout=(15, 300))
    except requests.exceptions.RequestException as e:
        logging.error(e)
    logging.info('url:%s,params:%s,headers:%s' % (url, data, header))
    try:
        logging.info('response:%s %s' % (response.status_code, json.dumps(response.json(), ensure_ascii=False)))
        return response.status_code, response.json()
    except Exception as e:
        logging.error(e)
        logging.error('response:%s %s' % (response.status_code, response.text))
        return response.status_code, response.text