使用simple_salesforce向Salesforce推送数据时遇SalesforceMalformedRequest错误

我之前也碰到过一模一样的问题，这个SalesforceMalformedRequest带着InvalidBatch错误，本质就是Salesforce没法处理你发过去的批量数据，大概率是字段匹配、数据格式或者必填项的问题。咱们一步步来排查：

先明确错误来源 #

你收到的错误提示：

SalesforceMalformedRequest: Malformed request https://subhotutorial-dev-ed.my.salesforce.com/services/async/38.0/job/7500o00000HtWP6AAN/batch/7510o00000Q15TnAAJ/result . Response content: {'exceptionCode': 'InvalidBatch', 'exceptionMessage': 'Records not processed'}

这个错误说明整个批量都没被处理，不是个别记录的问题，所以先从整体数据格式和字段匹配入手。

具体排查步骤 #

1. 核对字段名和Salesforce API名称完全一致 #

这是最常见的坑！Salesforce的自定义字段API名称是驼峰格式+__c后缀（比如Birth_Date__c），而你的DataFrame里用的是小写下划线（birth_date），两者不匹配的话Salesforce根本认不出字段，直接拒绝整个批量。

你可以登录Salesforce，进入Delivery_Detail__c对象的字段设置页面，把所有需要用到的字段API名称抄下来，然后在代码里做一个映射：

# 把DataFrame列名映射到Salesforce API字段名
field_mapping = {
    'birth_date': 'Birth_Date__c',
    'encounter_start_date': 'Encounter_Start_Date__c',
    'encounter_end_date': 'Encounter_End_Date__c',
    # 其他字段也要一一对应，比如name对应Name，或者其他自定义字段
}

bulk_data = []
for row in staging_df.itertuples():
    d = row._asdict()
    del d['Index']
    # 按映射转换字段名
    mapped_record = {field_mapping[k]: v for k, v in d.items() if k in field_mapping}
    bulk_data.append(mapped_record)

2. 检查日期格式是否符合Salesforce要求 #

你把日期转成了字符串，但Salesforce要求的日期格式必须是YYYY-MM-DD（比如2024-05-20）。如果你的原始日期格式是MM/DD/YYYY或者其他格式，转成字符串后肯定不被接受。

可以先打印几条日期值确认格式：

print(staging_df['birth_date'].head())

如果格式不对，建议用strftime来标准化：

# 假设birth_date是datetime类型，先转成指定格式的字符串
staging_df['birth_date'] = staging_df['birth_date'].dt.strftime('%Y-%m-%d')
staging_df['encounter_start_date'] = staging_df['encounter_start_date'].dt.strftime('%Y-%m-%d')
staging_df['encounter_end_date'] = staging_df['encounter_end_date'].dt.strftime('%Y-%m-%d')

3. 确认必填字段都有值 #

Salesforce对象通常有必填字段（比如标准的Name字段，或者自定义的必填字段），如果你的批量数据里有任何一条记录缺失必填字段，整个批量都会被拒绝。

你可以在Salesforce的对象设置里查看Delivery_Detail__c的必填字段，然后检查DataFrame里对应的列是否没有空值：

# 检查必填列是否有缺失值
print(staging_df[['必填列1', '必填列2']].isnull().sum())

4. 先测试单条数据插入，获取具体错误 #

批量错误的提示太模糊，不如先拿一条数据测试普通插入，这样能得到更具体的错误信息（比如哪个字段格式错了，哪个字段缺失）：

# 取批量数据里的第一条测试
test_record = bulk_data[0]
try:
    insert_result = sf.Delivery_Detail__c.insert(test_record)
    print("单条插入成功！结果:", insert_result)
except Exception as e:
    print("单条插入失败，具体错误:", e)

这个错误信息会直接告诉你问题出在哪，比批量的InvalidBatch有用多了。

5. 检查批量数据大小 #

Salesforce Bulk API对单批记录数量有默认限制（一般是10000条），如果你的bulk_data里记录太多，可以分成小批量插入：

# 分成每1000条一批
batch_size = 1000
for i in range(0, len(bulk_data), batch_size):
    batch = bulk_data[i:i+batch_size]
    sf.bulk.Delivery_Detail__c.insert(batch)

总结 #

这个InvalidBatch错误基本逃不出上面几个原因，先从字段名匹配和单条测试入手，很快就能定位问题。

内容的提问来源于stack exchange，提问作者Vineet