Solidify API for Bulk Upsert calls

Author: robcole

spec/avram/operations/save_operation_spec.cr

@@ -312,13 +312,12 @@ describe "Avram::SaveOperation" do
       it "should create the records" do
         record_args = (1..2).to_a.map do |i|
           {
-            :name     => "Test #{i}",
-            :nickname => "Test Nickname #{i}",
-          }.as(Avram::BulkUpsert::Params)
+            name:     "Test #{i}",
+            nickname: "Test Nickname #{i}",
+          }
         end
-        records = UserFactory.new.build_pair
 
-        SaveUser.bulk_upsert(record_args)
+        UpsertUserOperation.bulk_upsert(record_args)
       end
     end
 

src/avram/bulk_upsert.cr

@@ -1,57 +1,70 @@
 class Avram::BulkUpsert(T)
-  alias Params = Hash(Symbol, String) | Hash(Symbol, String?) | Hash(Symbol, Nil)
-
-  def initialize(@records : Array(Params))
+  def initialize(@records : Array(T), @column_names : Array(Symbol))
+    @records = set_timestamps(records)
   end
 
   def statement
     [
-      "insert into #{table}(#{fields})",
-      "values #{value_placeholders}",
-      "ON CONFLICT DO UPDATE SET #{updates}",
-      "returning #{returning}",
+      "INSERT INTO #{table}(#{fields})",
+      "VALUES #{value_placeholders}",
+      "ON CONFLICT (#{conflicts}) DO UPDATE SET #{updates}",
+      "RETURNING #{returning}",
     ].join(" ")
   end
 
+  private def conflicts
+    @column_names.join(", ")
+  end
+
+  private def set_timestamps(collection)
+    collection.map do |record|
+      record.created_at.value ||= Time.utc if record.responds_to?(:created_at)
+      record.updated_at.value = Time.utc if record.responds_to?(:updated_at)
+      record
+    end
+  end
+
   private def table
-    T.table_name
+    @records.first.table_name
   end
 
   private def updates
-    conflict_updates = T.column_names.uniq.map do |column|
-      "SET #{column}=EXCLUDED.#{column}"
-    end
+    update_keys = @records.first.insert_values.keys
 
-    if T.column_names.includes?(:updated_at)
-      conflict_updates.push("SET updated_at=NOW()").join(", ")
-    else
-      conflict_updates.join(", ")
-    end
+    (update_keys - [:created_at]).map do |column|
+      "#{column}=EXCLUDED.#{column}"
+    end.join(", ")
   end
 
   private def returning
-    "id"
+    T.column_names.join(", ")
   end
 
   private def fields
-    T.column_names.join(", ")
+    @records.first.insert_values.keys.map do |key|
+      <<-TEXT
+      "#{key}"
+      TEXT
+    end.join(", ")
   end
 
   def args
-    @records.map &.values
+    @records.flat_map do |record|
+      record.insert_values.values
+    end
   end
 
-  private def placeholder_values(record)
-    values = record.values.map_with_index(1) do |_value, index|
+  private def value_placeholders(record)
+    record.insert_values.map_with_index(1) do |_value, index|
       "$#{index}"
     end.join(", ")
-
-    "(#{values})"
   end
 
   private def value_placeholders
+    i = 0
+
     @records.map do |record|
-      placeholder_values(record)
+      "($#{i += 1}, $#{i += 1}, $#{i += 1}, $#{i += 1})"
     end.join(", ")
   end
 end

src/avram/save_operation.cr

@@ -367,6 +367,10 @@ abstract class Avram::SaveOperation(T)
     {{ T.constant(:PRIMARY_KEY_NAME).id }}.value.nil?
   end
 
+  def insert_values
+    attributes_to_hash(column_attributes).compact
+  end
+
   private def insert_or_update
     if persisted?
       update record_id
@@ -379,6 +383,10 @@ abstract class Avram::SaveOperation(T)
     @record.try &.id
   end
 
+  def self.column_names
+    T.column_names
+  end
+
   def before_save; end
 
   def after_save(_record : T); end
@@ -408,7 +416,6 @@ abstract class Avram::SaveOperation(T)
   end
 
   private def insert_sql
-    insert_values = attributes_to_hash(column_attributes).compact
     Avram::Insert.new(table_name, insert_values, T.column_names)
   end
 

src/avram/upsert.cr

@@ -81,6 +81,27 @@ module Avram::Upsert
       yield operation, operation.record
     end
 
+    def self.bulk_upsert(upserts)
+      operations = upserts.map do |upsert_args|
+        new(**upsert_args)
+      end
+
+      upsert = Avram::BulkUpsert(self).new(
+        operations,
+        {{ attribute_names }}.to_a
+      )
+
+      pp upsert.statement
+      pp upsert.args
+      records = [] of T
+
+      new.database.query upsert.statement, args: upsert.args do |rs|
+        records << T.from_rs(rs).first
+      end
+
+      pp records
+    end
+
     def self.find_existing_unique_record(operation) : T?
       T::BaseQuery.new
         {% for attribute in attribute_names %}
@@ -92,29 +113,13 @@ module Avram::Upsert
 
   # :nodoc:
   macro included
-    {% for method in ["upsert", "upsert!"] %}
+    {% for method in ["upsert", "upsert!", "bulk_upsert"] %}
       # Performs a create or update depending on if there is a conflicting row in the database.
       #
       # See `Avram::Upsert.upsert_lookup_columns` for full documentation and examples.
       def self.{{ method.id }}(*args, **named_args)
         \{% raise "Please use the 'upsert_lookup_columns' macro in #{@type} before using '{{ method.id }}'" %}
       end
     {% end %}
-
-    def self.bulk_upsert(upserts : Array(Avram::BulkUpsert::Params))
-      upsert_keys = upserts.map(&.keys).uniq
-
-      if upsert_keys.size > 1
-        raise "All hashes passed to bulk_upsert must have the same keys."
-      elsif upsert_keys.flatten.any? { |key| !T.column_names.includes?(key) }
-        raise "All keys in hashes must be column names in the table."
-      end
-
-      upsert = Avram::BulkUpsert(T).new(upserts)
-      pp upsert.statement
-      pp upsert.args
-
-      T.database.query(upsert.statement, args: upsert.args)
-    end
   end
 end