如何使用 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)
我正在尝试使用 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)