Python RESTful API 客户机随附了一个 REST 测试库,可协助 RESTful 服务的测试开发。
RESTful 客户机程序示例:
>>> import urllib2
>>> import json
>>> request = urllib2.Request("https://zfsssa.example:215/api/access/v1", "")
>>> request.add_header("X-Auth-User", "rest_user")
>>> request.add_header("X-Auth-Key", "letmein")
>>> response = urllib2.urlopen(request)
>>> response.getcode()
201
>>> info = response.info()
>>>
opener = urllib2.build_opener(urllib2.HTTPHandler)
>>> opener.addheaders = [("X-Auth-Session", info.getheader("X-Auth-Session")),
... ('Content-Type', 'application/json'), ('Accept', 'appplication/json')]
然后,可以使用 opener 打开那些已经预先验证并准备好发送/接收 JSON 数据的请求。
使用以下 Python 代码可以从任何 RESTful API 资源获取数据。
GET 示例:
>>> request = urllib2.Request("https://zfssa:215/api/network/v1/routes")
>>> response = opener.open(request)
>>> response.getcode()
200
>>> body = json.loads(response.read())
>>> print json.dumps(body, sort_keys=True, indent=4)
{
"routes": [
{
"destination": "0.0.0.0",
"family": "IPv4",
"gateway": "10.80.231.1",
"href":
"/api/network/v1/routes/ixgbe0,0.0.0.0,10.80.231.1",
"interface": "ixgbe0",
"mask": 0,
"type": "static"
}
]
}
用于创建新资源的 Python 示例代码:
>>> action = {'category': 'network'}
>>> post_data = json.dumps(action)
>>> request = urllib2.Request("https://zfssa:215/api/alert/v1/actions", post_data)
>>> request.add_header('Content-Type', 'application/json')
>>> response = opener.open(request)
>>> response.getcode()
201
>>> response.info().getheader('Location')
'/api/alert/v1/actions/actions-001'
>>> body = json.loads(response.read())
>>> print json.dumps(body, sort_keys=True, indent=4)
{
"actions": {
"category": "network",
"datalink_failed": true,
"datalink_ok": true,
"href":
"/api/alert/v1/actions/actions-001",
"ip_address_conflict": true,
"ip_address_conflict_resolved": true,
"ip_interface_degraded": true,
"ip_interface_failed":
true,
"ip_interface_ok": true,
"network_port_down": true,
"network_port_up":
true
}
}
用于修改现有资源的 Python 示例代码:
>>> put_data = '{"ip_address_conflict_resolved": false}'
>>>
request = urllib2.Request("https://zfssa:215/api/alert/v1/actions/actions-001", put_data)
>>> request.add_header('Content-Type', 'application/json')
>>> request.get_method = lambda: 'PUT'
>>> response = opener.open(request)
>>> response.getcode()
202
>>> body = json.loads(response.read())
>>> print json.dumps(body, sort_keys=True, indent=4)
{
"actions": {
"category": "network",
"datalink_failed": true,
"datalink_ok": true,
"href":
"/api/alert/v1/actions/actions-001",
"ip_address_conflict": true,
"ip_address_conflict_resolved": false,
"ip_interface_degraded": true,
"ip_interface_failed":
true,
"ip_interface_ok": true,
"network_port_down": true,
"network_port_up":
true
}
}
用于删除现有资源的 Python 示例代码:
>>> request = urllib2.Request("https://zfssa:215/api/alert/v1/actions/actions-001")
>>> request.get_method = lambda: 'DELETE'
>>> response = opener.open(request)
>>> response.getcode()
204