class ActiveRecord::Base
def self.synchronize(instances, keys = [primary_key])
posts.first.address # => "1245 Foo Ln" instead of whatever it was
Post.synchronize posts, [:name] # queries on the :name column and not the :id column
<.. out of system changes occur to change the address of author 'Zach' to 1245 Foo Ln ..>
posts = Post.where(author: "Zach").first
# Synchronizing using custom key fields
posts.first.author # => "Zachary" instead of Zach
Post.synchronize posts
<.. out of system changes occur to change author name from Zach to Zachary..>
posts = Post.where(author: "Zach").first
# Synchronizing existing models by matching on the primary key field
== Examples
instances rather sending one query for each instance
This uses one query for all instance updates and then updates existing
ActiveRecord instance but it is intended for use on multiple instances.
from the database. This is like calling reload on an individual
Synchronizes the passed in ActiveRecord instances with data
:nodoc:
:nodoc:
def self.synchronize(instances, keys = [primary_key]) return if instances.empty? conditions = {} key_values = keys.map { |key| instances.map(&key.to_sym) } keys.zip(key_values).each { |key, values| conditions[key] = values } order = keys.map { |key| "#{key} ASC" }.join(",") klass = instances.first.class fresh_instances = klass.where(conditions).order(order) instances.each do |instance| matched_instance = fresh_instances.detect do |fresh_instance| keys.all? { |key| fresh_instance.send(key) == instance.send(key) } end next unless matched_instance instance.send :clear_aggregation_cache instance.send :clear_association_cache instance.instance_variable_set :@attributes, matched_instance.instance_variable_get(:@attributes) if instance.respond_to?(:clear_changes_information) instance.clear_changes_information # Rails 4.2 and higher else instance.instance_variable_set :@attributes_cache, {} # Rails 4.0, 4.1 instance.changed_attributes.clear # Rails 3.2 instance.previous_changes.clear end # Since the instance now accurately reflects the record in # the database, ensure that instance.persisted? is true. instance.instance_variable_set '@new_record', false instance.instance_variable_set '@destroyed', false end end
def add_special_rails_stamps( column_names, array_of_attributes, options )
def add_special_rails_stamps( column_names, array_of_attributes, options ) AREXT_RAILS_COLUMNS[:create].each_pair do |key, blk| next unless self.column_names.include?(key) value = blk.call index = column_names.index(key) || column_names.index(key.to_sym) if index # replace every instance of the array of attributes with our value array_of_attributes.each { |arr| arr[index] = value if arr[index].nil? } else column_names << key array_of_attributes.each { |arr| arr << value } end end AREXT_RAILS_COLUMNS[:update].each_pair do |key, blk| next unless self.column_names.include?(key) value = blk.call index = column_names.index(key) || column_names.index(key.to_sym) if index # replace every instance of the array of attributes with our value array_of_attributes.each { |arr| arr[index] = value } else column_names << key array_of_attributes.each { |arr| arr << value } end if supports_on_duplicate_key_update? connection.add_column_for_on_duplicate_key_update(key, options) end end end
def establish_connection_with_activerecord_import(*args)
def establish_connection_with_activerecord_import(*args) conn = establish_connection_without_activerecord_import(*args) if !ActiveRecord.const_defined?(:Import) || !ActiveRecord::Import.respond_to?(:load_from_connection_pool) require "activerecord-import/base" end ActiveRecord::Import.load_from_connection_pool connection_pool conn end
def find_associated_objects_for_import(associated_objects_by_class, model)
We are eventually going to call Class.import
def find_associated_objects_for_import(associated_objects_by_class, model) associated_objects_by_class[model.class.name] ||= {} association_reflections = model.class.reflect_on_all_associations(:has_one) + model.class.reflect_on_all_associations(:has_many) association_reflections.each do |association_reflection| associated_objects_by_class[model.class.name][association_reflection.name] ||= [] association = model.association(association_reflection.name) association.loaded! # Wrap target in an array if not already association = Array(association.target) changed_objects = association.select { |a| a.new_record? || a.changed? } changed_objects.each do |child| child.public_send("#{association_reflection.foreign_key}=", model.id) # For polymorphic associations association_reflection.type.try do |type| child.public_send("#{type}=", model.class.name) end end associated_objects_by_class[model.class.name][association_reflection.name].concat changed_objects end associated_objects_by_class end
def import(*args)
* num_inserts - the number of insert statements it took to import the data
* failed_instances - an array of objects that fails validation and were not committed to the database. An empty array if no validation is performed.
This returns an object which responds to +failed_instances+ and +num_inserts+.
= Returns
BlogPost.import columns, attributes, on_duplicate_key_update: { conflict_target: :slug, columns: { title: :title } }
with what attributes on your model. Below is an example:
mappings. This gives you finer grained control over what fields are updated
The :columns option can be a hash of column names to model attribute name
======== Using a Hash
BlogPost.import columns, values, on_duplicate_key_update: { conflict_target: :slug, columns: [ :date_modified, :content, :author ] }
Below is an example:
are the only fields that are updated if a duplicate record is found.
The :columns attribute can be an array of column names. The column names
======== Using an Array
The :columns attribute can be either an Array or a Hash.
====== :columns
BlogPost.import columns, values, on_duplicate_key_update: { constraint_name: :blog_posts_pkey, columns: [ :date_modified ] }
of identifying an index unless absolutely necessary. Below is an example:
unique index by name. Postgres documentation discourages using this method
The :constraint_name attribute explicitly identifies the conflicting
====== :constraint_name
BlogPost.import columns, values, on_duplicate_key_update: { conflict_target: [:author_id, :slug], columns: [ :date_modified ] }
default to the primary key. Below is an example:
but it is the preferred method of identifying a constraint. It will
column names. This attribute is ignored if :constraint_name is included,
conflicting unique constraint and can be a single column or an array of
The :conflict_target attribute specifies the columns that make up the
====== :conflict_target
option allows you to specify a constraint other than the primary key.
requires the conflicting constraint to be explicitly specified. Using this
:conflict_target or constraint_name, and :columns. Unlike MySQL, Postgres
The :on_duplicate_update option can be a hash with up to two attributes,
==== Using a Hash
BlogPost.import columns, values, on_duplicate_key_update: [ :date_modified, :content, :author ]
if a duplicate record is found. Below is an example:
not work. The column names are the only fields that are updated
primary key. If a table does not have a primary key, this will
names. This option only handles inserts that conflict with the
The :on_duplicate_key_update option can be an array of column
==== Using an Array
two attributes, :conflict_target or :constraint_name and :columns.
The :on_duplicate_key_update option can be an Array or a Hash with up to
== On Duplicate Key Update (Postgres 9.5+)
BlogPost.import columns, attributes, on_duplicate_key_update: { title: :title }
model. Below is an example:
control over what fields are updated with what attributes on your
to model attribute name mappings. This gives you finer grained
The :on_duplicate_key_update option can be a hash of column names
==== Using A Hash
BlogPost.import columns, values, on_duplicate_key_update: [ :date_modified, :content, :author ]
a duplicate record is found. Below is an example:
names. The column names are the only fields that are updated if
The :on_duplicate_key_update option can be an array of column
==== Using an Array
The :on_duplicate_key_update option can be either an Array or a Hash.
== On Duplicate Key Update (MySQL)
puts posts.first.persisted? # => true
BlogPost.import posts, synchronize: posts, synchronize_keys: [:title]
posts = [BlogPost.new(title: "Foo"), BlogPost.new(title: "Bar")]
# Example synchronizing unsaved/new instances in memory by using a uniqued imported field
puts post.author_name # => 'yoda'
BlogPost.import posts, synchronize: [ post ]
values = [ [ 'yoda', 'test post' ] ]
columns = [ :author_name, :title ]
puts post.author_name # => 'zdennis'
post = BlogPost.where(author_name: 'zdennis').first
# Example synchronizing existing instances in memory
BlogPost.import( columns, values, validate: false )
values = [ [ 'zdennis', 'test post' ], [ 'jdoe', 'another test post' ] ]
columns = [ :author_name, :title ]
# Example using column_names, array_of_value and options
BlogPost.import columns, values
values = [ [ 'zdennis', 'test post' ], [ 'jdoe', 'another test post' ] ]
columns = [ :author_name, :title ]
# Example using column_names and array_of_values
BlogPost.import posts
BlogPost.new author_name: 'Zach Dennis', title: 'AREXT3' ]
BlogPost.new author_name: 'Zach Dennis', title: 'AREXT2',
posts = [ BlogPost.new author_name: 'Zach Dennis', title: 'AREXT',
# Example using array of model objects
class BlogPost < ActiveRecord::Base ; end
== Examples
include per insert. Defaults to the total number of records to import.
* +batch_size+ - an integer value to specify the max number of records to
newly imported objects.
associations if the adapter supports setting the primary keys of the
* +recursive+ - true|false, tells import to import all has_many/has_one
(if false) even if record timestamps is disabled in ActiveRecord::Base
* +timestamps+ - true|false, tells import to not add timestamps
existing model instances in memory with updates from the import.
that you are currently importing data into. This synchronizes
* +synchronize+ - an array of ActiveRecord instances for the model
DO UPDATE ability. See On Duplicate Key Update below.
use MySQL's ON DUPLICATE KEY UPDATE or Postgres 9.5+ ON CONFLICT
* +on_duplicate_key_update+ - an Array or Hash, tells import to
recursive import.
Postgres 9.5+ ON CONFLICT DO NOTHING. Cannot be enabled on a
* +on_duplicate_key_ignore+ - true|false, tells import to use
to discard records that contain duplicate keys.
* +ignore+ - true|false, tells import to use MySQL's INSERT IGNORE
ActiveRecord validations. Validations are enforced by default.
* +validate+ - true|false, tells import whether or not to use
== Options
below for what +options+ are available.
parameter, +options+, is a hash. This is optional. Please see
The first two parameters are the same as the above form. The third
==== Model.import column_names, array_of_values, options
the order of the +column_names+.
record. The order of values in each subarray should match up to
arrays. Each subarray is a single set of values for a new
The second parameter, +array_of_values+, is an array of
strings which specify the columns that you want to update.
The first parameter +column_names+ is an array of symbols or
==== Model.import column_names, array_of_values
objects that you want updated.
With this form you can call _import_ passing in an array of model
==== Model.import array_of_models
Model.import column_names, array_of_values, options
Model.import column_names, array_of_values
Model.import array_of_models
== Usage
performing the import.
the ActiveRecord::Callbacks during creation/modification while
This can be used with or without validations. It does not utilize
inserted.
about having ActiveRecord objects returned for each record
you want to create more than one record at a time and do not care
ActiveRecord::Base#save multiple times. This method works well if
This is more efficient than using ActiveRecord::Base#create or
Imports a collection of values to the database.
def import(*args) if args.first.is_a?( Array ) && args.first.first.is_a?(ActiveRecord::Base) options = {} options.merge!( args.pop ) if args.last.is_a?(Hash) models = args.first import_helper(models, options) else import_helper(*args) end end
def import!(*args)
first encountered validation error and raises ActiveRecord::RecordInvalid
Imports a collection of values if all values are valid. Import fails at the
def import!(*args) options = args.last.is_a?( Hash ) ? args.pop : {} options[:validate] = true options[:raise_error] = true import(*args, options) end
def import_associations(models, options)
def import_associations(models, options) # now, for all the dirty associations, collect them into a new set of models, then recurse. # notes: # does not handle associations that reference themselves # should probably take a hash to associations to follow. associated_objects_by_class = {} models.each { |model| find_associated_objects_for_import(associated_objects_by_class, model) } # :on_duplicate_key_update not supported for associations options.delete(:on_duplicate_key_update) associated_objects_by_class.each_value do |associations| associations.each_value do |associated_records| associated_records.first.class.import(associated_records, options) unless associated_records.empty? end end end
def import_from_table( options ) # :nodoc:
TODO import_from_table needs to be implemented.
def import_from_table( options ) # :nodoc: end
def import_helper( *args )
def import_helper( *args ) options = { validate: true, timestamps: true, primary_key: primary_key } options.merge!( args.pop ) if args.last.is_a? Hash # Don't modify incoming arguments if options[:on_duplicate_key_update] options[:on_duplicate_key_update] = options[:on_duplicate_key_update].dup end is_validating = options[:validate] is_validating = true unless options[:validate_with_context].nil? # assume array of model objects if args.last.is_a?( Array ) && args.last.first.is_a?(ActiveRecord::Base) if args.length == 2 models = args.last column_names = args.first else models = args.first column_names = self.column_names.dup end array_of_attributes = models.map do |model| # this next line breaks sqlite.so with a segmentation fault # if model.new_record? || options[:on_duplicate_key_update] column_names.map do |name| model.read_attribute_before_type_cast(name.to_s) end # end end # supports empty array elsif args.last.is_a?( Array ) && args.last.empty? return ActiveRecord::Import::Result.new([], 0, []) # supports 2-element array and array elsif args.size == 2 && args.first.is_a?( Array ) && args.last.is_a?( Array ) column_names, array_of_attributes = args array_of_attributes = array_of_attributes.map(&:dup) else raise ArgumentError, "Invalid arguments!" end # dup the passed in array so we don't modify it unintentionally column_names = column_names.dup # Force the primary key col into the insert if it's not # on the list and we are using a sequence and stuff a nil # value for it into each row so the sequencer will fire later if !column_names.include?(primary_key) && connection.prefetch_primary_key? && sequence_name column_names << primary_key array_of_attributes.each { |a| a << nil } end # record timestamps unless disabled in ActiveRecord::Base if record_timestamps && options.delete( :timestamps ) add_special_rails_stamps column_names, array_of_attributes, options end return_obj = if is_validating if models import_with_validations( column_names, array_of_attributes, options ) do |failed| models.each_with_index do |model, i| model = model.dup if options[:recursive] next if model.valid?(options[:validate_with_context]) if options[:raise_error] && model.respond_to?(:raise_validation_error, true) # Rails 5.0 and higher model.send(:raise_validation_error) elsif options[:raise_error] # Rails 3.2, 4.0, 4.1 and 4.2 model.send(:raise_record_invalid) end array_of_attributes[i] = nil failed << model end end else import_with_validations( column_names, array_of_attributes, options ) end else (num_inserts, ids) = import_without_validations_or_callbacks( column_names, array_of_attributes, options ) ActiveRecord::Import::Result.new([], num_inserts, ids) end if options[:synchronize] sync_keys = options[:synchronize_keys] || [primary_key] synchronize( options[:synchronize], sync_keys) end return_obj.num_inserts = 0 if return_obj.num_inserts.nil? # if we have ids, then set the id on the models and mark the models as clean. if support_setting_primary_key_of_imported_objects? set_ids_and_mark_clean(models, return_obj) # if there are auto-save associations on the models we imported that are new, import them as well import_associations(models, options.dup) if options[:recursive] end return_obj end
def import_with_validations( column_names, array_of_attributes, options = {} )
ActiveRecord::Base.import for more information on
+num_inserts+ is the number of inserts it took to import the data. See
+failed_instances+ is an array of instances that failed validations.
object with the methods +failed_instances+ and +num_inserts+.
given the passed in +options+ Hash with validations. Returns an
Imports the passed in +column_names+ and +array_of_attributes+
def import_with_validations( column_names, array_of_attributes, options = {} ) failed_instances = [] if block_given? yield failed_instances else # create instances for each of our column/value sets arr = validations_array_for_column_names_and_attributes( column_names, array_of_attributes ) # keep track of the instance and the position it is currently at. if this fails # validation we'll use the index to remove it from the array_of_attributes model = new arr.each_with_index do |hsh, i| hsh.each_pair { |k, v| model[k] = v } next if model.valid?(options[:validate_with_context]) raise(ActiveRecord::RecordInvalid, model) if options[:raise_error] array_of_attributes[i] = nil failed_instances << model.dup end end array_of_attributes.compact! num_inserts, ids = if array_of_attributes.empty? || options[:all_or_none] && failed_instances.any? [0, []] else import_without_validations_or_callbacks( column_names, array_of_attributes, options ) end ActiveRecord::Import::Result.new(failed_instances, num_inserts, ids) end
def import_without_validations_or_callbacks( column_names, array_of_attributes, options = {} )
information on +column_names+, +array_of_attributes_ and
validations or callbacks. See ActiveRecord::Base.import for more
of insert operations it took to create these records without
given the passed in +options+ Hash. This will return the number
Imports the passed in +column_names+ and +array_of_attributes+
def import_without_validations_or_callbacks( column_names, array_of_attributes, options = {} ) column_names = column_names.map(&:to_sym) scope_columns, scope_values = scope_attributes.to_a.transpose unless scope_columns.blank? scope_columns.zip(scope_values).each do |name, value| name_as_sym = name.to_sym next if column_names.include?(name_as_sym) is_sti = (name_as_sym == inheritance_column.to_sym && self < base_class) value = value.first if is_sti column_names << name_as_sym array_of_attributes.each { |attrs| attrs << value } end end columns = column_names.each_with_index.map do |name, i| column = columns_hash[name.to_s] raise ActiveRecord::Import::MissingColumnError.new(name.to_s, i) if column.nil? column end columns_sql = "(#{column_names.map { |name| connection.quote_column_name(name) }.join(',')})" insert_sql = "INSERT #{options[:ignore] ? 'IGNORE ' : ''}INTO #{quoted_table_name} #{columns_sql} VALUES " values_sql = values_sql_for_columns_and_attributes(columns, array_of_attributes) number_inserted = 0 ids = [] if supports_import? # generate the sql post_sql_statements = connection.post_sql_statements( quoted_table_name, options ) batch_size = options[:batch_size] || values_sql.size values_sql.each_slice(batch_size) do |batch_values| # perform the inserts result = connection.insert_many( [insert_sql, post_sql_statements].flatten, batch_values, "#{self.class.name} Create Many Without Validations Or Callbacks" ) number_inserted += result[0] ids += result[1] end else values_sql.each do |values| ids << connection.insert(insert_sql + values) number_inserted += 1 end end [number_inserted, ids] end
def set_ids_and_mark_clean(models, import_result)
def set_ids_and_mark_clean(models, import_result) return if models.nil? import_result.ids.each_with_index do |id, index| model = models[index] model.id = id.to_i if model.respond_to?(:clear_changes_information) # Rails 4.0 and higher model.clear_changes_information else # Rails 3.2 model.instance_variable_get(:@changed_attributes).clear end model.instance_variable_set(:@new_record, false) end end
def support_setting_primary_key_of_imported_objects?
supports setting the primary key of bulk imported models, otherwise
returns true if the current database connection adapter
def support_setting_primary_key_of_imported_objects? connection.respond_to?(:support_setting_primary_key_of_imported_objects?) && connection.support_setting_primary_key_of_imported_objects? end
def supports_import?(*args)
Returns true if the current database connection adapter
def supports_import?(*args) connection.respond_to?(:supports_import?) && connection.supports_import?(*args) end
def supports_on_duplicate_key_update?
supports on duplicate key update functionality, otherwise
Returns true if the current database connection adapter
def supports_on_duplicate_key_update? connection.supports_on_duplicate_key_update? end
def synchronize(instances, key = [ActiveRecord::Base.primary_key])
def synchronize(instances, key = [ActiveRecord::Base.primary_key]) self.class.synchronize(instances, key) end
def validations_array_for_column_names_and_attributes( column_names, array_of_attributes ) # :nodoc:
Returns an Array of Hashes for the passed in +column_names+ and +array_of_attributes+.
def validations_array_for_column_names_and_attributes( column_names, array_of_attributes ) # :nodoc: array_of_attributes.map { |values| Hash[column_names.zip(values)] } end
def values_sql_for_columns_and_attributes(columns, array_of_attributes) # :nodoc:
and +array_of_attributes+.
Returns SQL the VALUES for an INSERT statement given the passed in +columns+
def values_sql_for_columns_and_attributes(columns, array_of_attributes) # :nodoc: # connection gets called a *lot* in this high intensity loop. # Reuse the same one w/in the loop, otherwise it would keep being re-retreived (= lots of time for large imports) connection_memo = connection array_of_attributes.map do |arr| my_values = arr.each_with_index.map do |val, j| column = columns[j] # be sure to query sequence_name *last*, only if cheaper tests fail, because it's costly if val.nil? && column.name == primary_key && !sequence_name.blank? connection_memo.next_value_for_sequence(sequence_name) elsif column if respond_to?(:type_caster) && type_caster.respond_to?(:type_cast_for_database) # Rails 5.0 and higher connection_memo.quote(type_caster.type_cast_for_database(column.name, val)) elsif column.respond_to?(:type_cast_from_user) # Rails 4.2 and higher connection_memo.quote(column.type_cast_from_user(val), column) else # Rails 3.2, 4.0 and 4.1 if serialized_attributes.include?(column.name) val = serialized_attributes[column.name].dump(val) end connection_memo.quote(column.type_cast(val), column) end end end "(#{my_values.join(',')})" end end