苏州网站建设一站通,网络设计报告机房环境,北京建设工程信息网网站,微信公众平台登录页面本文尝试基于文心一言AI大模型#xff0c;编写一段python3程序以获取华为分布式块存储REST接口的实时数据。
一、用文心一言AI大模型将需求转化为样例代码
1、第一次对话#xff1a;“python3写一段从rest服务器获取数据的样例代码” 同时生成了以下注解 这段代码首先定义…本文尝试基于文心一言AI大模型编写一段python3程序以获取华为分布式块存储REST接口的实时数据。
一、用文心一言AI大模型将需求转化为样例代码
1、第一次对话“python3写一段从rest服务器获取数据的样例代码” 同时生成了以下注解 这段代码首先定义了一个函数 get_data_from_rest_server该函数使用 requests.get 方法向指定的 URL 发送 GET 请求。如果请求成功即响应的状态码为 200则将响应的 JSON 数据返回。否则它将打印出错误消息并返回 None。然后我们调用这个函数来获取数据并打印出获取到的数据。注意你需要将 url 替换为你的 REST 服务器的实际 URL。 2、第二次对话“以上请求需带入服务器的认证参数” 同时生成了以下注解 在上述代码中我们通过在 requests.get 方法中传递 auth(username, password) 来提供服务器的认证参数。这样当你发送请求时requests 会自动将认证参数添加到请求中从而完成身份验证过程。 3、第三次对话“以上代码需代入请求消息头” import requests # 定义服务器认证参数的用户名和密码
username your_username
password your_password def get_data_from_rest_server(): # 将此 URL 替换为你的 REST 服务器 URL url http://example.com/api/data # 定义请求头可以包含自定义的头部字段和值 headers { Content-Type: application/json, # 声明请求的内容类型为 JSON Authorization: fBasic {username}:{password}, # 添加基本的身份验证头部字段 } # 使用 GET 方法请求数据并携带服务器认证参数和请求头 response requests.get(url, headersheaders) # 检查响应的状态码如果请求成功状态码为 200则处理响应的数据 if response.status_code 200: data response.json() return data else: print(f请求失败状态码 {response.status_code}) return None # 调用函数获取数据
data get_data_from_rest_server()
if data is not None: print(data)
注解如下 在上述代码中我们通过创建一个字典对象来定义请求头其中包含了一个 Content-Type 字段和一个 Authorization 字段。Content-Type 字段声明了请求的内容类型为 JSON而 Authorization 字段包含了基本的身份验证值。然后我们将这个字典对象传递给 requests.get 方法的 headers 参数从而将请求头添加到请求中。 通过和文心一言AI大模型的三次对话我们大致了解了用python 的 requests 库从 REST 服务器获取数据的基本方法。
二、核查华为分布式块存储REST接口规范
1、RESTful API的请求必须包括如下消息头 2、拟测试请求接口官方样例 查询存储池 接口描述 查询存储池。 URI 路径 https://${ip}:${port}/dsware/service/cluster/storagepool/queryStoragePool 访问方法 GET 执行实例 Request: /dsware/service/cluster/storagepool/queryStorageNodeInfo?poolId0 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 } ] } 三、代码实写
根据需求进行代码编写并多次测试最终通过的代码如下以下代码存为名为“healthck.py”的文件 #!/usr/bin/python3
#codingutf-8##__author__daijianbingimport requests
import jsondef authenticate(username, password,url_auth):# 构建认证数据auth_data {user_name: username,password: password}# 将认证数据转换为JSON格式auth_json json.dumps(auth_data)#禁止未经验证的HTTPSrequests.packages.urllib3.disable_warnings()# 发送POST请求到REST API进行身份验证response requests.post(url_auth,verify False, dataauth_json, headers{Content-Type: application/json})# 检查响应状态码如果成功则返回认证令牌否则抛出异常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):# 构建认证数据headers {Connection: keep-alive,X-Auth-Token: token}#print(headers)# 发送请求到REST APIresponse requests.get(url, verify False, headersheaders)# 检查响应状态码如果成功则返回认证令牌否则抛出异常if response.status_code 200:#print(response.json())return response.json()[storagePools]else:raise Exception(Authentication failed)def queryPoolstatus(token,url,poolId):# 构建认证数据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)# 发送T请求到REST APIresponse requests.get(url, verify False, paramsparams, headersheaders)# 检查响应状态码如果成功则返回认证令牌否则抛出异常if response.status_code 200:#print(response.json())return response.json()[status]else:raise Exception(Authentication failed)def queryAlarms(token,url):# 构建认证数据headers {Connection: keep-alive,X-Auth-Token: token}params {filter:alarmStatus::1,}params_json json.dumps(params)# 发送请求到REST APIresponse requests.get(url, verify False, paramsparams, headersheaders)# 检查响应状态码如果成功则返回认证令牌否则抛出异常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/queryStoragePoolpools queryStoragePool(token,url_pool)url_poolstatushostport/dsware/service/cluster/storagepool/queryStatusfor pool in pools:poolstatusqueryPoolstatus(token,url_poolstatus,pool[poolId])url_alarms hostport/api/v2/common/alarm_countAlarmsqueryAlarms(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 本段代码实现了存储REST接口认证登录查询全部的pool并列出poolid,poolname,pool状态输出当前未恢复告警数并返回本次的检查状态1为检查成功。
