Полезная нагрузка ответа не завершена с использованием asyncio/aiohttp. ⇐ Python

[Anonymous]

Я написал сценарий Python 3.7, который асинхронно (asyncio 3.4.3 и aiohttp 3.5.4) создает массовый API Salesforce Задание/пакет (v45.0), использующее несколько объектов, запрашиваемых одним оператором SOQL, каждый ожидает завершения пакетов, после завершения загружает (потоковую передачу) результаты в сервер, выполняет некоторые преобразования данных, а затем, наконец, синхронно загружает результаты в SQL Server 2016 SP1 (13.0.4560.0). У меня было много успешных пробных запусков с этим, и я думал, что он работает отлично, однако в последнее время я начал периодически получать следующую ошибку и не знаю, как ее исправить, поскольку существует очень мало отчетов/решений по этой проблеме. в Интернете:

aiohttp.client_Exceptions.ClientPayloadError: полезные данные ответа не завершено

Пример фрагмента кода:

импортировать asyncio,aiohttp,aiofiles из simple_salesforce импортировать Salesforce из xml.etree импортировать ElementTree #Установим сеанс с помощью модуля simple_salesforce sf = Salesforce(имя пользователя=имя пользователя, пароль=пароль, Security_token = SecurityToken, идентификатор организации = идентификатор организации) sfAPIURL = 'https://myinstance.salesforce.com/servi ... /45.0/job/' sfDataPath = 'C:/Salesforce/Data/' #Словарь для хранения информации об объекте/задании/пакете во время выполнения сценария объектСловарь = {'Аккаунт': {'задание': {'batch': {'id': '8596P00000ihwpJulI', 'results': ['8596V00000Bo9iU'], 'state': 'Завершено'}, 'id': '8752R00000iUjtReqS'}, 'soql': 'выберите идентификатор, имя из учетной записи'}, 'Контакт': {'работа': {'batch': {'id': '9874G00000iJnBbVgg', 'results': ['7410t00000Ao9vp'], 'state': 'Завершено'}, 'id': '8800o00000POIkLlLa'}, 'soql': 'выберите идентификатор, имя из контакта'}} асинхронное определение извлечения результатов (jobId, BatchId, sfObject): заголовки = {"X-SFDC-Session": sf.session_id, 'Content-Encoding': 'gzip'} асинхронно с aiohttp.ClientSession() в качестве сеанса: асинхронно с session.get(url=f'{sfAPIURL}{jobId}/batch/{batchId}/result', headers=headers) как r: данные = ждут r.text() BatchResults = ElementTree.fromstring(data) #список результатов пакета для resultID в пакетеResults: асинхронно с session.get(url=f'{sfAPIURL}{jobId}/batch/{batchId}/result/{resultID.text}', headers=headers, timeout=None) как r: асинхронно с aiofiles.open(f'{sfDataPath}{sfObject}_TEMP_JOB_{jobId}_BATCH_{batchId}_RESULT_{resultID.text}.csv', 'wb') в качестве выходного файла: #сохранить во временном файле для дальнейших манипуляций пока правда: кусок = ожидание r.content.read(81920) если не кусок: перерыв ожидайте outfile.write(кусок) асинхронное определение asyncDownload(): await asyncio.gather(*[retriveResults(objectDictionary[sfObject]['job']['id'], objectDictionary[sfObject]['job']['batch']['id'], sfObject) для sfObject в objectDictionary ]) если __name__ == "__main__": asyncio.run(asyncDownload()) Обратная трассировка (строки ошибок не будут соответствовать приведенному выше фрагменту кода):

Обратная связь (последний вызов):

Файл «C:\Code\salesforce.py», строка 252, в asyncio.run(asyncDownload())

Файл «C:\Program Files\Python37\lib\asyncio\runners.py», строка 43, в бегать возврат цикла.run_until_complete(main)

Файл «C:\Program Files\Python37\lib\asyncio\base_events.py», строка 584, в run_until_complete вернуть будущее.результат()

Файл «C:\Code\salesforce.py», строка 241, в asyncDownload await asyncio.gather(*[getResults(objectDictionary[sfObject]['job']['id'], objectDictionary[sfObject]['job']['batch']['id'], sfObject) для sfObject в objectDictionary])

Файл «C:\Code\salesforce.py», строка 183, в получить результаты chunk = await r.content.read(81920)

Файл "C:\Program Files\Python37\lib\site-packages\aiohttp\streams.py", строка 369, в читать await self._wait('read')

Файл "C:\Program Files\Python37\lib\site-packages\aiohttp\streams.py", строка 297, в _ждать дождаться официанта

aiohttp.client_Exceptions.ClientPayloadError: полезные данные ответа не завершено

Похоже, что корень проблемы начинается с r.content.read(81920), который должен передавать данные фрагментами по 81920 байт, но это все, что я могу понять.

Я не думаю, что это проблема с сетью с моей стороны, поскольку на этом сервере есть другие небольшие задания, подключенные к внешним источникам, которые завершаются без проблем во время выполнения этого задания. Кто-нибудь знает, что здесь происходит?

Спасибо!

-Изменить:

Я попробовал iter_any() вместо read() и все равно получаю ту же ошибку...

async для данных в r.content.iter_any(): ожидайте outfile.write(данные) Я попробовал readline() и все равно получаю ту же ошибку...

async для строки в r.content.readline(): ожидайте outfile.write(строка) С тех пор я работал над некоторыми функциями повтора в части кода, обрабатывающей ошибки (не включенной в исходную задачу), что в конечном итоге позволяет завершить задания. Ошибки полезной нагрузки все еще происходят, и это по-прежнему является основной проблемой, но повторная попытка загрузки оказалась успешным решением. Проблема все еще сохраняется, если кто-нибудь сможет предоставить дополнительную информацию.