やりたいこと

SalesForce の bulk API で csv ファイルによって大量の情報を更新したい。

公式ドキュメント

Bulk API 2.0

大まかな流れ

SFログインでセッションID取得
ジョブ作成
ジョブにデータ追加
ジョブのステータスを更新(ジョブ開始)
ジョブの状態監視
ジョブの結果取得

実装

SFログインでセッションID取得

OAuthでの認証ができるみたいだけど今回は
SOAPのログイン処理でセッションID loginResult.sessionId を取得

ジョブ作成


/// <summary>
/// job作成
/// </summary>
private bool CreateJob()
{
    // エンドポイント
    string url = DOMAIN + "/services/data/v46.0/jobs/ingest";

    // メソッド
    string method = "POST";

    // content-type
    string contentType = "application/json";

    // リクエストヘッダ
    string header = "Authorization: Bearer " + loginResult.sessionId;

    // リクエストボディ
    // JSON形式 でリクエスト作成(シリアライズ処理等は割愛)
    JobCreateRequest jcr = new JobCreateRequest();
    jcr.operation = "upsert";   // 操作 : insert delete update upsert 
    jcr.obj = "Contact";        // 操作対象オブジェクト名
    jcr.contentType = "CSV";    // コンテンツタイプ CSVのみ
    jcr.lineEnding = "CRLF";    // CSVの改行タイプ :  LF CRLF
    jcr.externalIdFieldName = "Id"; // 外部 ID 項目　更新/挿入操作で必須
    string body = CreateJson(jcr);

    // HTTPリクエスト(具体的な実装は割愛)
    var res = HTTPAccess.GetResponse(method, url, body, contentType, header);

    if (res.StatusCode != 200) return false;

    // パースしてジョブIDを取得(具体的な実装は割愛)
    JobInfo jobInfo = ParseJson(res.body);
    this.jobId = jobInfo.id;

    return true;
}

ジョブにデータ追加

base64 エンコードで 150MB 以内の制限があるため 100MB 以下に抑える


/// <summary>
/// データ追加
/// </summary>
private bool AddDataToJob()
{
    // エンドポイント
    string url = DOMAIN + "/services/data/v46.0/jobs/ingest/" + this.jobId + "/batches";

    // メソッド
    string method = "PUT";

    // content-type
    string contentType = "text/csv";

    // リクエストヘッダ
    string header = "Authorization: Bearer " + loginResult.sessionId;

    // リクエストボディ
    // csv形式で入力　 ジョブ作成時に指定した外部 ID 項目は必須
    string body = "Id,LastName,FirstName,AccountId" + "\r\n";
    body += "003000000000000,バルク,太郎,001000000000000" + "\r\n";
    body += "003000000000001,バルク,次郎,001000000000000";

    // HTTPリクエスト(具体的な実装は割愛)
    var res = HTTPAccess.GetResponse(method, url, body, contentType, header);

    // 成功時は201
    return (res.StatusCode == 201);
}

ジョブのステータスを更新(ジョブ開始)


/// <summary>
/// ジョブステータス更新
/// </summary>
/// <param name="state">UploadComplete : データアップロード完了,ジョブをキューに追加 / Aborted : ジョブ中止</param>
/// <returns></returns>
private bool UpdateJobState(string state)
{
    // エンドポイント
    string url = DOMAIN + "/services/data/v46.0/jobs/ingest/" + this.jobId;

    // メソッド
    string method = "PATCH";

    // content-type
    string contentType = "application/json";

    // リクエストヘッダ
    string header = "Authorization: Bearer " + loginResult.sessionId;

    // リクエストボディ
    // state を更新する
    string body = "{\"state\":\"" + state + "\"}";

    // HTTPリクエスト(具体的な実装は割愛)
    var res = HTTPAccess.GetResponse(method, url, body, contentType, header);

    return (res.StatusCode == 200);
}

ジョブの状態監視

いつ処理されるかサーバ次第なので
適当なタイミングで完了チェックが必要になる


/// <summary>
/// ジョブ監視
/// </summary>
private void CheckJob()
{
    while (true) {
        // エンドポイント
        string url = DOMAIN + "/services/data/v46.0/jobs/ingest/" + this.jobId;

        // メソッド
        string method = "GET";

        // リクエストヘッダ
        string header = "Authorization: Bearer " + loginResult.sessionId;

        // HTTPリクエスト(具体的な実装は割愛)
        var res = HTTPAccess.GetResponse(method, url, null, null, header);

        // パースしてジョブIDを取得(具体的な実装は割愛)
        JobInfo jobInfo = ParseJson(res.body);
        if (jobInfo.state == "JobComplete" || jobInfo.state == "Failed")
        {
            // 完了または失敗になるまで監視する
            break;
        }
        else
        {
            Console.WriteLine("まだだよ");
            Thread.Sleep(3000);
        }
    }
}

ジョブの結果取得


/// <summary>
/// ジョブ結果取得
/// </summary>
private bool GetJobResult()
{
    // エンドポイント
    // 成功結果取得
    string url = DOMAIN + "/services/data/v46.0/jobs/ingest/" + this.jobId + "/successfulResults";
    // 失敗結果取得
    //string url = DOMAIN + "/services/data/v46.0/jobs/ingest/" + this.jobId + "/failedResults";

    // メソッド
    string method = "GET";

    // リクエストヘッダ
    string header = "Authorization: Bearer " + loginResult.sessionId;

    // HTTPリクエスト(具体的な実装は割愛)
    var res = HTTPAccess.GetResponse(method, url, null, null, header);

    MessageBox.Show(res.Body);

    return (res.StatusCode == 200);
}

おまけ

API のバージョン取得
ログインしなくても使用可能
エンドポイントのvXX.Xを切り替えて使用する


private void GetAPIVersion()
{
    // エンドポイント
    string url = DOMAIN + "/services/data";

    // メソッド
    string method = "GET";

    // HTTPリクエスト(具体的な実装は割愛)
    var res = HTTPAccess.GetResponse(method, url, null, null, null);

    List<APIVersion> apis = new List<APIVersion>();
    apis = ParseJson(res.body);

    // 必要あればソート
    //apis.Sort((a, b) => string.Compare(a.version, b.version));

    // 最新のバージョンを設定
    this.apiVersion = apis[apis.Count - 1].version;
}

C# で SalesForce の Bulk API を叩く