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 接口調用

原創文章,作者:投稿專員,如若轉載,請註明出處:https://www.506064.com/zh-hant/n/281141.html

(0)
打賞 微信掃一掃 微信掃一掃 支付寶掃一掃 支付寶掃一掃
投稿專員的頭像投稿專員
上一篇 2024-12-21 13:16
下一篇 2024-12-21 13:16

相關推薦

發表回復

登錄後才能評論