Запросы Python Post к HTTP RestAPI с аутентификацией JwToken генерируют повторяющиеся записи

Question

Я писал процедуру API для проверки публикации в http RestAPI с аутентификацией JwToken. В этом сценарии это система управления пациентами, и я назначаю встречу. Бизнес-правило API не допускает дублирование заказов одновременно.

Я использую виртуальную среду Python 3.5.3 (пишу в Pycharm IDE) и запускаю свои тесты с использованием фрейма Pytest. Также используется PyJWT 1.5.2., Запросы 2.18.3, simplejson 3.11.1 и urllib3 1.22 установлены (я предполагаю, что Requests использует urllib3). Я использую simplejson.dumps вместо обычного json.dumps, потому что в виртуальной среде этой библиотеки не было, и у меня возникли проблемы с ее добавлением. Насколько я могу сказать, simplejson имеет те же функции, что и процедура dumps.

Используя приведенный ниже код, я обнаружил, что могу выполнить вызов request.post, чтобы успешно доставить полезную нагрузку данных Json и сгенерировать сообщение, но затем кажется, что впоследствии выполняется 2-е сообщение, которое генерирует ошибку конфликта 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"}]}

У меня похожий код, который работал для вызовов Get в той же системе, но публикация 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, чтобы выяснить, что на самом деле отправляет мой вызов, но я подозреваю, что это просто скажет мне, что вызовы отправляются дважды

Answer 1

Комментарий: Но мне требуется строка в кодировке base64 для отправки запроса API.

Относительно источника

segments.append(base64url_encode(signature))
    return base64.urlsafe_b64encode(input).replace(b'=', b'')
return b'.'.join(segments)

все есть base64 и вернулся как bytes, Так что вы должны быть в порядке, используя

jwt_str = str(encoded_jwt_byte)

---

Извините, не могу использовать PyJWT,
Пробовал с python_jwt и работает как положено.

Протестировано с Python:3.4.2 - запросы:2.11.1

Вам действительно нужно encode(... а потом decode()?

#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()

Будет jwt_str не быть таким же, как claim_payload?

Answer 2

Хорошо, нашел причину моей проблемы с повторяющимся сообщением. Это была моя собственная глупая ошибка. Внизу моего файла Python, находящегося в нижней части экрана, я вызвал тестовую функцию и не заметил эту строку (следовательно, я пропустил публикацию в своем коде выше). Это поместило тестовую функцию в полный повторяющийся цикл. ....:(Такая глупая ошибка новичка. Спасибо stovfl за совет по обработке закодированного JwToken.

Оглядываясь назад, я обнаружил, что каждое сообщение о стековом потоке, касающееся дублированных сообщений в API, было связано с тем, что в коде пользователя был цикл, я просто не мог его найти.