connection.rb

module SalesforceBulk
  class Connection
    def initialize(username, password, api_version, sandbox)
      @username = username
      @password = password
      @api_version = api_version
      @sandbox = sandbox
      login
    end

    def login
      response = SalesforceBulk::Http.login(
        @sandbox,
        @username,
        @password,
        @api_version)

      @session_id = response[:session_id]
      @instance = response[:instance]
    end

    def org_id
      @session_id.split('!').first
    end

    def create_job operation, sobject, external_field
      SalesforceBulk::Http.create_job(
        @instance,
        @session_id,
        operation,
        sobject,
        @api_version,
        external_field)[:id]
    end

    def close_job job_id
      SalesforceBulk::Http.close_job(
        @instance,
        @session_id,
        job_id,
        @api_version)[:id]
    end

    def add_query job_id, data_or_soql
      SalesforceBulk::Http.add_batch(
        @instance,
        @session_id,
        job_id,
        data_or_soql,
        @api_version)[:id]
    end

    def query_batch job_id, batch_id
      SalesforceBulk::Http.query_batch(
        @instance,
        @session_id,
        job_id,
        batch_id,
        @api_version,
      )
    end

    def query_batch_result_id job_id, batch_id
      SalesforceBulk::Http.query_batch_result_id(
        @instance,
        @session_id,
        job_id,
        batch_id,
        @api_version,
      )
    end

    def query_batch_result_data job_id, batch_id, result_id
      SalesforceBulk::Http.query_batch_result_data(
        @instance,
        @session_id,
        job_id,
        batch_id,
        result_id,
        @api_version,
      )
    end

    def add_batch job_id, records
      keys = records.first.keys

      rows = keys.to_csv
      records.each do |r|
        fields = []
        keys.each do |k|
          fields.push(r[k])
        end
        rows << fields.to_csv
      end

      SalesforceBulk::Http.add_batch(
        @instance,
        @session_id,
        job_id,
        rows,
        @api_version)[:id]
    end
  end
end

salesforce_bulk.rb

require 'csv'
require 'salesforce_bulk/version'
require 'salesforce_bulk/batch'
require 'salesforce_bulk/http'
require 'salesforce_bulk/connection'

module SalesforceBulk
  class Api
    attr_reader :connection
    SALESFORCE_API_VERSION = '27.0'

    def initialize(username, password, sandbox = false, api_version = SALESFORCE_API_VERSION)
      @connection = SalesforceBulk::Connection.new(username,
        password,
        api_version,
        sandbox)
    end

    def upsert(sobject, records, external_field)
      start_job('upsert', sobject, records, external_field)
    end

    def update(sobject, records)
      start_job('update', sobject, records)
    end

    def create(sobject, records)
      start_job('insert', sobject, records)
    end

    def delete(sobject, records)
      start_job('delete', sobject, records)
    end

    def query(sobject, query)
      job_id = @connection.create_job(
        'query',
        sobject,
        nil
      )
      batch_id = @connection.add_query(job_id, query)
      @connection.close_job job_id
      batch_reference = SalesforceBulk::Batch.new @connection, job_id, batch_id
      batch_reference.init_result_id
      batch_reference.final_status
    end

    private
    def start_job(operation, sobject, records, external_field=nil)
      job_id = @connection.create_job(
        operation,
        sobject,
        external_field)
      batch_id = @connection.add_batch job_id, records
      @connection.close_job job_id
      SalesforceBulk::Batch.new @connection, job_id, batch_id
    end
  end
end