python - 使用 Python 的 Salesforce Bulk API 批量下载

Question

我有一个异步 Python 脚本，可在 Salesforce 中创建批量 API 作业/批处理。批处理完成后，我会下载 csv 文件进行处理。

这是我的问题:使用此异步代码，使用 Python 流式下载一个 ~300 MB csv 文件可能需要 3 分钟以上:

If you're familiar with Salesforce bulk jobs, you can enter your information into the variables below and download your batch results for testing. This is a working example of code provided you enter the necessary information.

import asyncio, aiohttp, aiofiles
from simple_salesforce import Salesforce
from credentials import credentials as cred

sf_data_path = 'C:/Users/[USER NAME]/Desktop/'
job_id = '[18 DIGIT JOB ID]'
batch_id = '[18 DIGIT BATCH ID]'
result_id = '[18 DIGIT RESULT ID]'
instance_name = '[INSTANCE NAME]'
result_url = f'https://{instance_name}.salesforce.com/services/async/45.0/job/{job_id}/batch/{batch_id}/result/{result_id}'

sf = Salesforce(username=['SALESFORCE USERNAME'],
                password=['SALESFORCE PASSWORD'],
                security_token=['SALESFORCE SECURITY TOKEN'],
                organizationId=['SALESFORCE ORGANIZATION ID'])

async def download_results():
    err = None
    retries = 3
    status = 'Not Downloaded'
    for _ in range(retries):
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(url=result_url,
                                       headers={"X-SFDC-Session": sf.session_id, 'Content-Encoding': 'gzip'},
                                       timeout=300) as resp:
                    async with aiofiles.open(f'{sf_data_path}_DOWNLOAD_.csv', 'wb') as outfile:
                        while True:
                            chunk = await resp.content.read(10485760)  # = 10Mb
                            if not chunk:
                                break
                            await outfile.write(chunk)
                    status = 'Downloaded'
        except Exception as e:
            err = e
            retries -= 1
            status = 'Retrying'
            continue
        else:
            break
    else:
        status = 'Failed'
        return err, status, retries

asyncio.run(download_results())

但是，如果我在 Developer Workbench 中下载批处理结果:https://workbench.developerforce.com/asyncStatus.php?jobId= '[18 位作业 ID]' 相同的文件可能会在 5 秒内下载。

显然这里发生了一些我想念的事情。我知道 Workbench 使用 PHP，此功能甚至可用于 Python 吗？我认为异步调用会使下载速度很快，但这似乎并没有使它的下载速度与浏览器中的功能一样快。有什么想法吗？

谢谢!

Answer 1

您可以尝试 curl 请求来获取 csv。此方法与您在工作台中看到的一样快。您可以在此处阅读更多信息:

https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/query_walkthrough.htm

https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/query_get_job_results.htm#example_locator