# BulkUpsert

> #### Note
>
> This library is pre-1.0: the API may change between minor versions (see the changelog).
> The test suite exercises it against a real Postgres database, but review the documented
> limitations before using it in production.

Upsert multiple Ecto schema structs, along with their nested associations, to the database with a single function call.

Unlike a plain `insert_all/3`, this package passes each list of attrs through Ecto changesets. This lets it validate your data and upsert a parent **and its children across multiple tables** in one call.

Supported features:
  - Nested associations: upsert a parent and its `has_many`, `has_one`, and `many_to_many` associations across multiple tables from a single list of attrs, recursively at any depth (embedded schemas are stored inline on the parent)
  - Validation and data processing (via Ecto changesets)
  - Custom values for autogenerated fields (e.g. insert/update timestamps)
  - Recovery of invalid field values via per-schema fallbacks
  - A single transaction wraps the entire upsert, so any failure rolls back all changes

For more information, see [this project's documentation](https://hexdocs.pm/bulk_upsert/BulkUpsert.html).

---

## Getting started

### Installation

Add this package to your list of dependencies in `mix.exs`, then run `mix deps.get`:

```elixir
def deps do
  [
    {:bulk_upsert, "~> 0.3"}
  ]
end
```

### Usage

After the package has been installed, you may call `BulkUpsert.bulk_upsert/4` directly, or create a wrapper function to use in your context modules:

`lib/your_project/repo.ex`
```elixir
defmodule YourProject.Repo do
  use Ecto.Repo,
    otp_app: :your_project,
    adapter: Ecto.Adapters.Postgres

  @doc "Wraps `BulkUpsert.bulk_upsert/4`."
  def bulk_upsert(schema_module, attrs_list, opts \\ []),
    do: BulkUpsert.bulk_upsert(__MODULE__, schema_module, attrs_list, opts)
end
```

#### Basic working example

Here is a contrived migration and schema that we can work with:

`priv/repo/migrations/0001_create_persons.exs`
```elixir
defmodule YourProject.Repo.Migrations.CreatePersons do
  use Ecto.Migration

  def change do
    create table(:persons) do
      add :name, :string
    end
  end
end
```

`lib/your_project/persons/person.ex`
```elixir
defmodule YourProject.Persons.Person do
  use Ecto.Schema
  import Ecto.Changeset

  schema "persons" do
    field :name, :string
  end

  def changeset(person \\ %__MODULE__{}, attrs) do
    person
    |> cast(attrs, [:id, :name])
    |> validate_required([:id, :name])
  end
end
```

Now, after running the migrations with `mix ecto.reset`, we can enter an IEx shell with `iex -S mix` and make sure everything works:

```text
Interactive Elixir (1.18.3) - press Ctrl+C to exit (type h() ENTER for help)

iex> YourProject.Repo.bulk_upsert(
...>   YourProject.Persons.Person,
...>   [%{id: 1, name: "Alice"}, %{id: 2, name: "Bob"}]
...> )
{:ok, %{upserted: 2, skipped: 0}}

iex> YourProject.Repo.all(YourProject.Persons.Person)
[
  %YourProject.Persons.Person{
    __meta__: #Ecto.Schema.Metadata<:loaded, "persons">,
    id: 1,
    name: "Alice"
  },
  %YourProject.Persons.Person{
    __meta__: #Ecto.Schema.Metadata<:loaded, "persons">,
    id: 2,
    name: "Bob"
  }
]

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

iex> YourProject.Repo.all(YourProject.Persons.Person)
[
  %YourProject.Persons.Person{
    __meta__: #Ecto.Schema.Metadata<:loaded, "persons">,
    id: 1,
    name: "Alicia"
  },
  %YourProject.Persons.Person{
    __meta__: #Ecto.Schema.Metadata<:loaded, "persons">,
    id: 2,
    name: "Bobby"
  }
]
```

Rows whose changesets are invalid are skipped rather than upserted. The counts in the return value make this visible, and each skipped row is logged at the `:warning` level.

#### Working with nested associations

The main reason to reach for this package over a plain `insert_all/3` is that it can upsert a parent and its children at the same time, from a single list of attrs. The parent and each association are upserted into their own tables, all within one transaction.

Here we extend the `Person` example with a `has_many :pets` association:

`priv/repo/migrations/0002_create_pets.exs`
```elixir
defmodule YourProject.Repo.Migrations.CreatePets do
  use Ecto.Migration

  def change do
    create table(:pets) do
      add :person_id, references(:persons)
      add :name, :string
    end
  end
end
```

`lib/your_project/persons/pet.ex`
```elixir
defmodule YourProject.Persons.Pet do
  use Ecto.Schema
  import Ecto.Changeset

  schema "pets" do
    field :person_id, :integer
    field :name, :string
  end

  def changeset(pet \\ %__MODULE__{}, attrs) do
    pet
    |> cast(attrs, [:id, :person_id, :name])
    |> validate_required([:id, :person_id, :name])
  end
end
```

`lib/your_project/persons/person.ex`
```elixir
defmodule YourProject.Persons.Person do
  use Ecto.Schema
  import Ecto.Changeset

  schema "persons" do
    field :name, :string

    has_many :pets, YourProject.Persons.Pet
  end

  def changeset(person \\ %__MODULE__{}, attrs) do
    person
    |> cast(attrs, [:id, :name])
    |> validate_required([:id, :name])
    |> cast_assoc(:pets)
  end
end
```

> #### Note
>
> Each child's foreign key (here, `person_id`) must be present in its own attrs. Associations are upserted via `insert_all/3`, so the foreign key is not inferred from the parent.

Now a single call upserts both the persons and their pets across both tables:

```text
iex> YourProject.Repo.bulk_upsert(
...>   YourProject.Persons.Person,
...>   [
...>     %{id: 1, name: "Alice", pets: [
...>       %{id: 10, person_id: 1, name: "Rex"},
...>       %{id: 11, person_id: 1, name: "Whiskers"}
...>     ]},
...>     %{id: 2, name: "Bob", pets: [
...>       %{id: 20, person_id: 2, name: "Buddy"}
...>     ]}
...>   ]
...> )
{:ok, %{upserted: 2, skipped: 0}}

iex> YourProject.Repo.all(YourProject.Persons.Pet)
[
  %YourProject.Persons.Pet{id: 10, person_id: 1, name: "Rex"},
  %YourProject.Persons.Pet{id: 11, person_id: 1, name: "Whiskers"},
  %YourProject.Persons.Pet{id: 20, person_id: 2, name: "Buddy"}
]
```

Running the same call again with changed pet names upserts the existing rows in place, exactly like the top-level structs. This is an upsert-only operation: children absent from the attrs are never deleted or nilified, at any level.

`has_one` and `many_to_many` associations work the same way: cast them in the changeset and include them in the attrs. For `has_many` and `has_one`, each child must carry its own foreign key (as shown above with `person_id`). For `many_to_many`, the associated records and the join table rows are both upserted for you, and duplicate records and links are removed automatically. Embedded schemas (`embeds_one`, `embeds_many`) have no table of their own, so they are stored inline on the parent row.

Nesting works recursively at any depth — a child's own associations (e.g. the pets' vet appointments) are upserted the same way.

## Recipes

### Autogenerated timestamps

Ecto's built-in `insert_all/3` function does not autogenerate fields such as timestamps, so schemas with `timestamps()` fields need those values supplied during the bulk upsert. The simplest way is the `:placeholders` option, which sets fields from shared values (sent to the database once):

```elixir
YourProject.Repo.bulk_upsert(
  YourProject.Persons.Person,
  [%{id: 1, name: "Alice"}, %{id: 2, name: "Bob"}],
  placeholders: %{
    YourProject.Persons.Person => %{inserted_at: DateTime.utc_now(), updated_at: DateTime.utc_now()}
  }
)
```

Each placeholder value is injected into the attrs before validation, so a placeholder field is cast and validated like any other field (and may be marked as required in your changeset). The shared value replaces any per-row value supplied for the field.

To preserve the original insert timestamp when an existing row is updated, combine placeholders with `:replace_all_except`:

```elixir
YourProject.Repo.bulk_upsert(
  YourProject.Persons.Person,
  attrs_list,
  placeholders: %{
    YourProject.Persons.Person => %{inserted_at: DateTime.utc_now(), updated_at: DateTime.utc_now()}
  },
  # On conflict, replace every field except the primary key and :inserted_at
  replace_all_except: [:inserted_at]
)
```

If you need per-call logic that placeholders cannot express, pass a custom function that accepts the same arguments as `insert_all/3` via the `:insert_all_function_atom` (or `:insert_all_function_module`) option — see the [options documentation](https://hexdocs.pm/bulk_upsert/BulkUpsert.html#bulk_upsert/4-options).

### Recovering dirty data

When importing messy data, `:recover_changeset_errors` replaces invalid field values with per-schema fallbacks instead of skipping the whole row. Here, a person with a missing (required) name is upserted with the fallback name instead of being skipped:

```elixir
YourProject.Repo.bulk_upsert(
  YourProject.Persons.Person,
  [%{id: 1}, %{id: 2, name: "Bob"}],
  recover_changeset_errors: %{YourProject.Persons.Person => %{name: "UNKNOWN"}}
)
```

Fallbacks apply recursively to nested association and embedded changesets, and a row is only recovered if every error in it has a fallback.

### Per-schema conflict handling

By default, a conflicting row has all of its fields replaced except the primary key. Use `:insert_all_opts` to override the conflict behavior for specific schemas (or join-table sources):

```elixir
YourProject.Repo.bulk_upsert(
  YourProject.Persons.Person,
  attrs_list,
  insert_all_opts: %{
    # Never update existing persons; only insert new ones
    YourProject.Persons.Person => [on_conflict: :nothing],
    # Only update a pet's name on conflict
    YourProject.Persons.Pet => [on_conflict: {:replace, [:name]}]
  }
)
```

---

For more information, see [this project's documentation on HexDocs](https://hexdocs.pm/bulk_upsert/BulkUpsert.html).

---

This project made possible by Interline Travel and Tour Inc.

https://www.perx.com/

https://www.touchdown.co.uk/

https://www.touchdownfrance.com/