pg_copy.rb

def self.bulk_import(data, table_name, field_list, options={})

    with = options[:with] ? " WITH #{options[:with]}" : ''
    self.with_connection(options[:connection]) do |c|
      #this is in a transaction...
      conn = c.raw_connection
      conn.set_error_verbosity(PG::PQERRORS_VERBOSE)
      #conn.transaction do

      cleaned_up_field_list = field_list.map { |f|
        f.to_s.include?('"') ? f.to_s : f.to_s #"\"#{f}\""
      }.join(',')
      sql                   = "COPY #{table_name} (#{cleaned_up_field_list}) FROM STDIN  #{with}"
                  #puts sql
      conn.exec(sql)
      errmsg = '' # scope this outside of the rescue below so it's visible later
      begin
        data.each_line { |line| conn.put_copy_data(line) }
      rescue Errno => err
        errmsg << "%s while reading copy data: %s" % [err.class.name, err.message]
        # puts "an error occured"
      end

      if errmsg.length > 0
        conn.put_copy_end(errmsg)
        # puts "ERROR #{errmsg}"
      else
        conn.put_copy_end
      end


      while res = conn.get_result
        st = res.res_status(res.result_status)
        unless  [PG::PGRES_COPY_IN, PG::PGRES_COMMAND_OK].include? res.result_status
          #we raise the error so the transaction rolls back automatically.
          raise res.error_message
        end
      end

    end #end #connection

  end