<commit_before>require 'restforce'
require 'salesforce_bulk_api'
require 'remi/sf_bulk_helper'

module Remi
  module DataSubject::Salesforce
    def field_symbolizer
      Remi::FieldSymbolizers[:salesforce]
    end

    def restforce_client
      @restforce_client ||= begin
        client = Restforce.new(@credentials)

        #run a dummy query to initiate a connection. Workaround for Bulk API problem
        # https://github.com/yatish27/salesforce_bulk_api/issues/33
        client.query('SELECT Id FROM Contact LIMIT 1')
        client
      end
    end
  end


  class DataSource::Salesforce < Remi::DataSubject
    include Remi::DataSubject::DataSource
    include Remi::DataSubject::Salesforce

    def initialize(*args, **kargs, &block)
      super
      init_salesforce(*args, **kargs, &block)
    end

    # Public: Called to extract data from the source.
    #
    # Returns data in a format that can be used to create a dataframe.
    def extract!
      @extract = sf_bulk.query(@sfo, @query, 10000)

      check_for_errors(@extract)
      @extract
    end

    def sf_bulk
      @sf_bulk ||= SalesforceBulkApi::Api.new(restforce_client).tap { |o| o.connection.set_status_throttle(5) }
    end

    # Public: Converts extracted data to a dataframe.
    # Currently only supports Daru DataFrames.
    #
    # Returns a Remi::DataFrame
    def to_dataframe
      @logger.info "Converting salesforce query results to a dataframe"

      hash_array = {}
      extract['batches'].each do |batch|
        next unless batch['response']

        batch['response'].each do |record|
          record.each do |field, value|
            next if ['xsi:type','type'].include? field
            (hash_array[field.to_sym] ||= []) << case value.first
              when Hash
                value.first["xsi:nil"] == "true" ? nil : value.first
              else
                value.first
              end
          end
        end

        # delete raw result at end of processing to free memory
        batch['response'] = nil
      end

      Remi::DataFrame.create(@remi_df_type, hash_array, order: hash_array.keys)
    end


    private

    def init_salesforce(*args, object:, query:, credentials:, api: :bulk, **kargs, &block)
      @sfo = object
      @query = query
      @credentials = credentials
      @api = api
    end

    def check_for_errors(sf_result)
      sf_result['batches'].each do |batch|
        raise "Error with batch #{batch['id']} - #{batch['state']}: #{batch['stateMessage']}" unless batch['state'].first == 'Completed'
      end
    end
  end


  class DataTarget::Salesforce < Remi::DataSubject
    include Remi::DataSubject::DataTarget
    include Remi::DataSubject::Salesforce

    def initialize(*args, **kargs, &block)
      super
      init_salesforce(*args, **kargs, &block)
    end

    # Public: Performs the load operation, regardless of whether it has
    # already executed.
    #
    # Returns true if the load operation was successful
    def load!
      @logger.info "Performing Salesforce #{@operation} on object #{@sfo}"

      if @operation == :update
        Remi::SfBulkHelper::SfBulkUpdate.update(restforce_client, @sfo, df_as_array_of_hashes, logger: @logger)
      elsif @operation == :create
        Remi::SfBulkHelper::SfBulkCreate.create(restforce_client, @sfo, df_as_array_of_hashes, logger: @logger)
      elsif @operation == :upsert
        Remi::SfBulkHelper::SfBulkUpsert.upsert(restforce_client, @sfo, df_as_array_of_hashes, external_id: @external_id, logger: @logger)
      else
        raise ArgumentError, "Unknown operation: #{@operation}"
      end
<commit_msg>FIX - Missed a df to array conversion needed by salesforce<commit_after>require 'restforce'
require 'salesforce_bulk_api'
require 'remi/sf_bulk_helper'

module Remi
  module DataSubject::Salesforce
    def field_symbolizer
      Remi::FieldSymbolizers[:salesforce]
    end

    def restforce_client
      @restforce_client ||= begin
        client = Restforce.new(@credentials)

        #run a dummy query to initiate a connection. Workaround for Bulk API problem
        # https://github.com/yatish27/salesforce_bulk_api/issues/33
        client.query('SELECT Id FROM Contact LIMIT 1')
        client
      end
    end
  end


  class DataSource::Salesforce < Remi::DataSubject
    include Remi::DataSubject::DataSource
    include Remi::DataSubject::Salesforce

    def initialize(*args, **kargs, &block)
      super
      init_salesforce(*args, **kargs, &block)
    end

    # Public: Called to extract data from the source.
    #
    # Returns data in a format that can be used to create a dataframe.
    def extract!
      @extract = sf_bulk.query(@sfo, @query, 10000)

      check_for_errors(@extract)
      @extract
    end

    def sf_bulk
      @sf_bulk ||= SalesforceBulkApi::Api.new(restforce_client).tap { |o| o.connection.set_status_throttle(5) }
    end

    # Public: Converts extracted data to a dataframe.
    # Currently only supports Daru DataFrames.
    #
    # Returns a Remi::DataFrame
    def to_dataframe
      @logger.info "Converting salesforce query results to a dataframe"

      hash_array = {}
      extract['batches'].each do |batch|
        next unless batch['response']

        batch['response'].each do |record|
          record.each do |field, value|
            next if ['xsi:type','type'].include? field
            (hash_array[field.to_sym] ||= []) << case value.first
              when Hash
                value.first["xsi:nil"] == "true" ? nil : value.first
              else
                value.first
              end
          end
        end

        # delete raw result at end of processing to free memory
        batch['response'] = nil
      end

      Remi::DataFrame.create(@remi_df_type, hash_array, order: hash_array.keys)
    end


    private

    def init_salesforce(*args, object:, query:, credentials:, api: :bulk, **kargs, &block)
      @sfo = object
      @query = query
      @credentials = credentials
      @api = api
    end

    def check_for_errors(sf_result)
      sf_result['batches'].each do |batch|
        raise "Error with batch #{batch['id']} - #{batch['state']}: #{batch['stateMessage']}" unless batch['state'].first == 'Completed'
      end
    end
  end


  class DataTarget::Salesforce < Remi::DataSubject
    include Remi::DataSubject::DataTarget
    include Remi::DataSubject::Salesforce

    def initialize(*args, **kargs, &block)
      super
      init_salesforce(*args, **kargs, &block)
    end

    # Public: Performs the load operation, regardless of whether it has
    # already executed.
    #
    # Returns true if the load operation was successful
    def load!
      @logger.info "Performing Salesforce #{@operation} on object #{@sfo}"

      df_as_array_of_hashes = df.to_a[0] # This probably wouldn't work with a non-Daru df
      if @operation == :update
        Remi::SfBulkHelper::SfBulkUpdate.update(restforce_client, @sfo, df_as_array_of_hashes, logger: @logger)
      elsif @operation == :create
        Remi::SfBulkHelper::SfBulkCreate.create(restforce_client, @sfo, df_as_array_of_hashes, logger: @logger)
      elsif @operation == :upsert
        Remi::SfBulkHelper::SfBulkUpsert.upsert(restforce_client, @sfo, df_as_array_of_hashes, external_id: @external_id, logger: @logger)
      else
        raise ArgumentError, "Unknown operation: #{@operation}"
      end

{'fraction_non_alphanumeric': 0.10352750371671847, 'fraction_numerical': 0.003108528179483714, 'mean_word_length': 2.850156087408949, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3728418', 'n_tokens_mistral': 2306, 'n_tokens_neox': 2212, 'n_words': 593}