python访问数据库api接口的方法(python如何写api接口)
1. 适用版本
适用于RPA2020.4以及以上版本,
2. 接口 API 手册
调用方式及字段,请参考论坛手册:
暂未在论坛公开,请向当地厂商技术索取
3. Python 调用代码
如果想用机器人来调 API 接口,可参考以下代码,其它语言逻辑一样:
import json
import time
import requests
from urllib import parse
import hmac
import base64
from hashlib import sha256
def rpa_rest_2020_4(host='',rest_type='',data_json=None,mode='',port=443,accesstoken=None,retry=2):
'''
host:地址,str型,示例:'https://192.168.202.11'
rest_type:str型,接口类型,示例:'/oapi/v1/job/create'
data_json:字典型,发送的报文数据json格式
mode:接口请求方式(get、post、delete及put)
port:int型,https端口
retry:int型,重试次数
返回值:get_field_json:字典型,
'''
get_field_json={'code': 40, 'msg': 'fail,意外情况','result':None}
#url参数转换
def json2Params(data_json):
get_field_json={'code': 41, 'msg': 'fail,转换URL参数失败!','result':None}
try:
data_json=json.loads(data_json)
url_str = ''
nums = 0
max_nums = len(data_json)
for key in data_json:
nums += 1
if nums == max_nums:
url_str += str(key) + '=' + str(data_json[key])
else:
url_str += str(key) + '=' + str(data_json[key]) + '&'
except Exception as e:
print('参数转化失败:',e)
url_str=''
finally:
return url_str
if mode != 'get' and mode != 'put' and mode != 'post' and mode != 'delete':
get_field_json={'code': 42, 'msg': 'fail,mode错误,只能为get、put、post或delete!','result':None}
return get_field_json
#获得签名sign
sign = ''
sign_yc = ''
url = host
if port != 443:
url = url + ':' + str(port)
url += rest_type
timestamp = None
#获取签名值sign:
try:
get_field_json={'code': 41, 'msg': 'fail,获取签名失败!','result':None}
#获得毫秒级时间戳(时间出入不能大于10分钟)
timestamp=str(int(round(time.time() * 1000)))
#处理json数据不为str类型的情况
for key in data_json:
data_json[key]=str(data_json[key])
#json排序
data_json=json.dumps(data_json,sort_keys=True)
#构造源串sign_yc
if mode == 'get':
json_str=json2Params(data_json)
sign_yc=json_str + '&token=' + accesstoken + '×tamp=' + timestamp
url = url+'?'+json_str
else:
sign_yc=str(data_json) + '&token=' + accesstoken + '×tamp=' + timestamp
print('源串:',sign_yc)
print('URL:',url)
#URL编码
url_bm = parse.quote(sign_yc, encoding="utf-8")
#sha256加密密码
byte_key = bytes(accesstoken, encoding="utf-8")
byte_url_bm = bytes(url_bm, encoding="utf-8")
hn256 = hmac.new(byte_key, byte_url_bm, digestmod=sha256)
hh256 = hn256.hexdigest()
#Base64编码
bb64 = base64.b64encode(bytes(hh256, encoding="utf-8"))
#获取sign
sign = str(bb64, "utf-8")
#替换rn
sign=sign.replace('rn', '')
print('签名值sign:',sign)
except Exception as e:
print(e)
return get_field_json
#开始尝试发送API
print('开始尝试发送API')
for count in range(retry):
print('尝试第',count+1,'次API请求任务。')
try:
print('开始尝试:',mode)
#post、delete、put
if mode!='get':
header_dict = {'Content-Type': 'application/json;charset=UTF-8','accesstoken':accesstoken,'timestamp':timestamp,'signature':sign}
if mode=='post':
if rest_type == '/oapi/v1/job/create':
header_dict = {'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8','accesstoken':accesstoken,'timestamp':timestamp,'signature':sign}
json_str=json2Params(data_json)
url = url+'?'+json_str
res = requests.post(url, data=str(data_json), headers=header_dict,verify=False)
else:
json_str=json2Params(data_json)
url = url+'?'+json_str
header_dict = {'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8','accesstoken':accesstoken,'timestamp':timestamp,'signature':sign}
res = requests.post(url, data=str(data_json), headers=header_dict,verify=False)
if mode=='put':
res = requests.put(url, data=str(data_json), headers=header_dict,verify=False)
if mode=='delete':
res = requests.delete(url, data=str(data_json), headers=header_dict,verify=False)
if mode=='get':
header_dict = {'accesstoken':accesstoken,'timestamp':timestamp,'signature':sign}
res = requests.get(url,headers=header_dict,verify=False)
#获取返回值
res_text=res.text
#转化成json
get_field_json=json.loads(res_text)
print('请求成功!')
break
except Exception as e:
print('请求失败:',e)
get_field_json={'code': 40, 'msg': 'fail,发送API失败!','result':None}
return get_field_json
def get_token(host='',port='',accessKey='',secretKey='',retry=2):
'''
host:地址,str型,示例:'https://192.168.202.11'
port:int型,https端口
accessKey:str型,服务平台应用AppKey
secretKey:str型,服务平台应用AppSecret
retry:int型,重试次数
'''
get_field_json={'code': 44, 'msg': 'fail,获取token失败','result':None}
for i in range(retry):
try:
url = host
if port != 443:
url = url + ':' + str(port)
json_str='accessKey='+accessKey+'&secretKey='+secretKey
url = url+'/oapi/v1/token?'+json_str
res = requests.get(url,verify=False)
res_text=res.text
#转化成json
get_field_json=json.loads(res_text)
print('获取token,第'+str(i+1)+'次,成功')
break
except Exception as e:
print('获取token,第'+str(i+1)+'次,失败',e)
time.sleep(1)
return get_field_json
def refresh_token(host='',port='',refleshtoken='',retry=2):
'''
host:地址,str型,示例:'https://192.168.202.11'
port:int型,https端口
refleshtoken:str型,刷新token
retry:int型,重试次数
'''
get_field_json={'code': 44, 'msg': 'fail,刷新token失败','result':None}
for i in range(retry):
try:
url = host
if port != 443:
url = url + ':' + str(port)
json_str='refreshToken='+refleshtoken
url = url+'/oapi/v1/token?'+json_str
res = requests.get(url,verify=False)
res_text=res.text
#转化成json
get_field_json=json.loads(res_text)
print('刷新token,第'+str(i+1)+'次,成功')
break
except Exception as e:
print('刷新token,第'+str(i+1)+'次,失败',e)
time.sleep(1)
return get_field_json
4. 其它平台或客户端调用
4.1 其它平台调用
按照第 4 章的逻辑自行写调用代码即可。
4.2 机器人调用
按照第 4 章添加一个全局函数,在需要调用的地方使用全局函数控件即可。
注:需要提前获取token后调用,如无第三方平台对接,获取的token可存在共享变量里,分配权限调用,参考下图:
-
无相关信息
- 1bat的大数据(BAT的大数据来源)
- 2三星s8屏幕上端2(三星s8屏幕上端2个按键)
- 3三星屏幕坏了如何导出(三星屏幕摔坏了如何导出数据么)
- 4红米3x怎么关闭自动更新(红米k40s怎么关闭自动更新)
- 5微信上防止app自动下载软件怎么办(微信上防止app自动下载软件怎么办啊)
- 6押镖多少钱(押镖一个月有多少储备金)
- 7瑞星个人防火墙胡功能(瑞星个人防火墙协议类型有哪些)
- 8cf现在等级是多少(cf等级2020最新)
- 9老滑头多少条鱼(钓鱼老滑头有什么用)
- 10WPS自动调整语法(wps如何修改语法)
- 11dell控制面板防火墙(dell的防火墙怎么关闭)
- 12丑女技能升多少(丑女技能需要满级吗)
- 13智能家居系统怎么样(智能家居系统好吗)
- 14戴尔屏幕(戴尔屏幕闪烁)
- 15y85屏幕信息(vivoy85息屏显示时间怎么设置)
- 16魅蓝note3屏幕出现方格(魅蓝note屏幕竖条纹)
- 17v8手指按屏幕(触屏手指)
- 18金为液晶广告机(液晶广告机lb420)
- 19三星显示器怎么校色(三星显示器 调色)
- 20hkc显示器dvi音频(hkc显示器有音响么)
- 21康佳液晶智能电视机(康佳液晶智能电视机怎么样)
- 22做液晶画板电脑(做液晶画板电脑怎么操作)
- 23液晶屏极化现象原理(液晶屏极化现象原理是什么)
- 24企业网络安全防火墙(企业网络防护)
- 256splus黑屏屏幕不亮(苹果6s plus屏幕突然黑屏)
- 26充电导致屏幕失灵(充电导致屏幕失灵怎么办)
- 27超极本屏幕旋转(笔记本电脑屏幕旋转,怎么转过来?)
- 28igmp防火墙(防火墙配置ipv6)
- 29荣耀王者多少经验(王者荣耀经验多少一级)
- 30lol老将还剩多少(qg老将)