ruby-on-rails - 一个接一个地保存和提交新记录，而不是在 block 的末尾

Question

在我的 Rails 应用程序中，我有一项服务被延迟作业工作人员调用以逐行解析 CSV 文件，并将新记录保存到联系人表中。

当它运行时，一旦它保存了它的第一个联系人，联系人表就会锁定在 postgres 中，并且在完成对每一行的解析之前不会解锁。当它遍历每一行时，它正在保存新的联系人记录或更新现有记录，但它们没有被提交。由于该表已锁定，其他用户无法创建联系人记录。

当服务完成 CSV 的每一行检查后，它会提交所有新的联系人记录 - 它们在数据库中可见，并且联系人表不再锁定。

是否可以在每次循环中为正在处理的 CSV 文件的每一行保存、提交和释放锁，而不是将所有内容都保留到结束？

这是类:

class CsvParsingService

  attr_accessor :csv_file, :contact

  def initialize(csv_file)
    @csv_file = csv_file
    @contact = nil
  end

  def perform
    process_csv
    csv_file.finish_import!
  end

  def process_csv
    parser = ::ImportData::SmartCsvParser.new(csv_file.file_url)

    parser.each do |smart_row|
      csv_file.increment!(:total_parsed_records)
      begin
        self.contact = process_row(smart_row)
      rescue => e
        row_parse_error(smart_row, e)
      end
    end
  rescue => e # parser error or unexpected error
    csv_file.save_import_error(e)
  end

  private

  def process_row(smart_row)
    new_contact, existing_records = smart_row.to_contact
    self.contact = ContactMergingService.new(csv_file.user, new_contact, existing_records).perform
    init_contact_info self.contact

    if contact_valid?
      save_imported_contact(new_contact)
    else
      reject_imported_contact(new_contact, smart_row)
    end
  end

  def contact_valid?
    self.contact.first_name || self.contact.last_name ||
      self.contact.email_addresses.first || self.contact.phone_numbers.first
  end

  def save_imported_contact(new_contact)
    self.contact.save!
    csv_file.increment!(:total_imported_records)
    log_processed_contacts new_contact
  end

  def reject_imported_contact(new_contact, smart_row)
    csv_file.increment!(:total_failed_records)
    csv_file.invalid_records.create!(
      original_row: smart_row.row.to_csv,
      contact_errors: ["Contact rejected. Missing name, email or phone number"]
    )
    log_processed_contacts new_contact
    false
  end

  def row_parse_error(smart_row, e)
    csv_file.increment!(:total_failed_records)
    csv_file.invalid_records.create!(
      original_row: smart_row.row.to_csv,
      contact_errors: contact.try(:errors).try(:full_messages) || [e.inspect]
    )
  end

  def init_contact_info(contact)
    unless contact.persisted?
      contact.user = csv_file.user
      contact.created_by_user = csv_file.user
      contact.import_source = csv_file
    end
    contact.required_salutations_to_set = true # will be used for envelope/letter saluation
  end

  def log_processed_contacts(new_contact)
    Rails.logger.info(
      "[CSV.parsing] Records parsed:: parsed: #{csv_file.total_parsed_records}"\
        " : imported: #{csv_file.total_imported_records} : failed: "\
        "#{csv_file.total_failed_records}"
    )
    Rails.logger.info(
      "[CSV.parsing] Contact- New : #{new_contact.email_addresses.map(&:email)}"\
        " : #{new_contact.first_name} : #{new_contact.last_name} "\
        "#{new_contact.phone_numbers.map(&:number)} :: Old : "\
        "#{self.contact.email_addresses.map(&:email)} :"\
        "#{self.contact.phone_numbers.map(&:number)}\n"
    )
  end

end

Answer 1

@SeanHuber 走在正确的 rails 上。我们正在使用 gem state-machine_activerecord ，工作人员运行 csv_file.import!，它将 csv_file 从 uploaded 状态转换为 processing 状态，并调用 CsvParsingService。

默认情况下，state-machine_activerecord将所有转换包装在 transaction 中.这意味着 CsvParsingService 对数据库所做的每项更改都不会在转换完成之前提交。

解决方案是使用选项use_transactions: false定义状态机

这是 worker :

class ImportCsvFileWorker
  def self.perform(csv_file_id)
    csv_file = CsvFile.find(csv_file_id)

    csv_file.import!
    csv_file.send_report!
  end
end

这是正确配置了状态机的 CsvFile 模型:

require "import_data/smart_csv_parser"

class CsvFile < ActiveRecord::Base
  TwiceImportError = Class.new(StandardError)
  ReportBeforeImportError = Class.new(StandardError)

  belongs_to :user
  has_many :invalid_records, class_name: '::CsvFile::InvalidRecord', dependent: :destroy
  has_many :contacts, as: :import_source

  mount_uploader :file, CsvUploader

  attr_accessor :file_url

  def filename
    self[:file]
  end

  state_machine initial: :uploaded, use_transactions: false do
    state :processing
    state :imported

    event :start_import! do
      transition uploaded: :processing
    end
    after_transition :uploaded => :processing,  do: :parse_data!

    event :finish_import! do
      transition processing: :imported
    end
  end

  def import!(file_url=nil)
    if file_url.nil?
      file_url = Rails.env.development? ? file.path : file.url
    end

    self.file_url = file_url
    raise TwiceImportError, "cannot import same file twice" unless uploaded?

    start_import!
  end

  def import_failed?
    import_result[:error].present?
  end

  def send_report!
    raise ReportBeforeImportError, 'please #import! before reporting' unless imported?
    Mailer.delay.csv_import_report(self)
  end

  def save_import_error(exception)
    import_result[:error_class] = exception.class.to_s
    import_result[:error] = exception.message
    import_result[:backtrace] = exception.backtrace
    import_result_will_change!
    save(validate: false)
  end

  private

  def parse_data!
    binding.pry
    CsvParsingService.new(self).perform
  end

  def initialize(*args, &block)
    super(*args, &block) # NOTE: This *must* be called, otherwise states won't get initialized
  end

end