我已經寫了一個API程序來測試張貼到HTTP RESTAPI與JwToken認證。在這種情況下,它是用於患者管理系統,並且我正在生成預約。 API業務規則不允許同時進行重複預訂。Python的POST請求的HTTP RESTAPI與JwToken認證產生重複的帖子
我使用一個Python 3.5.3虛擬環境(在pycharm IDE寫),然後運行使用Pytest框架的工作我的測試。 也使用PyJWT 1.5.2。 ,請求2.18.3,安裝simplejson 3.11.1和urllib3 1.22(我假設請求正在使用urllib3)。我使用simplejson.dumps而不是普通的json.dumps,因爲虛擬環境沒有這個庫,我添加它時遇到了麻煩。據我所知,simplejson對轉儲過程具有相同的功能。
使用下面的代碼,我發現我可以運行requests.post調用成功提供一個JSON數據有效載荷,並生成一個帖子,但隨後似乎隨後進行第二次後,其產生409衝突的錯誤。我可以訪問日誌數據庫中的相應API服務器上,可以看到,它實際上已經試圖發佈兩次,但我不能工作了,爲什麼發生這種情況,我覺得有東西在請求庫,被稱爲兩次。可能是由於我發佈了Json。
輸出看起來是這樣的:
https://targerserver.url.com.au/API/Core/v2.1/appointment/
200
{'statusMessages': [], 'appointment': {'startDateTime': '2017-08-15T11:00:00 +10:00', 'appointmentReferenceNumber': '39960337', 'notes': '', 'clients': [{'clientId': 'abeff2be-ce6e-4324-9b57-e28ab7967b6c'}], 'status': 'Booked', 'locationId': 'd8d4fe7c-765a-46a3-a389-54ce298a27e9', 'notifyPractitioner': False, 'endDateTime': '2017-08-15T11:30:00 +10:00', 'subject': 'Jim Beam ', 'appointmentId': '08b37ce3-25e1-4e2a-9bb7-9ec2d716f83b', 'practitioner': {'practitionerId': 'a630f4ad-8b4b-4e06-8cee-7db56ba8b9bf'}}}
collected 1 item
test_PMSAPI_availability.py https://targerserver.url.com.au/API/Core/v2.1/appointment/
409
的json需要一個對象(這是一個字典),也爲另一場鍵(即中有一個條目)的列表,我想知道如果請求庫沒有處理這個。這是json的樣子
payload_str = {"startDateTime":"2017-08-15T11:00+10:00","endDateTime":"2017-08-15T11:30+10:00","practitioner": {"practitionerId":"a630f4ad-8b4b-4e06-8cee-7db56ba8b9bf"}, "locationId":"d8d4fe7c-765a-46a3-a389-54ce298a27e9","clients":[{"clientId":"abeff2be-ce6e-4324-9b57-e28ab7967b6c"}]}
我已經已經爲獲得相同的系統上調用的工作類似的代碼,但張貼的Json似乎真的是有問題的樣本。我們還有其他工具可以調用相同的API端點,但似乎沒有這種麻煩。日誌會建議我提供的JSON數據與具有相同數據的另一個工具的數據完全相同。
我可以從放出來看看是收到的初始響應成功的200碼,但這時如果查詢response.status_code它已成爲一個409響應。我也試過在響應請求沒有做任何處理的情況下導致請求重新發送併產生衝突。
我的代碼如下所示:
import jwt
import _datetime
import requests
import simplejson
from requests.packages.urllib3.exceptions import InsecureRequestWarning
from string import Template
def app_undertest_credentials(keyname):
app_credentials = {'app_consumer_id': 'MyTestApp',
'app_consumer_secret': 'where my secret goes',
'app_access_token': 'where my access token goes',
'base_url': 'https://targerserver.url.com.au'
}
return app_credentials.get(keyname)
def end_points_dict(keynameStr, versionStr):
end_points = {'location': '/API/Core/$version/location/',
'practitioner': '/API/Core/$version/practitioner/',
'availabilityslot': '/API/Core/$version/AvailabilitySlot/',
'client': '/API/Core/$version/client/',
'healthfundproviderlist': '/API/Core/$version/healthfundproviderlist/',
'timezone': '/API/Core/$version/timezone/',
'clientgroup': '/API/Core/$version/clientgroup/',
'appointment': '/API/Core/$version/appointment/'
}
lower_keynameStr = keynameStr.lower()
url_extension_no_version = Template(end_points.get(lower_keynameStr))
url_extension_with_version = url_extension_no_version.safe_substitute(version=versionStr)
return url_extension_with_version
def test_api_appointment_post():
# Set Client app credentials
app_consumer_id = app_undertest_credentials('app_consumer_id')
app_consumer_secret = app_undertest_credentials('app_consumer_secret')
app_access_token = app_undertest_credentials('app_access_token')
base_url = app_undertest_credentials('base_url')
end_point_url_sfx_str = end_points_dict('Appointment', 'v2.1')
httpmethod = 'POST'
# Create dictionary for json post payload
data_payload = {'startDateTime':'2017-08-15T11:00+10:00',
'endDateTime':'2017-08-15T11:30+10:00',
'practitioner': {'practitionerId':'a630f4ad-8b4b-4e06-8cee-7db56ba8b9bf'},
'locationId': 'd8d4fe7c-765a-46a3-a389-54ce298a27e9',
'clients': [{'clientId':'abeff2be-ce6e-4324-9b57-e28ab7967b6c'}]
# Create claims dictionary payload for generation of JwToken
claims = {
'iss': 'http://myappsdomain.com.au',
'aud': 'https://targetservers.domain.com.au',
'nbf': _datetime.datetime.utcnow(),
'exp': _datetime.datetime.utcnow() + _datetime.timedelta(seconds=60),
'consumerId': app_consumer_id,
'accessToken': app_access_token,
'url': base_url + end_point_url_sfx_str,
'httpMethod': http_method
}
#create jwtoken and then convert to string
encoded_jwt_byte = jwt.encode(claim_payload, app_consumer_secret, algorithm='HS256')
jwt_str = encoded_jwt_byte.decode()
#Create authentication header
headers = {'Authorization': 'JwToken' + ' ' + jwt_str, 'content-type': 'application/json'}
uri = base_url + end_point_url_sfx_str
response = requests.post(uri, headers=headers, json=datapayload)
print(response.status)
print(response.json())
response.close()
我使用Wireshark來找出我的電話是實際發送考慮,但我懷疑,只會告訴我的電話被髮送兩次
[編輯]您正在使用的Python模塊和版本的問題,無論是'python_jwt'還是'pyjwt'? – stovfl
我已經在其他模塊中添加版本正在使用PyJWT 1.5.2。,請求2.18.3,simplejson 3.11.1和urllib3 1.22 – Roochiedoor