sky_httptrans.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@Email: ludq1@chinaunicom.cn
@date: 2023/04/07 11:40:00
@Description:
"""

import json
import re

import requests

from .globalconst import HttpConst
from .my_stringutils import MyStringUtils
from .my_utils import MyUtils
from .sky_baseexception import MyBaseException, create_base_exception


class HttpTrans:
    """
    http传输处理
    """

    # 定义要透传的header_key
    _header_key = [
        'Cookie'
        , 'x-request-id'
        , 'x-b3-traceid'
        , 'x-b3-spanid'
        , 'x-b3-parentspanid'
        , 'x-b3-sampled'
        , 'x-b3-flags'
        , 'x-ot-span-context'
    ]

    def __init__(self, http_header: dict = None):
        r"""

        Args:
            http_header: dict http的请求头
        """

        if http_header is not None:
            try:
                # 尝试将 http_header 转成 dict 并赋值
                self._header = dict(http_header)
            except:
                raise ValueError('http_header 不能转换为字典')
        else:
            self._header = None

    def generate_header(self, content_type: str = None,
                        dict_header: dict = None) -> dict:
        """
         生成http头信息
        Args:
            content_type: string 传输类型
            dict_header: dict 额外头信息

        Returns: dict

        """

        # 检查参数
        if not (dict_header is None or isinstance(dict_header, dict)):
            raise ValueError('dict_header 不是字典')

        ret_header = {}

        # 加入需要透传的 self._header
        if self._header:
            # 转换为标准的字符串kv字典
            formatted_dict = MyUtils.format_dict_to_str_dict(self._header)
            # 过滤出指定的key
            filtered_keys = filter(lambda tmp_k: tmp_k in self._header_key, formatted_dict.keys())
            # 生成过滤后的字典
            filtered_dict = {tmp_k: formatted_dict.get(tmp_k) for tmp_k in filtered_keys}
            # 更新 ret_dict
            ret_header.update(filtered_dict)

        # 设置参数默认值-- str_content_type
        if not MyStringUtils.is_empty(content_type):
            str_content_type = str(content_type)
        else:
            str_content_type = HttpConst.CONTENT_TYPE_URLENCODED

        # 更新 ret_dict
        ret_header.update({
            HttpConst.HEADER_KEY_CONTENT_TYPE: str_content_type
        })

        # 加入额外的HEADER KEY
        if dict_header:
            formatted_dict = MyUtils.format_dict_to_str_dict(dict_header)
            # 更新 ret_dict
            ret_header.update(formatted_dict)

        # 移除  Request Method 这个key,因为该key存在会造成requests请求失效

        ret_header.pop(HttpConst.HEADER_KEY_REQUEST_METHOD, None)

        return ret_header

    def get_data(self, urlstr: str, trans_data: dict = None, dict_header: dict = None) -> requests.Response:
        """
         post 传输数据
        Args:
            urlstr:
            trans_data: 要附加到 urlstr 上的 key-value 字典 , kv 会自动编码
            dict_header: 传输时使用的header

        Returns:

        """
        # 检查参数
        if not (trans_data is None or isinstance(trans_data, dict)):
            raise ValueError('trans_data 不是字典')

        if trans_data:
            trans_data = MyUtils.format_dict_to_str_dict(trans_data)

        content_type = HttpConst.CONTENT_TYPE_URLENCODED
        if dict_header is not None and HttpConst.HEADER_KEY_CONTENT_TYPE in dict_header.keys():
            content_type = dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE)
        headers = self.generate_header(
            content_type=content_type,
            dict_header=dict_header
        )

        cookies = None
        if HttpConst.HEADER_KEY_COOKIE in headers:
            cookies = headers.get(HttpConst.HEADER_KEY_COOKIE)

        regexp = (r'^https://')
        match = re.search(regexp, urlstr.strip(), re.U)

        if (match is not None):
            return requests.get(url=urlstr, headers=headers, params=trans_data, cookies=cookies, verify=False)
        else:
            return requests.get(url=urlstr, headers=headers, params=trans_data, cookies=cookies)

    def post_encodeddata_standard(self, urlstr: str, trans_data: dict = None,
                                  dict_header: dict = None) -> requests.Response:
        """
         post 传输数据,返回 Response对象
        Args:
            urlstr: string httpurl
            trans_data: 传输的内容 ,字典 , kv会自动编码
            dict_header: 传输时使用的header

        Returns:

        """

        if not (trans_data is None or isinstance(trans_data, dict)):
            raise ValueError('trans_data 不是字典')

        trans_data_tmp = None
        content_type = HttpConst.CONTENT_TYPE_URLENCODED
        if dict_header is not None and HttpConst.HEADER_KEY_CONTENT_TYPE in dict_header.keys():
            content_type = dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE)
            if HttpConst.CONTENT_TYPE_JSON == dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE):
                trans_data_tmp = json.dumps(trans_data)
        if trans_data_tmp is None:
            trans_data_tmp = MyUtils.format_dict_to_str_dict(trans_data)

        headers = self.generate_header(
            content_type=content_type,
            dict_header=dict_header
        )

        regexp = (r'^https://')
        match = re.search(regexp, urlstr.strip(), re.U)

        if (match is not None):
            return requests.post(url=urlstr, headers=headers, data=trans_data_tmp, verify=False)
        else:
            return requests.post(url=urlstr, headers=headers, data=trans_data_tmp)

    def post_encodedata(self, urlstr: str,
                        trans_data: dict = None,
                        dict_header: dict = None,
                        **dict_arg) -> str:
        """
         post 传输数据,返回rensponse内容字符串,传输异常时返回 标准错误的json字符串,所以该方法适用于标准的传输
        Args:
            urlstr: string httpurl
            trans_data: 传输的内容 , 字典,kv会自动编码
            dict_header: 传输时使用的header
            **dict_arg: 额外参数

        Returns: 标准格式的 json字符串

        """
        # 以后额外扩展
        # if dict_arg:
        #    pass

        try:
            m_response = self.post_encodeddata_standard(
                urlstr=urlstr,
                trans_data=trans_data,
                dict_header=dict_header
            )

            if not m_response.ok:
                raise create_base_exception(
                    '访问url出错,访问地址:{0},返回码{1}'.format(urlstr, m_response.status_code)
                )
            else:
                str_ret = m_response.text

            # 检查返回结果是不是有效的 jsonstr
            dict_ret = MyUtils.jsonstr2dict(str_ret)
            if dict_ret is None:
                raise create_base_exception(
                    '返回的内容不是有效的json字符串:{}'.format(str_ret)
                )

            return str_ret
        except requests.exceptions.RequestException as exception:
            return MyUtils.dict2jsonstr(MyBaseException.format_exception_to_standard_dict(exception))
        except MyBaseException as exception:
            return MyUtils.dict2jsonstr(MyBaseException.format_exception_to_standard_dict(exception))
        except Exception as exception:
            return MyUtils.dict2jsonstr(MyBaseException.format_exception_to_standard_dict(exception))

    def patch_encodeddata_standard(self, urlstr: str, trans_data: dict = None,
                                   dict_header: dict = None) -> requests.Response:
        """
         patch 传输数据,返回 Response对象
        Args:
            urlstr: string httpurl
            trans_data: 传输的内容 ,字典 , kv会自动编码
            dict_header: 传输时使用的header

        Returns:

        """

        if not (trans_data is None or isinstance(trans_data, dict)):
            raise ValueError('trans_data 不是字典')

        trans_data_tmp = None
        content_type = HttpConst.CONTENT_TYPE_PATH_JSON
        if dict_header is not None and HttpConst.HEADER_KEY_CONTENT_TYPE in dict_header.keys():
            content_type = dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE)
            if HttpConst.CONTENT_TYPE_PATH_JSON == dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE):
                trans_data_tmp = json.dumps(trans_data)
        if trans_data_tmp is None:
            trans_data_tmp = MyUtils.format_dict_to_str_dict(trans_data)

        headers = self.generate_header(
            content_type=content_type,
            dict_header=dict_header
        )

        regexp = (r'^https://')
        match = re.search(regexp, urlstr.strip(), re.U)

        if (match is not None):
            return requests.patch(url=urlstr, headers=headers, data=trans_data_tmp, verify=False)
        else:
            return requests.patch(url=urlstr, headers=headers, data=trans_data_tmp)

    def delete_encodeddata_standard(self, urlstr: str, trans_data: dict = None,
                                    dict_header: dict = None) -> requests.Response:
        """
         patch 传输数据,返回 Response对象
        Args:
            urlstr: string httpurl
            trans_data: 传输的内容 ,字典 , kv会自动编码
            dict_header: 传输时使用的header

        Returns:

        """

        if not (trans_data is None or isinstance(trans_data, dict)):
            raise ValueError('trans_data 不是字典')

        trans_data_tmp = None
        content_type = HttpConst.CONTENT_TYPE_JSON
        if dict_header is not None and HttpConst.HEADER_KEY_CONTENT_TYPE in dict_header.keys():
            content_type = dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE)
            if HttpConst.CONTENT_TYPE_JSON == dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE):
                trans_data_tmp = json.dumps(trans_data)

        # 如果不是json，以字符串进行传递
        if trans_data_tmp is None:
            trans_data_tmp = MyUtils.format_dict_to_str_dict(trans_data)

        headers = self.generate_header(
            content_type=content_type,
            dict_header=dict_header
        )

        regexp = (r'^https://')
        match = re.search(regexp, urlstr.strip(), re.U)

        if (match is not None):
            return requests.delete(url=urlstr, headers=headers, verify=False)
        else:
            return requests.delete(url=urlstr, headers=headers)