如何使用 salesforce Bulk API 登录沙箱

How to login to the sandbox using salesforce Bulk API

我正在尝试使用 python 连接到 Salesforce Bulk API。但是,我不想在真正的销售队伍中测试我的代码。我想用我的沙箱进行测试。但是,我不知道如何只连接到沙箱...我尝试添加 sandbox=True 但它不起作用...

import salesforce_bulk

bulk=salesforce_bulk.SalesforceBulk(username="username",password="password")

你检查包安装了吗

此库将使用 salesforce-oauth-request 包(您必须安装)运行 Salesforce OAUTH2 Web 流程和return 访问令牌。

密码你有没有令牌?

老问题,但我今天遇到了同样的问题,所以也许这会对某人有所帮助。

这是一个完整的 hack,但它有效 - 可能更好的 hack 是使用 salesforce-oauth-request(它确实有一个 "sandbox=True" 选项)来做到这一点,但我是通过 beatbox 登录的不管怎样,先试试这个。

要点是您使用 beatbox(它允许您指定 serverUrl)登录到沙盒,然后使用该 sessionId 和 instance_url 通过 salesforce_bulk.

登录
import beatbox
from salesforce_bulk import SalesforceBulk

# log in to sandbox using beatbox
service = beatbox.PythonClient()

service.serverUrl = 'https://test.salesforce.com/services/Soap/u/20.0'
user = 'user@user.com'
password = 'secret'
token = '12345'

service.login(user, password+token)

# the _Client_serverUrl has the instance url + some 
# SOAP stuff, so we need to strip that off
groups = service._Client__serverUrl.split('/')
instance_url = '/'.join(groups[:3])

# now we can use the instance_url and sessionId to 
# log into Salesforce through SalesforceBulk
bulk = sfdc_bulk_connect(instance_url, service.sessionId)

这里的建议可能有点不赞成。通过组合 Salesforce 和 SalesforceBulk 库,我能够进行批量上传。请注意我传递给 api 的域以及需要附加到用户名的 sand_box 名称。

from simple_salesforce import Salesforce
from salesforce_bulk import SalesforceBulk
import json
from salesforce_bulk.util import IteratorBytesIO
from urllib.parse import urlparse

USER = "user@domain.com.<sandbox_name>"
PASS = "pass"
SEC_TOKEN = "token"
DOMAIN = "<domain>--<sandbox_name>.<instance>.my"

sf = Salesforce(username=USER, password=PASS, security_token=SEC_TOKEN, domain=DOMAIN)
bulk = SalesforceBulk(sessionId=sf.session_id, host=sf.sf_instance)
job = bulk.create_query_job("table", contentType='JSON')
batch = bulk.query(job, "select Id,LastName from table limit 5000")
bulk.close_job(job)
while not bulk.is_batch_done(batch):
    sleep(10)

for result in bulk.get_all_results_for_query_batch(batch):
    result = json.load(IteratorBytesIO(result))
    for row in result:
        print(row)