@spec bulk_upsert(module(), module(), [map()], options()) ::
  {:ok, %{upserted: non_neg_integer(), skipped: non_neg_integer()}}

Validate a list of attrs maps (attrs_list) by passing them through an Ecto changeset, then upsert the valid items to the database that corresponds to a given Ecto repo_module (e.g. YourProject.Repo).

Using a changeset serves two purposes:

1. The changeset can be used to validate and transform the data.
2. Using a changeset allows this function to perform bulk upserts with nested associations.

For validation, each list item in the attrs_list is converted to a changeset for a given schema_module. The changeset function is called with a single argument (the attrs map), so the schema module must expose a 1-arity changeset function — e.g. a changeset/2 whose first argument defaults to an empty struct. By default, this function is called :changeset. (See the Options section below for more info.)

Basic example

iex> BulkUpsert.bulk_upsert(
...>   YourProject.Repo,
...>   YourProject.Persons.Person,
...>   _attrs_list = [
...>     %{id: 1, name: "Alice", age: 25, phone_number: "555-1234"},
...>     %{id: 2, name: "Bob", age: 35, phone_number: "555-2345"},
...>   ]
...> )
{:ok, %{upserted: 2, skipped: 0}}

Return value

Returns {:ok, %{upserted: upserted_count, skipped: skipped_count}}, where the counts refer to the top-level attrs: :upserted is the number of items sent to the database, and :skipped is the number of items dropped because their changesets were invalid. (Each skipped item is also logged at the :warning level.) A database error rolls back the entire upsert and raises.

Options

• :changeset_function_atom - The name of the changeset function to apply for the given schema_module. It is called with one argument: the attrs map. (Default: :changeset)

• :chunk_size - The number of parent attrs items to insert into the database in a single query. Can be increased or decreased as needed to avoid exceeding the Postgres parameter limit for a single query. (Default: 1000)

• :insert_all_function_module - Instead of using the :insert_all function in the given repo_module, you may specify the name of a custom module to use instead. (Default: Inherited from the value specified in the repo_module function argument, e.g. YourProject.Repo)

  • Example: YourProject.OtherRepo

• :insert_all_function_atom - Instead of using your repo module's :insert_all function, you may pass a compatible equivalent that accepts the same arguments. (Default: :insert_all)

  • Example: :insert_all_with_autogenerated_timestamps

• :insert_all_opts - Pass custom opts to the insert_all/3 function. This option consists of a map whose key is the schema or source that may have items being upserted, and the value is the YourProject.Repo.insert_all/3 opts that will be applied when items for that schema are being upserted. By default, a conflicting row has all of its values replaced except the primary key(s) (see the :replace_all_except option). (Default: %{})

  • Example: %{YourProject.Persons.Person => [on_conflict: {:nothing}]}
  • A many_to_many join table is keyed by its source, e.g. %{"persons_topics" => [...]}.
  • :conflict_target defaults to the schema's primary key, so a schema without a primary key must supply its own :conflict_target here (otherwise the upsert fails at the database).

• :placeholders - Set fields from shared values that are sent to the database once instead of once per row, using the :placeholders feature of Ecto's insert_all/3. This option is a map whose key is the schema or source being upserted, and the value is a map of field => value. The fields do not need to appear in the attrs. (Default: %{})

  • Example: %{YourProject.Persons.Person => %{inserted_at: DateTime.utc_now()}}
  • Each placeholder value is injected into the attrs before the changeset is built, so a placeholder field is cast and validated like any other field and may be included in the changeset's validate_required/2.
  • The shared value replaces any per-row value supplied for the field in the attrs.
  • Embedded schemas are stored inline on their parent row and are never upserted as their own source, so placeholder values keyed by an embedded schema module are ignored.

• :recover_changeset_errors - If the given fields in a changeset have errors, then replace them with a custom fallback value. (Default: %{})

  • Example: %{YourProject.Persons.Person => %{phone_number: "INVALID"}}
  • Applies recursively to nested association and embedded changesets, with fallbacks looked up by each changeset's schema (for embeds, the embedded schema module). A parent's association error is cleared once all of that association's child changesets have been recovered.
  • A changeset is only recovered if every one of its error fields has a fallback and every nested changeset is recoverable by the same rule; otherwise the row is skipped.
  • A fallback value is applied without re-running the changeset function, so it must be valid for the schema.
  • Errors on the association and embed fields themselves (e.g. an association whose attrs could not be cast at all) are never recoverable.

• :replace_all_except - If a row already exists, then all fields will be replaced except the primary key, and any fields specified here. (Default: [])

  • Example: [:field, :other_field]

• :timeout - The maximum timeout for the transaction that wraps the entire bulk upsert (all chunks), also applied to each insert_all/3 query. (Default: 15000)

  • Example: 60_000

Examples

Upsert a list of Person attrs using the changeset function YourProject.Persons.Person.upsert_changeset/2 to validate the attrs:

iex> attrs_list = [%{id: 1, name: "Alice", ...}]

iex> BulkUpsert.bulk_upsert(
...>   YourProject.Repo,
...>   YourProject.Persons.Person,
...>   attrs_list,
...>   changeset_function_atom: :upsert_changeset
...> )
{:ok, %{upserted: 1, skipped: 0}}

Upsert a list of attrs, overwriting only the :name field if there is a conflict. Schemas that are not given custom :insert_all_opts keep the default conflict behavior (replace all fields except the primary key):

iex> insert_all_opts = %{
...>   YourProject.Persons.Person => [on_conflict: {:replace, [:name]}]
...> }

iex> BulkUpsert.bulk_upsert(
...>   YourProject.Repo,
...>   YourProject.Persons.Person,
...>   _attrs_list = [%{id: 1, name: "Alicia"}],
...>   insert_all_opts: insert_all_opts
...> )
{:ok, %{upserted: 1, skipped: 0}}

Associations

Nested associations are upserted in the same call as the parent, recursively: a child's own nested associations (at any depth) are upserted the same way as the parent's.

This is an upsert-only operation: rows absent from the attrs are left untouched at every level. Unlike Ecto.Changeset.cast_assoc/3's :on_replace behavior, absent children are never deleted or nilified.

• has_many and has_one: the associated records are upserted into their own table. Each child must include its foreign key in its attrs, since it is upserted directly via insert_all/3.

• many_to_many: the associated records are upserted into their own table, and the join table rows linking each parent to its associations are upserted as well. Duplicate records and links are removed automatically.

• embeds_one and embeds_many: embedded data has no table of its own, so it is stored inline on the parent row as part of the parent upsert.

Known limitations

• Nested belongs_to associations are not upserted. To associate with a belongs_to parent, include its foreign key field in the attrs (e.g. category_id). This applies at every level of nesting.