<menu id="ycqsw"></menu><nav id="ycqsw"><code id="ycqsw"></code></nav>
<dd id="ycqsw"><menu id="ycqsw"></menu></dd>
  • <nav id="ycqsw"></nav>
    <menu id="ycqsw"><strong id="ycqsw"></strong></menu>
    <xmp id="ycqsw"><nav id="ycqsw"></nav>
  • 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可存在共享變量里,分配權限調用,參考下圖:

    PRA2020.4 平臺 API 接口調用

    版權聲明:本文內容由互聯網用戶自發貢獻,該文觀點僅代表作者本人。本站僅提供信息存儲空間服務,不擁有所有權,不承擔相關法律責任。如發現本站有涉嫌抄襲侵權/違法違規的內容, 請發送郵件至 舉報,一經查實,本站將立刻刪除。

    發表評論

    登錄后才能評論
    国产精品区一区二区免费