Class: Spree::ImportRow

Inherits:
Object
  • Object
show all
Defined in:
app/models/spree/import_row.rb

Constant Summary collapse

STALLED_PROCESSING_AFTER =

How long a row may sit in processing before we consider its owning worker dead (OOM, SIGKILL, deploy without graceful drain — none of these trigger a Sidekiq retry). After this window, stalled rows no longer block the import from completing.

1.hour

Instance Method Summary collapse

Instance Method Details

#attribute_by_schema_field(schema_field, mappings = nil) ⇒ Object 



88
89
90
91
92
93
94
95
 
# File 'app/models/spree/import_row.rb', line 88

def attribute_by_schema_field(schema_field, mappings = nil)
  mappings ||= import.mappings.mapped

  mapping = mappings.find { |m| m.schema_field == schema_field }
  return unless mapping&.mapped?

  data_json[mapping.file_column]
end

#bulk_process!(mappings:, schema_fields:) ⇒ Object

Bulk processing mode for large imports. Uses update_columns to skip callbacks, validations, and event publishing.



110
111
112
113
114
115
116
117
118
 
# File 'app/models/spree/import_row.rb', line 110

def bulk_process!(mappings:, schema_fields:)
  update_columns(status: 'processing', updated_at: Time.current)
  processor = import.row_processor_class.new(self, mappings: mappings, schema_fields: schema_fields)
  self.item = processor.process!
  update_columns(status: 'completed', item_type: item.class.name, item_id: item.id, validation_errors: nil, updated_at: Time.current)
rescue StandardError => e
  Rails.error.report(e, handled: true, context: { import_row_id: id }, source: 'spree.core')
  update_columns(status: 'failed', validation_errors: e.message, updated_at: Time.current)
end

#data_json ⇒ Object 



69
70
71
72
73
 
# File 'app/models/spree/import_row.rb', line 69

def data_json
  @data_json ||= JSON.parse(data)
rescue JSON::ParserError
  {}
end

#process!(mappings: nil, schema_fields: nil) ⇒ Object 



97
98
99
100
101
102
103
104
105
106
 
# File 'app/models/spree/import_row.rb', line 97

def process!(mappings: nil, schema_fields: nil)
  start_processing!
  self.item = import.row_processor_class.new(self, mappings: mappings, schema_fields: schema_fields).process!
  self.validation_errors = nil # clear a stale error when a retried row succeeds
  complete!
rescue StandardError => e
  Rails.error.report(e, handled: true, context: { import_row_id: id }, source: 'spree.core')
  self.validation_errors = e.message
  fail!
end

#publish_import_row_completed_event ⇒ Object 



120
121
122
 
# File 'app/models/spree/import_row.rb', line 120

def publish_import_row_completed_event
  publish_event('import_row.completed')
end

#publish_import_row_failed_event ⇒ Object 



124
125
126
 
# File 'app/models/spree/import_row.rb', line 124

def publish_import_row_failed_event
  publish_event('import_row.failed')
end

#to_schema_hash ⇒ Object 



75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'app/models/spree/import_row.rb', line 75

def to_schema_hash
  @to_schema_hash ||= begin
    mappings = import.mappings.mapped
    schema_fields = import.schema_fields

    attributes = {}
    schema_fields.each do |field|
      attributes[field[:name]] = attribute_by_schema_field(field[:name], mappings)
    end
    attributes
  end
end