Ограничения и апсерты

Исходный код

В этом руководстве мы узнаем, как использовать ограничения и апсерты. Чтобы показать эти возможности, мы рассмотрим практический сценарий: изучим связь «многие ко многим» между публикациями и тегами.

put_assoc и cast_assoc

Представим, что мы создаём приложение с публикациями в блоге, у каждой из которых может быть много тегов. Более того, один тег может относиться ко многим публикациям. Это классический сценарий, в котором мы использовали бы ассоциации many_to_many. Наши миграции выглядели бы так:

create table(:posts) do
  add :title, :string
  add :body, :text
  timestamps()
end

create table(:tags) do
  add :name, :string
  timestamps()
end

create unique_index(:tags, [:name])

create table(:posts_tags, primary_key: false) do
  add :post_id, references(:posts)
  add :tag_id, references(:tags)
end

Обратите внимание: мы добавили уникальный индекс для имени тега, поскольку не хотим, чтобы в нашей базе данных были повторяющиеся теги. Важно добавить индекс на уровне базы данных, а не использовать валидацию, поскольку всегда есть вероятность, что два тега с одинаковым именем пройдут валидацию и будут вставлены одновременно, что приведёт к появлению дубликатов.

Теперь представим, что мы хотим, чтобы пользователь вводил теги в виде списка слов, разделённых запятыми, например: "elixir, erlang, ecto". Получив эти данные на сервере, мы разделим их на несколько тегов и свяжем с публикацией, создав все теги, которых ещё нет в базе данных.

Хотя описанные выше ограничения кажутся разумными, именно они создают проблемы при использовании cast_assoc/3. Функция changeset cast_assoc/3 предназначена для получения внешних параметров и их сравнения со связанными данными в наших структурах. Чтобы сделать это правильно, Ecto требует передавать теги в виде списка карт. Пример можно увидеть в разделе Полиморфные ассоциации «многие ко многим». Однако здесь мы ожидаем, что теги будут переданы в виде строки, разделённой запятыми.

Кроме того, cast_assoc/3 использует поле первичного ключа каждого переданного тега, чтобы определить, следует ли его вставить, обновить или удалить. Но поскольку пользователь просто передаёт строку, у нас нет информации об ID.

Когда мы не можем справиться с cast_assoc/3, пора использовать put_assoc/4. В put_assoc/4 мы передаём Ecto структуры или changeset, а не параметры, и поэтому можем манипулировать данными как угодно. Определим схему и функцию changeset для публикации, которая может получать теги в виде строки:

defmodule MyApp.Post do
  use Ecto.Schema

  schema "posts" do
    field :title
    field :body

    many_to_many :tags, MyApp.Tag,
      join_through: "posts_tags",
      on_replace: :delete

    timestamps()
  end

  def changeset(struct, params \\ %{}) do
    struct
    |> Ecto.Changeset.cast(params, [:title, :body])
    |> Ecto.Changeset.put_assoc(:tags, parse_tags(params))
  end

  defp parse_tags(params)  do
    (params["tags"] || "")
    |> String.split(",")
    |> Enum.map(&String.trim/1)
    |> Enum.reject(& &1 == "")
    |> Enum.map(&get_or_insert_tag/1)
  end

  defp get_or_insert_tag(name) do
    Repo.get_by(MyApp.Tag, name: name) ||
      Repo.insert!(%Tag{name: name})
  end
end

В приведённой выше функции changeset мы вынесли всю обработку тегов в отдельную функцию с именем parse_tags/1. Она проверяет наличие параметра, разделяет каждый тег с помощью String.split/2, затем удаляет оставшиеся пробелы с помощью String.trim/1, отбрасывает пустые строки и, наконец, проверяет, существует ли тег в базе данных, создавая его, если он не найден.

Функция parse_tags/1 возвращает список структур MyApp.Tag, которые затем передаются в put_assoc/4. Вызывая put_assoc/4, мы сообщаем Ecto, что с этого момента с публикацией должны быть связаны именно эти теги. Если с публикацией был связан тег, которого нет в put_assoc/4, Ecto вызовет поведение, заданное параметром :on_replace, которому мы присвоили значение :delete. Поведение :delete удалит из базы данных связь между публикацией и удалённым тегом.

Вот и всё, что нужно для использования ассоциаций many_to_many с put_assoc/4. put_assoc/4 очень полезен, когда требуется более явно управлять ассоциациями. Он также работает с has_many, belongs_to и всеми остальными типами ассоциаций.

Однако наш код ещё не готов к эксплуатации. Давайте разберёмся почему.

Ограничения и состояния гонки

Помните, что при создании таблицы тегов мы добавили уникальный индекс для столбца :name. Это защищает нас от появления в базе данных повторяющихся тегов.

Добавив уникальный индекс, а затем используя Repo.get_by вместе с Repo.insert! для поиска или вставки тега, мы создали потенциальную ошибку в приложении. Если одновременно отправить две публикации с одинаковым тегом, может случиться так, что обе проверки существования тега выполнятся одновременно. В результате обе операции решат, что такого тега в базе данных нет. Тогда одна операция завершится успешно, а другая завершится ошибкой. Это состояние гонки: иногда ваш код будет выдавать ошибку, только когда выполняются определённые условия. Причём эти условия зависят от времени.

К счастью, Ecto предоставляет механизм обработки ошибок ограничений, возникающих в базе данных.

Обработка ошибок ограничений

Поскольку функция get_or_insert_tag(name) завершается ошибкой, если тег уже существует в базе данных, нам нужно правильно обработать такие ситуации. Перепишем её с учётом состояний гонки:

defp get_or_insert_tag(name) do
  %Tag{}
  |> Ecto.Changeset.change(name: name)
  |> Ecto.Changeset.unique_constraint(:name)
  |> Repo.insert()
  |> case do
    {:ok, tag} -> tag
    {:error, _} -> Repo.get_by!(MyApp.Tag, name: name)
  end
end

Вместо прямой вставки тега мы теперь создаём changeset, что позволяет использовать аннотацию unique_constraint. Если операция Repo.insert завершится неудачей из-за нарушения уникального индекса для :name, Ecto не вызовет исключение, а вернёт кортеж {:error, changeset}. Поэтому, если Repo.insert завершается успешно, значит, тег был сохранён; иначе тег уже существует, и мы получаем его с помощью Repo.get_by!.

Хотя описанный выше механизм устраняет состояние гонки, он довольно затратен: для каждого тега, который уже есть в базе данных, нам нужно выполнить два запроса — неудачную вставку, а затем поиск в репозитории. Поскольку это наиболее распространённый сценарий, можно переписать код следующим образом:

defp get_or_insert_tag(name) do
  Repo.get_by(MyApp.Tag, name: name) ||
    maybe_insert_tag(name)
end

defp maybe_insert_tag(name) do
  %Tag{}
  |> Ecto.Changeset.change(name: name)
  |> Ecto.Changeset.unique_constraint(:name)
  |> Repo.insert
  |> case do
    {:ok, tag} -> tag
    {:error, _} -> Repo.get_by!(MyApp.Tag, name: name)
  end
end

Приведённый выше код выполняет 1 запрос для каждого уже существующего тега, 2 запроса для каждого нового тега и, возможно, 3 запроса в случае состояния гонки. Хотя в среднем этот вариант будет немного эффективнее, у Ecto есть решение получше.

Апсерты

Ecto поддерживает так называемую команду «upsert» — сокращение от «update or insert» («обновить или вставить»). Идея заключается в том, что мы пытаемся вставить запись, а если она конфликтует с уже существующей, например из-за уникального индекса, то можем выбрать, как должна действовать база данных: вызвать ошибку (поведение по умолчанию), проигнорировать вставку (без ошибки) или обновить конфликтующие записи в базе данных.

В Ecto «upsert» выполняется с помощью параметра :on_conflict. Перепишем get_or_insert_tag(name) ещё раз, но теперь воспользуемся параметром :on_conflict. Помните, что «upsert» — это новая возможность PostgreSQL 9.5, поэтому убедитесь, что у вас установлена актуальная версия.

Для начала можно попробовать использовать :on_conflict, задав ему значение :nothing, как показано ниже:

defp get_or_insert_tag(name) do
  Repo.insert!(
    %MyApp.Tag{name: name},
    on_conflict: :nothing
  )
end

Приведённый выше вариант не вызовет ошибку при конфликте, но и не обновит переданную структуру, поэтому вернёт тег без ID. Одно из решений — принудительно выполнять обновление при конфликте, даже если оно сводится к присваиванию имени тега его текущего значения. В таких случаях PostgreSQL также требует указывать параметр :conflict_target — столбец (или список столбцов), по которому может произойти конфликт:

defp get_or_insert_tag(name) do
  Repo.insert!(
    %MyApp.Tag{name: name},
    on_conflict: [set: [name: name]],
    conflict_target: :name
  )
end

Вот и всё! Мы пытаемся вставить тег с указанным именем, а если такой тег уже существует, сообщаем Ecto, что нужно обновить его имя до текущего значения, обновив тег и получив его ID. Это, безусловно, шаг вперёд по сравнению со всеми предыдущими решениями, но всё ещё выполняет по одному запросу для каждого тега. Если передано 10 тегов, мы выполним 10 запросов. Можно ли сделать ещё лучше?

Апсерты и insert_all

Ecto принимает параметр :on_conflict не только в Ecto.Repo.insert/2, но и в функции Ecto.Repo.insert_all/3. Это значит, что мы можем сформировать один запрос для вставки всех отсутствующих тегов, а затем другой запрос для получения их всех сразу. Посмотрим, как будет выглядеть схема Post после этих изменений:

defmodule MyApp.Post do
  use Ecto.Schema

  # We need to import Ecto.Query
  import Ecto.Query

  # Schema is the same
  schema "posts" do
    add :title
    add :body

    many_to_many :tags, MyApp.Tag,
      join_through: "posts_tags",
      on_replace: :delete

    timestamps()
  end

  # Changeset is the same
  def changeset(struct, params \\ %{}) do
    struct
    |> Ecto.Changeset.cast(params, [:title, :body])
    |> Ecto.Changeset.put_assoc(:tags, parse_tags(params))
  end

  # Parse tags has slightly changed
  defp parse_tags(params)  do
    (params["tags"] || "")
    |> String.split(",")
    |> Enum.map(&String.trim/1)
    |> Enum.reject(& &1 == "")
    |> insert_and_get_all()
  end

  defp insert_and_get_all([]) do
    []
  end
  defp insert_and_get_all(names) do
    timestamp =
      NaiveDateTime.utc_now()
      |> NaiveDateTime.truncate(:second)

    placeholders = %{timestamp: timestamp}

    maps =
      Enum.map(names, &%{
        name: &1,
        inserted_at: {:placeholder, :timestamp},
        updated_at: {:placeholder, :timestamp}
      })

    Repo.insert_all(
      MyApp.Tag,
      maps,
      placeholders: placeholders,
      on_conflict: :nothing
    )

    Repo.all(from t in MyApp.Tag, where: t.name in ^names)
  end
end

Вместо того чтобы получать и вставлять каждый тег отдельно, приведённый выше код обрабатывает все теги одновременно. Сначала он формирует список карт, который передаётся в Repo.insert_all. Затем мы ищем все теги с указанными именами. Независимо от количества тегов мы выполним только 2 запроса — если только не передано ни одного тега, в этом случае мы сразу вернём пустой список. Это решение возможно благодаря параметру :on_conflict, который гарантирует, что insert_all не завершится ошибкой при нарушении уникального индекса, например из-за повторяющихся имён тегов. Помните, что insert_all не генерирует автоматически такие значения, как временные метки. Поэтому мы определяем заполнитель для временной метки и повторно используем его для полей inserted_at и updated_at.

Наконец, обратите внимание, что ни в одном из приведённых примеров мы не использовали транзакции. Это было сделано намеренно: мы исходили из того, что получение или вставка тегов — идемпотентная операция, то есть для одних и тех же входных данных её можно повторять сколько угодно раз, и результат всегда будет одинаковым. Поэтому, даже если добавление публикации в базу данных не удастся из-за ошибки валидации, пользователь сможет отправить форму ещё раз, и мы просто повторим попытку получить или вставить те же теги. Недостаток такого подхода в том, что теги будут созданы, даже если создание публикации завершится неудачей, а значит, некоторые теги могут остаться без связанных с ними публикаций. Если это нежелательно, всю операцию можно обернуть в транзакцию или смоделировать с помощью Ecto.Multi.

← Предыдущая страница Агрегаты и подзапросы

Следующая страница → Сопоставление данных и валидация