Skip to main content

README.md

# BulkUpsert

> #### This package has been renamed to `bulkinup`
>
> `bulk_upsert` continues as [`bulkinup`](https://hex.pm/packages/bulkinup), and this package
> will receive no further updates. Migrating is a rename: `BulkUpsert.bulk_upsert/4` is now
> `Bulkinup.upsert/4` — same arguments, same options, same return shape. See the
> [migration guide](https://hexdocs.pm/bulkinup/migrating_from_bulk_upsert.html).

> #### 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 (by default), so any failure rolls back all changes
  - Streaming input: pass any `Enumerable` (including a lazy `Stream`) to upsert arbitrarily large inputs with bounded memory
  - Optional concurrent upserts via `:max_concurrency`, trading the single transaction for throughput

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.5.0"}
  ]
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 call that skips rows emits one `:warning` log summarizing them (with per-row detail at the `:debug` 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.

### Streaming large imports

`attrs_list` accepts any `Enumerable`, so a large input can be streamed instead of loaded into memory:

```elixir
"people.csv"
|> File.stream!()
|> Stream.map(&parse_csv_row/1)
|> then(&YourProject.Repo.bulk_upsert(YourProject.Persons.Person, &1))
```

Rows are validated and upserted in chunks of `:chunk_size` as the stream is consumed, so memory stays bounded no matter how large the input is. Note that the single transaction stays open for the stream's full duration. To trade the single-transaction guarantee for throughput (chunks upserted concurrently, each in its own transaction), see the `:max_concurrency` option in the [options documentation](https://hexdocs.pm/bulk_upsert/BulkUpsert.html#bulk_upsert/4-options) — and read its caveats first.

### 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/