http_util
http_util
1 使用
1.1 常规使用
GET 请求，无参数
from xlib.util import http_util
res = http_util.get("http://127.0.0.1:8585/x/ping")
if res.success():
print(res.output())GET 请求，校验结果，校验失败时会记录日志
from xlib.util import http_util
res = http_util.get("http://127.0.0.1:8585/x/ping", check_key="stat", check_value=0)
assert res.success() == False
res = http_util.get("http://127.0.0.1:8585/x/ping", check_key="stat", check_value="OK")
assert res.success() == True
print res.output()GET 请求, 有参数
http_util.get("http://127.0.0.1:8585/x/hello", params={"str_info": "meetbill"})post 请求，使用 JSON 方式
http_util.post_json("http://127.0.0.1:8585/x/hello", data={"str_info": "meetbill"}, debug=True)post 请求，使用 FORM 方式（butterfly 服务端不支持此方式）
print("[POST], post form---------------------------------[exe error]")
post_form("http://127.0.0.1:8585/x/hello", data={"str_info": "meetbill"}, debug=True)GET 请求, 有 header
# GET，have headers
headers = {"X_USERNAME": "meetbill"}
res = get("http://127.0.0.1:8585/demo_api/ping", headers=headers, debug=True)
if res.success():
print(res.output())1.2 替换 requests 使用
1.2.1 requests
依赖包
requests-2.22.0.tar.gz
urllib3_1_26_2-1.26.2.0.tar.gz
idna-2.8.tar.gz
chardet-3.0.4.tar.gz
certifi-2019.6.16.tar.gz代码
import requests
def main():
...
r = requests.request(method, url, headers=headers, params=params, data=data, json=json, timeout=timeout)
curl_cmd = http_util.to_curl_for_requests(r.request)
logger.info(("cmd={cmd}\t"
"status_code={status_code}\t"
"response={response}".format(cmd=curl_cmd, status_code=r.status_code, response=r.text)))
return res
-----
r.status_code : 响应状态码
r.json() : JSON 响应内容1.2.2 http_util
from xlib.util import http_util
def main():
...
rsp = http_util.request(method, url, headers=headers, params=params, data=data, json=json, timeout=timeout)
return rsp
-----
rsp.status_code : 响应状态码
rsp.json() : JSON 响应内容1.2.3 SDK 装饰器
from xlib.util import wrapt
from xlib.util import funcsigs
from xlib.util import http_util
from xlib.util import config_util
def requested(method, uri):
"""
添加 endpoint
"""
@wrapt.decorator
def wrapper(wrapped, instance, args, kwargs):
"""
wrapper
"""
# 提取原 func 中的 req, 不论使用位置参数还是关键字参数
sig = funcsigs.signature(wrapped)
kargs = dict(zip([k for k in sig.parameters if k not in kwargs], args))
req = kargs.get("req", kwargs.get("req"))
_service_name = "vm_manager"
host = config_util.get_config(req, _service_name, "host")
port = config_util.get_config(req, _service_name, "port")
url = "http://{}:{}{}".format(host, port, uri)
kwargs["method"] = method
kwargs["url"] = url
return wrapped(*args, **kwargs)
return wrapper
------------------------------------------------------example
@requested(method="POST",
uri="/v1/x1Resouce/instance/delete"
)
def resource_delete(req, resource_id, **kwargs):
"""
resource delete
"""
data = {
"instanceIds": [
resource_id
],
"userId": ""
}
headers = {"x_username": "x1_cli"}
r = http_util.request(headers=headers, json=data, **kwargs)
return r2 常见问题
2.1 Unicode 解码错误
错误信息
File "py_menu.py", line 46, in main
self.first_menu()
File "py_menu.py", line 102, in first_menu
self.secondary_menu()
File "py_menu.py", line 132, in secondary_menu
fun[1](self.screen, **func_kargs)
File "/home/work/chunfeng/sinan_cli_bak/three_page.py", line 65, in select_node
return _exe_command(screen, service, offsets, action)
File "/home/work/chunfeng/sinan_cli_bak/three_page.py", line 31, in _exe_command
res = http_util.post_json("http://{ip}:5000/sinan/operation".format(ip=nodes[int(offset)]["ip"]), data=data)
File "/home/work/chunfeng/sinan_cli_bak/w_lib/http_util.py", line 99, in post_json
return RequestTool(url, **kwargs)
File "/home/work/chunfeng/sinan_cli_bak/w_lib/http_util.py", line 297, in __init__
res_attr=json.dumps(self.header)
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe6 in position 12: ordinal not in range(128)程序中添加如下配置
import sys
reload(sys)
sys.setdefaultencoding('utf8')异常原因
ERROR: 12-20 13:43:40: http_util.py:303 * 140336365434624 [file=/home/work/chunfeng/sinan_cli/three_page.py:_exe_command:31 type=http_POST req_path=http://xx.xx.xx.xx:5000/sinan/operation req_data={'action': 'status', 'service': 'redis_proxy', 'offset': 2} cost=0.000272 is_success=False err_no=-1 err_msg=[Errno 111] 拒绝连接 res_len=0 res_data= res_attr={}]
err_msg 中含有中文Last updated