使用 asyncio/aiohttp 未完成响应有效负载-解网

问：

我编写了一个脚本，该脚本使用多个对象异步创建批量 API 作业/批处理，每个对象由单个语句查询，等待批处理完成，完成后将结果下载（流式传输）到服务器，进行一些数据转换，最后将结果同步上传到 .我已经对此进行了大量成功的试运行，并认为它运行良好，但是，我最近开始间歇性地收到以下错误，并且对如何解决有点不知所措，因为网络上很少有这样的报告/解决方案：Python 3.7(asyncio 3.4.3 and aiohttp 3.5.4)Salesforce(v45.0)SOQLSQL Server 2016 SP1 (13.0.4560.0)

aiohttp.client_exceptions。ClientPayloadError：响应有效负载不是完成

示例代码片段：

import asyncio,aiohttp,aiofiles
from simple_salesforce import Salesforce
from xml.etree import ElementTree

#Establish a session using the simple_salesforce module
sf = Salesforce(username=username,
                password=password,
                security_token=securityToken,
                organizationId=organizationId)
sfAPIURL = 'https://myinstance.salesforce.com/services/async/45.0/job/'
sfDataPath = 'C:/Salesforce/Data/'

#Dictionary to store information for the object/job/batch while the script is executing
objectDictionary = 
{'Account': {'job':
                {'batch': {'id': '8596P00000ihwpJulI','results': ['8596V00000Bo9iU'],'state': 'Completed'},
             'id': '8752R00000iUjtReqS'},
             'soql': 'select Id,Name from Account'},

 'Contact': {'job':
                {'batch': {'id': '9874G00000iJnBbVgg','results': ['7410t00000Ao9vp'],'state': 'Completed'},
             'id': '8800o00000POIkLlLa'},
             'soql': 'select Id,Name from Contact'}}

async def retrieveResults(jobId, batchId, sfObject):
    headers = {"X-SFDC-Session": sf.session_id, 'Content-Encoding': 'gzip'}
    async with aiohttp.ClientSession() as session:
        async with session.get(url=f'{sfAPIURL}{jobId}/batch/{batchId}/result', headers=headers) as r:
            data = await r.text()
            batchResults = ElementTree.fromstring(data) #list of batch results
            for resultID in batchResults:
                async with session.get(url=f'{sfAPIURL}{jobId}/batch/{batchId}/result/{resultID.text}', headers=headers, timeout=None) as r:
                    async with aiofiles.open(f'{sfDataPath}{sfObject}_TEMP_JOB_{jobId}_BATCH_{batchId}_RESULT_{resultID.text}.csv', 'wb') as outfile: #save in temporary file for manipulation later
                        while True:
                            chunk = await r.content.read(81920)
                            if not chunk:
                                break
                            await outfile.write(chunk)

async def asyncDownload():
    await asyncio.gather(*[retrieveResults(objectDictionary[sfObject]['job']['id'], objectDictionary[sfObject]['job']['batch']['id'], sfObject) for sfObject in objectDictionary])

if __name__ == "__main__":
    asyncio.run(asyncDownload())

回溯（错误行与上面的代码片段不匹配）：

回溯（最近一次调用最后一次）：

文件“C：\Code\salesforce.py”，第 252 行，在 asyncio.run（asyncDownload（））

文件“C：\Program Files\Python37\lib\asyncio\runners.py”，第 43 行，在跑回程loop.run_until_complete（主）

文件“C：\Program Files\Python37\lib\asyncio\base_events.py”，行 584，在 run_until_complete 返回 future.result（）

文件“C：\Code\salesforce.py”，第 241 行，在 asyncDownload 中 await asyncio.gather（*[retrieveResults（objectDictionary[sfObject]['job']['id']， objectDictionary[sfObject]['job']['batch']['id']， sfObject） for objectDictionary 中的 sfObject]）

文件“C：\Code\salesforce.py”，第 183 行，在检索结果块 = 等待 r.content.read（81920）

文件“C：\Program Files\Python37\lib\site-packages\aiohttp\streams.py“，第 369 行，在读等待self._wait（'读取'）

文件“C：\Program Files\Python37\lib\site-packages\aiohttp\streams.py“，第 297 行，在 _等等待服务员

aiohttp.client_exceptions。ClientPayloadError：响应有效负载不是完成

问题的根源似乎始于 81920 字节块中的流式传输数据，但据我所知，这已经是我能做到的了。r.content.read(81920)

我不认为这是我这边的网络问题，因为还有其他小作业连接到此服务器上的外部源，这些作业在此作业运行时没有问题。有谁知道这是怎么回事？

谢谢！

-编辑：

我已经尝试过，但仍然遇到同样的错误......iter_any()read()

async for data in r.content.iter_any():
    await outfile.write(data)

我试过了，但仍然遇到同样的错误......readline()

async for line in r.content.readline():
    await outfile.write(line)

从那以后，我在代码的错误处理部分（未包含在原始问题中）中使用了一些重试功能，这最终允许作业完成。有效负载错误仍在发生，这仍然是主要问题，但重试下载是一种成功的解决方法。如果有人能够提供进一步的信息，问题仍然存在。

python-3.x 异步 salesforce aiohttp

不知道，对不起。如果您从 Postman、curl、SF Workbench 发出相同的请求怎么办？如果它在沙盒中似乎运行良好，但在生产/开发人员版本中死亡 - 也许您正在用尽滚动的 24 小时 API 请求限制？（理想情况下，您会看到一个关于它的 HTTP 标头）

0赞 gbeaven 5/30/2019

我应该补充一点，我相信这会在数据流式传输到磁盘时发生（该部分），因为将开始 200Mb 的下载，然后在下载过程中的某个时间随机出现错误，但并非总是如此。我知道我没有用尽 API 限制 - 我经常关注它，并且使用率始终低于 5%。我将定期尝试，看看我是否至少可以完成我的下载，只是不喜欢失去异步功能。while True:requests

0赞 JerodG 11/12/2019

我有一个脚本抛出相同的错误，并且正在检索已成功运行数月的 JSON 数据。对我来说，问题出在服务器上;磁盘已满。一旦我清理了一些空间，它又开始正常工作了。您可能需要联系服务器管理员进行检查。

答：

0赞 Louis LIETAER 9/7/2021 #1

嗨，您是否尝试在以下位置插入 await asyncio.sleep（0）：

                    ...
                    while True:
                        chunk = await r.content.read(81920)
                        await asyncio.sleep(0)
                        if not chunk:
                            break
                        await outfile.write(chunk)
                    ...

    async def retrieveResults(session, jobId, batchId, sfObject):
        headers = {"X-SFDC-Session": sf.session_id, 'Content-Encoding': 'gzip'}
        
        async with session.get(url=f'{sfAPIURL}{jobId}/batch/{batchId}/result', headers=headers) as r:
            data = await r.text()
            batchResults = ElementTree.fromstring(data)  # list of batch results
    
            for resultID in batchResults:
                async with session.get(
                    url=f'{sfAPIURL}{jobId}/batch/{batchId}/result/{resultID.text}',
                    headers=headers,
                    timeout=None
                ) as r:
                    async with aiofiles.open(
                        f'{sfDataPath}{sfObject}_TEMP_JOB_{jobId}_BATCH_{batchId}_RESULT_{resultID.text}.csv',
                        'wb'
                    ) as outfile:
                        while True:
                            chunk = await r.content.read(81920)
                            if not chunk:
                                break
                            await outfile.write(chunk)

修改 async def downloadResults：将 async 与 as session：
一起使用，以在函数的上下文中创建会话。将会话传递给 retrieveResults 函数。aiohttp.ClientSession()

    async def asyncDownload():
        async with aiohttp.ClientSession() as session:
            tasks = [
                retrieveResults(session, objectDictionary[sfObject]['job']['id'], objectDictionary[sfObject]['job']['batch']['id'], sfObject)
                for sfObject in objectDictionary
            ]
            await asyncio.gather(*tasks)
    
    if __name__ == "__main__":
        asyncio.run(asyncDownload())

上一个：有什么方法可以更改同步 Windows API SendARP 的行为吗？

下一个：如何使用 GRPC c++ 读取异步服务器端流式处理

使用 asyncio/aiohttp 未完成响应有效负载

Response payload is not completed using asyncio/aiohttp

评论

评论

评论