python--积累--带有用户认证Authorization的post请求

逃离我推掉我的手 2023-02-13 14:40 100阅读 0赞

class.py 如下：

import ssl
    ssl._create_default_https_context = ssl._create_unverified_context
    import json
    import urllib
    import urllib.request
    import base64
    
    from api_module.logger import logger
    
    
    class ProductRequestFactory:
        def __init__(self, config):
            self.hostname = config['host']
            self.username = config['username']
            self.password = config['password']
    
        def upload_match_result_2_product(self, body):
            '''
            上报匹配结果到产品端
            '''
    
            url = self.hostname + '/v1/match_results/batch_commit'
            print(f'{self.hostname}:{self.username}:{self.password}')
            auth = str(base64.b64encode(f'{self.username}:{self.password}'.encode('utf-8')), 'utf-8')
            headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:23.0) Gecko/20100101 Firefox/23.0', "Content-Type": "application/json", 'Authorization': f'Basic {auth}'}
            jsonbody = json.dumps(body, default=lambda o: o.__dict__, sort_keys=True, indent=0)
            jsonstr = str(jsonbody).replace("\n", "").replace("\\", "").replace("\"[", "[").replace("]\"", "]")
            print(f' str 为{jsonstr}')
            print({'headers': json.dumps(headers)})
            r = urllib.request.Request(url, data=jsonstr.encode('utf-8'), headers=headers)
            try:
                response = urllib.request.urlopen(r, timeout=6)
                status = response.status
                print(f"上报返回为{response}")
                print(status)
                if status == 200:
                    print("上报成功")
                    the_page = response.read()
                    content = the_page.decode("utf8")
                    print(content)
                    return content
                else:
                    logger.error(f'调用 product-api 异常: {barcode}, {status}, {ret}')
                    logger.error(response.data)
    
            except Exception as e:
                logger.error(f'调用 product-api 异常')
                logger.error(e)

配置为

productapi {
            host = "https://gateway-api-test.test.com"
            username = "123"
            password = "123-test"
        }

使用方式

return_json = []
    pr = PersonResult(str(id), matchRecords)
    return_json.append(pr)
    product_api = ProductRequestFactory(config['productapi'])
    product_api.upload_match_result_2_product(return_json)

PersonResult

class PersonResult(object):
    
        def __init__(self, personId, matchRecords):
            self.personId = personId
            self.matchRecords = matchRecords