defmodule BulkUpsert do @moduledoc "Bulk upsert a list of Ecto structs and their nested associations in one call." require Logger @default_timeout 15_000 # Cap the number of skipped-item IDs included in the summary warning's metadata @skipped_item_ids_log_limit 50 @valid_options [ :changeset_function_atom, :chunk_size, :insert_all_function_atom, :insert_all_function_module, :insert_all_opts, :placeholders, :recover_changeset_errors, :replace_all_except, :timeout ] # `:timeout` and `:placeholders` are also valid `insert_all/3` options, so they may appear # inside `:insert_all_opts` values as well @options_misplaced_inside_insert_all_opts @valid_options -- [:timeout, :placeholders] @typedoc """ Options accepted by `bulk_upsert/4`. See that function's documentation for details. Map keys typed `module() | Ecto.Schema.source()` accept a schema module or, for `many_to_many` join tables, the source as a string (e.g. `"persons_topics"`). """ @type options :: [ changeset_function_atom: atom(), chunk_size: pos_integer(), insert_all_function_module: module(), insert_all_function_atom: atom(), insert_all_opts: %{optional(module() | Ecto.Schema.source()) => Keyword.t()}, placeholders: %{ optional(module() | Ecto.Schema.source()) => %{optional(atom()) => term()} }, recover_changeset_errors: %{optional(module()) => %{optional(atom()) => term()}}, replace_all_except: [atom()], timeout: timeout() ] @doc """ 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. (Skipped items are summarized in one `:warning` log per call, with per-item detail at the `:debug` level.) A database error rolls back the entire upsert and raises. ## Options Unknown option names raise an `ArgumentError`, as does passing a BulkUpsert option (other than `:timeout` or `:placeholders`, which `insert_all/3` also accepts) inside an `:insert_all_opts` value. > #### Warning {: .warning} > > The `:changeset_function_atom`, `:insert_all_function_module`, and `:insert_all_function_atom` > options are invoked via `apply/3`. Never build these option values from untrusted (e.g. > user-supplied) input. - `: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: `#{@default_timeout}`) - 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. """ @spec bulk_upsert(module(), module(), [map()], options()) :: {:ok, %{upserted: non_neg_integer(), skipped: non_neg_integer()}} def bulk_upsert(repo_module, schema_module, attrs_list, opts \\ []) do validate_opts!(opts) # Parse all options once; `config` is threaded through every helper below config = %{ changeset_function_atom: Keyword.get(opts, :changeset_function_atom, :changeset), chunk_size: Keyword.get(opts, :chunk_size, 1000), recover_changeset_errors: Keyword.get(opts, :recover_changeset_errors, %{}), insert_all_function_module: Keyword.get(opts, :insert_all_function_module, repo_module), insert_all_function_atom: Keyword.get(opts, :insert_all_function_atom, :insert_all), insert_all_opts: Keyword.get(opts, :insert_all_opts, %{}), replace_all_except: Keyword.get(opts, :replace_all_except, []), placeholders: Keyword.get(opts, :placeholders, %{}), timeout: Keyword.get(opts, :timeout, @default_timeout) } attrs_list = if config.placeholders == %{} do attrs_list else # Inject each placeholder value 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 # `validate_required/2`). The value is swapped for a `{:placeholder, field}` tuple after # validation, in `insert_all_entries/4` Enum.map(attrs_list, &inject_placeholder_values(&1, schema_module, config.placeholders)) end valid_changesets = attrs_list # Convert to changesets so the data can be validated before upsertion |> Enum.map(fn attrs -> apply(schema_module, config.changeset_function_atom, [attrs]) end) |> then(&handle_invalid_changesets(schema_module, &1, config.recover_changeset_errors)) # Work around Postgres bulk limits by chunking large payloads changeset_chunks = Enum.chunk_every(valid_changesets, config.chunk_size) # Wrap all bulk upserts in a single transaction so that any failure rolls back all changes # made to every chunk of parents and all of their associations repo_module.transaction( fn -> # Use `Enum.each/2` instead of `Task.async_stream/2`. (This slightly decreases # performance, but prevents issues when using the Ecto sandbox (i.e. in the `:test` # configuration environment) since other functions may also call `Task.async_stream/2` # before calling this function. These nested async calls cause issues with the sandbox. # If the additional performance is required, the caller may be able to pass in its PID to # the `Repo.insert_all/3` opts to work around this issue, at the cost of additional # complexity in the codebase) Enum.each(changeset_chunks, &do_bulk_upsert(schema_module, &1, config)) end, timeout: config.timeout ) upserted_count = length(valid_changesets) {:ok, %{upserted: upserted_count, skipped: length(attrs_list) - upserted_count}} end # Raise on unknown option names, and on BulkUpsert-level options nested inside # `:insert_all_opts` values. The keys of each `:insert_all_opts` value are otherwise not # checked, since the set of valid `insert_all/3` options belongs to Ecto. defp validate_opts!(opts) do unknown_options = opts |> Keyword.keys() |> Enum.uniq() |> Kernel.--(@valid_options) if unknown_options != [] do raise ArgumentError, """ unknown option(s) #{inspect(unknown_options)}. Valid options: #{inspect(@valid_options)}\ """ end insert_all_opts = Keyword.get(opts, :insert_all_opts, %{}) if not is_map(insert_all_opts) do raise ArgumentError, """ the `:insert_all_opts` option must be a map of `schema_or_source => insert_all_opts`, \ got: #{inspect(insert_all_opts)}\ """ end Enum.each(insert_all_opts, fn {schema_or_source, schema_insert_all_opts} -> misplaced_options = if Keyword.keyword?(schema_insert_all_opts), do: schema_insert_all_opts |> Keyword.keys() |> Enum.uniq() |> Enum.filter(&(&1 in @options_misplaced_inside_insert_all_opts)), else: [] if misplaced_options != [] do raise ArgumentError, """ the option(s) #{inspect(misplaced_options)} given for #{inspect(schema_or_source)} are \ BulkUpsert options, and have no effect inside `:insert_all_opts`. Pass them at the top \ level of `opts` instead.\ """ end end) end # Inject the placeholder values configured for a schema into an attrs map, recursing into the # attrs of every association that will be upserted. The shared placeholder value replaces any # per-row value for the field. defp inject_placeholder_values(attrs, schema_module, placeholders) do attrs = placeholders |> Map.get(schema_module, %{}) |> Enum.reduce(attrs, fn {field, value}, acc -> put_placeholder_attr(acc, field, value) end) # Recurse into the associations that are upserted (`has_many`, `has_one`, `many_to_many`) (get_schema_associations(schema_module, :has) ++ get_schema_associations(schema_module, :many_to_many)) |> Enum.reduce(attrs, fn association, acc -> related_schema_module = schema_module.__schema__(:association, association).related update_association_attrs( acc, association, &inject_placeholder_values(&1, related_schema_module, placeholders) ) end) end # Attrs maps may use string or atom keys, and Ecto's `cast/4` raises on maps that mix both, so # the injected key must match the keys already present. defp put_placeholder_attr(attrs, field, value) do if Enum.any?(Map.keys(attrs), &is_binary/1), do: Map.put(attrs, Atom.to_string(field), value), else: Map.put(attrs, field, value) end # Apply `fun` to each attrs map under an association key, if the association is present. # Values that are not attrs maps (or lists of them) are left untouched, so the changeset cast # reports them as errors in the usual way. defp update_association_attrs(attrs, association, fun) do key = cond do Map.has_key?(attrs, association) -> association Map.has_key?(attrs, Atom.to_string(association)) -> Atom.to_string(association) true -> nil end if is_nil(key) do attrs else Map.update!(attrs, key, fn children when is_list(children) -> Enum.map(children, fn child when is_map(child) -> fun.(child) not_a_map -> not_a_map end) child when is_map(child) -> fun.(child) not_attrs -> not_attrs end) end end defp attrs_from_changeset(changeset) do struct = Ecto.Changeset.apply_action!(changeset, :build_for_bulk_upsert) struct |> Map.from_struct() |> Map.reject(fn {k, _v} -> k not in Map.keys(changeset.changes) end) end # Upsert a list of entries into a schema or source, applying any configured placeholders and # chunking large payloads to stay within Postgres bulk limits. defp insert_all_entries(entries, schema_or_source, insert_all_opts, context) do placeholder_values = context.placeholders[schema_or_source] || %{} # Placeholder fields are set here, after changeset validation, because a `{:placeholder, key}` # tuple cannot pass through a changeset. Each placeholder value is sent to Postgres once. {entries, insert_all_opts} = if placeholder_values == %{} do {entries, insert_all_opts} else placeholder_attrs = Map.new(placeholder_values, fn {field, _value} -> {field, {:placeholder, field}} end) { Enum.map(entries, &Map.merge(&1, placeholder_attrs)), Keyword.put(insert_all_opts, :placeholders, placeholder_values) } end entries |> Enum.chunk_every(context.chunk_size) |> Enum.each(fn entries_chunk -> apply(context.insert_all_function_module, context.insert_all_function_atom, [ schema_or_source, entries_chunk, insert_all_opts ]) end) end defp do_bulk_upsert(schema_module, changesets, config) do %{insert_all_opts: insert_all_opts, replace_all_except: replace_all_except, timeout: timeout} = config # Perform bulk upsert for all parent attrs attrs_list = changesets # Drop all assoc data from the changeset (assocs are handled separately in a later step) |> Enum.map(&drop_association_changes(&1, schema_module)) |> Enum.map(&attrs_from_changeset/1) # Build `insert_all` opts for the parent schema parent_insert_all_opts = Keyword.merge( _default_parent_insert_all_opts = [ conflict_target: schema_module.__schema__(:primary_key), on_conflict: {:replace_all_except, schema_module.__schema__(:primary_key) ++ replace_all_except}, timeout: timeout ], insert_all_opts[schema_module] || [] ) insert_all_entries(attrs_list, schema_module, parent_insert_all_opts, config) # Perform bulk upsert for all `has_many` and `has_one` associations for association <- get_schema_associations(schema_module, :has) do association_schema_module = schema_module.__changeset__()[association] |> elem(1) |> Map.fetch!(:related) association_changesets = changesets |> Enum.map(& &1.changes) |> Enum.map(&Map.get(&1, association)) # `has_many` changes are a list of changesets; `has_one` changes are a single # changeset; an absent association is `nil`. `List.wrap/1` normalizes each into a # (possibly empty) list, which `flat_map` concatenates. |> Enum.flat_map(&List.wrap/1) # Recurse so each child's own nested associations are upserted as well do_bulk_upsert(association_schema_module, association_changesets, config) end # Perform bulk upsert for all `many_to_many` associations for association <- get_schema_associations(schema_module, :many_to_many) do %Ecto.Association.ManyToMany{ related: related_schema_module, join_through: join_through, join_keys: [{owner_join_key, owner_key}, {related_join_key, related_key}] } = schema_module.__changeset__()[association] |> elem(1) # Pair each parent's primary key with each of its related changesets, so the related # records and the join rows can both be derived from the same data. parent_related_pairs = Enum.flat_map(changesets, fn parent_changeset -> parent_changeset.changes |> Map.get(association) |> List.wrap() |> Enum.map(fn related_changeset -> {parent_changeset, related_changeset} end) end) # Upsert the related records into their own table. The same record may be referenced by # multiple parents, so duplicates are removed to avoid conflicting twice in one query. related_changesets = parent_related_pairs |> Enum.map(fn {_parent_changeset, related_changeset} -> related_changeset end) |> Enum.uniq_by(fn related_changeset -> related_schema_module.__schema__(:primary_key) |> Enum.map(&Ecto.Changeset.get_field(related_changeset, &1)) end) # Recurse so each related record's own nested associations are upserted as well do_bulk_upsert(related_schema_module, related_changesets, config) # Upsert the join table rows that link each parent to its related records. The same link # may be listed more than once, so duplicate rows are removed for the same reason. join_attrs_list = parent_related_pairs |> Enum.map(fn {parent_changeset, related_changeset} -> %{ owner_join_key => Ecto.Changeset.fetch_field!(parent_changeset, owner_key), related_join_key => Ecto.Changeset.fetch_field!(related_changeset, related_key) } end) |> Enum.uniq() join_insert_all_opts = Keyword.merge( [ on_conflict: :nothing, conflict_target: [owner_join_key, related_join_key], timeout: timeout ], insert_all_opts[join_through] || [] ) insert_all_entries( join_attrs_list, join_through, join_insert_all_opts, config ) end end # Association changes cannot pass through `insert_all/3`, so they are dropped from the row's # own upsert. (They are upserted separately, by recursing into each association's changesets.) defp drop_association_changes(%Ecto.Changeset{} = changeset, schema_module) do Map.update!(changeset, :changes, &Map.drop(&1, schema_module.__schema__(:associations))) end # Get all `has_many` and `has_one` associations for a given schema. defp get_schema_associations(schema_module, :has) do schema_module.__changeset__() |> Enum.filter(fn {_k, v} -> match?({:assoc, %Ecto.Association.Has{}}, v) end) |> Keyword.keys() end # Get all `many_to_many` associations for a given schema. defp get_schema_associations(schema_module, :many_to_many) do schema_module.__changeset__() |> Enum.filter(fn {_k, v} -> match?({:assoc, %Ecto.Association.ManyToMany{}}, v) end) |> Keyword.keys() end defp handle_invalid_changesets(schema_module, changesets, recover_changeset_errors) do changesets |> recover_changesets_with_recoverable_errors(recover_changeset_errors) |> then(&reject_invalid_changesets(schema_module, &1)) end defp log_on_bulk_upsert_changeset_error(schema_module, changeset) do item_id_or_ids = schema_module.__schema__(:primary_key) |> Enum.reduce(%{}, fn primary_key_field, acc -> acc |> Map.put(primary_key_field, changeset.changes[primary_key_field]) end) invalid_parent_attrs = changeset.errors |> Enum.reduce(%{}, fn {k, _v}, acc -> Map.put(acc, k, changeset.changes[k]) end) # If a parent has an error in an association, the error will appear as a changeset, which # clutters up the logs. So, remove association errors from the invalid attrs map. The error # message for the field will still appear in the logs, so the information about the error # will still get passed along |> Map.new(fn {k, v} -> if k in schema_module.__schema__(:associations), do: {k, :changesets_hidden_to_keep_logs_shorter}, else: {k, v} end) invalid_association_attrs = schema_module |> get_schema_associations(:has) # Only check associations that are present in the changeset's changes (i.e. they aren't nil) |> Enum.reject(&is_nil(changeset.changes[&1])) |> Enum.reduce(%{}, fn association, acc -> association_error_items = changeset.changes[association] # `has_one` changes are a single changeset; `has_many` changes are a list of changesets. |> List.wrap() |> Enum.reject(fn changeset -> Enum.empty?(changeset.errors) end) |> Enum.reduce([], fn changeset, acc -> changeset_error_items = changeset.errors |> Keyword.keys() |> Enum.reduce([], fn key, acc -> acc |> Keyword.put(key, changeset.changes[key]) end) changeset_error_items ++ acc end) if Enum.empty?(association_error_items), do: acc, else: acc |> Map.put(association, association_error_items) end) invalid_attrs = Map.merge(invalid_parent_attrs, invalid_association_attrs) Logger.debug( """ This changeset has one or more unrecoverable errors. The item associated with this \ changeset will not be upserted.\ """, reason: :bulk_upsert_changeset_error, schema_module: inspect(schema_module), item_id_or_ids: item_id_or_ids, # NOTE: If one item in an array contains an invalid value, the whole array will be logged fields_with_invalid_attrs: Map.keys(invalid_attrs), changeset_errors: changeset.errors ) end defp recover_changesets_with_recoverable_errors(changesets, recover_changeset_errors) when changesets == [] or recover_changeset_errors == %{} do changesets end defp recover_changesets_with_recoverable_errors(changesets, recover_changeset_errors) do Enum.map(changesets, &recover_changeset(&1, recover_changeset_errors)) end # Recover a single changeset, recursing into its nested association changesets first # (bottom-up). An association's error on the parent is cleared once all of that association's # child changesets are valid. The changeset itself is then recovered only if every remaining # error field has a fallback configured for the changeset's schema. defp recover_changeset(%Ecto.Changeset{valid?: true} = changeset, _recover_changeset_errors) do changeset end defp recover_changeset(changeset, recover_changeset_errors) do schema_module = changeset.data.__struct__ # Recover the nested association and embed changesets before the changeset's own errors, # since a nested error can only be cleared once all of its child changesets are valid changeset = (schema_module.__schema__(:associations) ++ schema_module.__schema__(:embeds)) |> Enum.reduce(changeset, fn association, acc_changeset -> recover_nested_changesets(acc_changeset, association, recover_changeset_errors) end) fallbacks = Map.get(recover_changeset_errors, schema_module, %{}) error_fields = changeset.errors |> Keyword.keys() |> Enum.uniq() # An error on an association or embed field itself (e.g. attrs that could not be cast) is # never recoverable: a fallback value would replace the field's changesets in the changes # with a bare value nested_fields = schema_module.__schema__(:associations) ++ schema_module.__schema__(:embeds) recoverable_field? = &(Map.has_key?(fallbacks, &1) and &1 not in nested_fields) cond do # An invalid child does not always leave an error entry on its parent (`cast_assoc/3` may # only set `valid?: false`), so the children's own validity is checked directly. A # changeset with an unrecovered child cannot be recovered not nested_changesets_valid?(changeset, schema_module) -> changeset error_fields == [] -> # Every error was an association error, and all child changesets have been recovered %{changeset | valid?: true} Enum.all?(error_fields, recoverable_field?) -> error_fields |> Enum.reduce(changeset, &recover_changeset_field(&2, &1, Map.fetch!(fallbacks, &1))) # Clear the changeset's errors and mark the changeset as valid |> Map.merge(%{errors: [], valid?: true}) true -> # The changeset has errors with no configured fallback. The changeset (or its parent, for # a nested changeset) will be removed later in the pipeline changeset end end # Check that every changeset in the association and embed changes is valid. An invalid child # does not always leave an error entry on its parent, so the parent's error list alone cannot # prove that all children have been recovered. defp nested_changesets_valid?(changeset, schema_module) do (schema_module.__schema__(:associations) ++ schema_module.__schema__(:embeds)) |> Enum.all?(fn association -> changeset.changes |> Map.get(association) |> List.wrap() |> Enum.all?(fn %Ecto.Changeset{} = child_changeset -> child_changeset.valid? _not_a_changeset -> true end) end) end # Recover the changesets in one association's (or embed's) changes, clearing the association's # error on the parent once every child changeset is valid. defp recover_nested_changesets(changeset, association, recover_changeset_errors) do case changeset.changes[association] do nil -> changeset children -> # `has_many`, `many_to_many`, and `embeds_many` changes are a list of changesets; # `has_one` and `embeds_one` changes are a single changeset. `List.wrap/1` normalizes # both into a list for recovery, and the original shape is restored when the changes are # updated recovered = children |> List.wrap() |> Enum.map(&recover_changeset(&1, recover_changeset_errors)) recovered_children = if is_list(children), do: recovered, else: hd(recovered) changeset = %{ changeset | changes: Map.put(changeset.changes, association, recovered_children) } if Enum.all?(recovered, & &1.valid?), do: Map.update!(changeset, :errors, &Keyword.delete(&1, association)), else: changeset end end defp recover_changeset_field(changeset, field, recover_to_value) do primary_key_info = changeset.data.__struct__.__schema__(:primary_key) |> Keyword.new(fn primary_key_field -> # The primary key may be absent from the changes (e.g. if the changeset function does # not require it), so avoid `Map.fetch!/2` {primary_key_field, Map.get(changeset.changes, primary_key_field)} end) Logger.debug(""" Recovered changeset error for struct #{Macro.to_string(changeset.data.__struct__)} with \ primary key(s) `#{inspect(primary_key_info)}` in the field `#{field}`.\ """) %{changeset | changes: Map.put(changeset.changes, field, recover_to_value)} end defp reject_invalid_changesets(schema_module, changesets) do {valid_changesets, invalid_changesets} = Enum.split_with(changesets, & &1.valid?) if invalid_changesets != [] do log_skipped_changesets_summary(schema_module, changesets, invalid_changesets) Enum.each(invalid_changesets, &log_on_bulk_upsert_changeset_error(schema_module, &1)) end valid_changesets end # One `:warning` per bulk upsert call summarizes every skipped item. The per-item details are # logged at the `:debug` level, so a large batch of invalid rows cannot flood the log. defp log_skipped_changesets_summary(schema_module, changesets, invalid_changesets) do skipped_count = length(invalid_changesets) item_ids = invalid_changesets |> Enum.take(@skipped_item_ids_log_limit) |> Enum.map(fn changeset -> schema_module.__schema__(:primary_key) |> Map.new(fn primary_key_field -> {primary_key_field, changeset.changes[primary_key_field]} end) end) truncation_note = if skipped_count > @skipped_item_ids_log_limit, do: " The first #{@skipped_item_ids_log_limit} skipped item IDs are listed.", else: "" Logger.warning( """ Skipped #{skipped_count} of #{length(changesets)} items because their changesets had \ unrecoverable errors. The skipped items were not upserted. Details for each skipped item \ are logged at the `:debug` level.#{truncation_note}\ """, reason: :bulk_upsert_items_skipped, schema_module: inspect(schema_module), skipped_count: skipped_count, item_ids: item_ids ) end end