浙江平板網(wǎng)站建設(shè)關(guān)鍵詞搜索排名
本文嘗試基于文心一言AI大模型，編寫一段python3程序以獲取華為分布式塊存儲REST接口的實時數(shù)據(jù)。
一、用文心一言AI大模型將需求轉(zhuǎn)化為樣例代碼
1、第一次對話：“python3寫一段從rest服務(wù)器獲取數(shù)據(jù)的樣例代碼”
同時生成了以下注解?
這段代碼首先定義了一個函數(shù)?
get_data_from_rest_server，該函數(shù)使用?requests.get?方法向指定的 URL 發(fā)送 GET 請求。如果請求成功（即響應(yīng)的狀態(tài)碼為 200），則將響應(yīng)的 JSON 數(shù)據(jù)返回。否則，它將打印出錯誤消息并返回 None。然后，我們調(diào)用這個函數(shù)來獲取數(shù)據(jù)，并打印出獲取到的數(shù)據(jù)。注意你需要將?url?替換為你的 REST 服務(wù)器的實際 URL。?
??2、第二次對話：“以上請求需帶入服務(wù)器的認證參數(shù)”
同時生成了以下注解?
?在上述代碼中，我們通過在?
requests.get?方法中傳遞?auth=(username, password)?來提供服務(wù)器的認證參數(shù)。這樣，當你發(fā)送請求時，requests?會自動將認證參數(shù)添加到請求中，從而完成身份驗證過程。
3、第三次對話：“以上代碼需代入請求消息頭”
?
import requests # 定義服務(wù)器認證參數(shù)的用戶名和密碼
username = 'your_username'
password = 'your_password' def get_data_from_rest_server(): # 將此 URL 替換為你的 REST 服務(wù)器 URL url = 'http://example.com/api/data' # 定義請求頭，可以包含自定義的頭部字段和值 headers = { 'Content-Type': 'application/json', # 聲明請求的內(nèi)容類型為 JSON 'Authorization': f'Basic {username}:{password}', # 添加基本的身份驗證頭部字段 } # 使用 GET 方法請求數(shù)據(jù)，并攜帶服務(wù)器認證參數(shù)和請求頭 response = requests.get(url, headers=headers) # 檢查響應(yīng)的狀態(tài)碼，如果請求成功（狀態(tài)碼為 200），則處理響應(yīng)的數(shù)據(jù) if response.status_code == 200: data = response.json() return data else: print(f"請求失敗，狀態(tài)碼： {response.status_code}") return None # 調(diào)用函數(shù)獲取數(shù)據(jù)
data = get_data_from_rest_server()
if data is not None: print(data)注解如下?：
在上述代碼中，我們通過創(chuàng)建一個字典對象來定義請求頭，其中包含了一個?
Content-Type?字段和一個?Authorization?字段。Content-Type?字段聲明了請求的內(nèi)容類型為 JSON，而?Authorization?字段包含了基本的身份驗證值。然后，我們將這個字典對象傳遞給?requests.get?方法的?headers?參數(shù)，從而將請求頭添加到請求中。?
?通過和文心一言AI大模型的三次對話，我們大致了解了用python 的?requests?庫從 REST 服務(wù)器獲取數(shù)據(jù)的基本方法。
二、核查華為分布式塊存儲REST接口規(guī)范
1、RESTful API的請求必須包括如下消息頭
2、擬測試請求接口官方樣例
查詢存儲池 接口描述
查詢存儲池。
URI 路徑
https://${ip}:${port}/dsware/service/cluster/storagepool/queryStoragePool
訪問方法
GET
執(zhí)行實例
Request: /dsware/service/cluster/storagepool/queryStorageNodeInfo?poolId=0
Response: {
"result": 0,
"storagePools": [
{
"poolId": 0,
"totalCapacity": 675305,
"usedCapacity": 0,
"reductionInvolvedCapacity": 0,
"allocatedCapacity": 0,
"usedCapacityRate": 0.0,
"deduplicationSaved": 0,
"compressionSaved": 0,
"deduplicationRatio": 1.0,
"compressionRatio": 1.0,
"dataReductionRatio": 1.0,
"thinRate": 0,
"replicationFactor": 3,
"poolName": "rep3",
"poolServerType": 1,
"thinThreshold": 70,
"poolSpec": "normal_pool",
"redundancyPolicy": "replication",
"numDataUnits": 0,
"numParityUnits": 0,
"numFaultTolerance": 0,
"cellSize": 0,
"ecCacheMediaType": "false",
"ecCacheRate": "0",
"ecCacheWbCacheRate": "0",
"compressionAlgorithm": "performance",
"enableAdvanceVolume": "true",
"encryptType": 0,
"supportEncryptForMainStorageMedia": 0,
"serviceType": 1,
"storageMediaType": "ssd_card",
"physicalTotalCapacity": 482361,
"usedCapacityAfterDedup": 0,
"writableCapacity": 482361,
"markDelCapacity": 0,
"bbuInfo": "close",
"fastGcTime": 0
}
]
}
三、代碼實寫
根據(jù)需求進行代碼編寫，并多次測試，最終通過的代碼如下（以下代碼存為名為“healthck.py”的文件）：
#!/usr/bin/python3
#coding=utf-8##__author__='daijianbing'import requests
import jsondef authenticate(username, password,url_auth):# 構(gòu)建認證數(shù)據(jù)auth_data = {'user_name': username,'password': password}# 將認證數(shù)據(jù)轉(zhuǎn)換為JSON格式auth_json = json.dumps(auth_data)#禁止未經(jīng)驗證的HTTPSrequests.packages.urllib3.disable_warnings()# 發(fā)送POST請求到REST API進行身份驗證response = requests.post(url_auth,verify = False, data=auth_json, headers={'Content-Type': 'application/json'})# 檢查響應(yīng)狀態(tài)碼，如果成功則返回認證令牌，否則拋出異常if response.status_code == 200:#print(response.json())return response.json()['data']['x_auth_token']else:connect = 0raise Exception('Authentication failed')def queryStoragePool(token,url):# 構(gòu)建認證數(shù)據(jù)headers = {'Connection': 'keep-alive','X-Auth-Token': token}#print(headers)# 發(fā)送請求到REST APIresponse = requests.get(url, verify = False, headers=headers)# 檢查響應(yīng)狀態(tài)碼，如果成功則返回認證令牌，否則拋出異常if response.status_code == 200:#print(response.json())return response.json()['storagePools']else:raise Exception('Authentication failed')def queryPoolstatus(token,url,poolId):# 構(gòu)建認證數(shù)據(jù)headers = {'Connection': 'keep-alive','X-Auth-Token': token}params = {'poolId':str(poolId),'vbs': 'true'}params_json = json.dumps(params)#print("url:",url,"headers:",headers,"params_json:",params_json)# 發(fā)送T請求到REST APIresponse = requests.get(url, verify = False, params=params, headers=headers)# 檢查響應(yīng)狀態(tài)碼，如果成功則返回認證令牌，否則拋出異常if response.status_code == 200:#print(response.json())return response.json()['status']else:raise Exception('Authentication failed')def queryAlarms(token,url):# 構(gòu)建認證數(shù)據(jù)headers = {'Connection': 'keep-alive','X-Auth-Token': token}params = {'filter':"alarmStatus::1",}params_json = json.dumps(params)# 發(fā)送請求到REST APIresponse = requests.get(url, verify = False, params=params, headers=headers)# 檢查響應(yīng)狀態(tài)碼，如果成功則返回認證令牌，否則拋出異常if response.status_code == 200:#print(response.json())return response.json()['data']else:raise Exception('Authentication failed')# 使用示例
username = 'zabbix'
password = 'passwd'
hostip = '192.168.0.1'
port = '8088'
hostport = "https://"+hostip+":"+port
url_auth = hostport+'/api/v2/aa/sessions'
checkstatus = 1
requests.packages.urllib3.disable_warnings()
try:token = authenticate(username, password, url_auth)print('x_auth_token:', token)url_pool = hostport+'/dsware/service/resource/queryStoragePool'pools = queryStoragePool(token,url_pool)url_poolstatus=hostport+'/dsware/service/cluster/storagepool/queryStatus'for pool in pools:poolstatus=queryPoolstatus(token,url_poolstatus,pool['poolId'])url_alarms = hostport+'/api/v2/common/alarm_count'Alarms=queryAlarms(token,url_alarms)print('poolid:',pool['poolId'],'poolname:',pool['poolName'],'poolstatus:',poolstatus,'alarmcount:',Alarms['count'])
except:checkstatus = 0
finally:print("checkstatus:",checkstatus)print("chech over")四、驗證運行
實際運行:
# python3 --version
Python 3.9.9
# python3 healthck.py
x_auth_token: MzMwMDE0OTkyNkEzMDAxNGUwOTcwMzE2
poolid: 0 poolname: Pool01 poolstatus: 0 alarmcount: 0
checkstatus: 1
chech over本段代碼實現(xiàn)了存儲REST接口認證登錄，查詢?nèi)康膒ool并列出poolid,poolname,pool狀態(tài)，輸出當前未恢復(fù)告警數(shù)并返回本次的檢查狀態(tài)（1為檢查成功）。