Ecto.Multiを使ってみた

はじめに

本記事はfukuoka.ex#51：Elixirお茶会〜Ectoを学ぼうの会 LT資料になります
Multiのよく使うであろう関数の紹介と実務で使いそうなユースケースの実装を紹介します

Ecto

Ecto is the database wrapper and query generator for Elixir. 
EctoはElixirのためのDBラッパー, クエリージェネレーターです

from https://hexdocs.pm/ecto/getting-started.html#content

Ecto.Multi

Ecto.Multi is a data structure for grouping multiple Repo operations.
Ecto.Multiは複数のRepoオペレーションをグルーピングするためのデータ構造です

オペレーションのグルーピングは主にDB更新を行う changeset系とDB操作以外も行える runがあります

changeset系

色々処理をしてchangesetを作成して、Repo insert, update, delete等を実行します
なので最後を%Book{title: "Elixir"}のようなスキーマかChangesetにする必要があります

run

メール送信やファイルアップロードなどDBとは関係ないことや、　
MyApp.Books.create_bookなどDB更新系も実行できます
最後は構造体である必要がありません

Ecto.Multi functions

今回はよく使うであろう基本的な関数を紹介します

new

単体では使いません 実行結果をボコボコ入れていきます

insert

Ecto.Multi.new()
|> Ecto.Multi.insert(:insert, %Post{title: "first"})
|> MyApp.Repo.transaction()

第１引数に　Multi.new, ２に key, 3に　構造体を指定します
keyはAtomもStringもいけます
最後に transaction()を実行しているので失敗した場合はdbがrollbackされます

Ecto.Multi.new()
|> Ecto.Multi.insert(:post, %Post{title: "first"})
|> Ecto.Multi.insert(:comment, fn %{post: post} ->
     Ecto.build_assoc(post, :comments)
   end)
|> MyApp.Repo.transaction()

こちらはinsertを２つつなげています
第３引数は関数にもできて、build_assocしてリレーション情報をつけたり、ファイルアップロードしたりと色々できます

最後はSchemaかChangesetを返してください
transaction()をつけているので、comment insertで失敗したらpostもrollbackされます

第２引数で指定するkeyですが、第３引数の関数でパターンマッチで取得する際に使用します
１つ目のinsertで:postと名前をつけているので、
2つ目のinsertでEcto.Multiから %{post: post} で参照できます

insert_all

posts = [%{title: "My first post"}, %{title: "My second post"}]
Ecto.Multi.new()
|> Ecto.Multi.insert_all(:insert_all, Post, posts)
|> MyApp.Repo.transaction()

複数のデータをまとめて追加します
keyとSchema module, Mapのリストを引数に取ります

注意点があって
そのままEcto.Changeset.change |> Ecto.Repo.insert_allで実行されるので、
castで実行される関数やバリデーションやSchemaで指定している初期値は無視されるので
inserted_at,updated_atをMapの方に追加していないとnot null属性なのでエラーになります

update

post = MyApp.Repo.get!(Post, 1)
changeset = Ecto.Changeset.change(post, title: "New title")
Ecto.Multi.new()
|> Ecto.Multi.update(:update, changeset)
|> MyApp.Repo.transaction()

最後に返すのがChangesetで固定になるくらいであとはinsertと同じです

update_all

Ecto.Multi.new()
|> Ecto.Multi.update_all(:update_all, Post, set: [title: "New title"])
|> MyApp.Repo.transaction()

更新する項目にupdate operateを実行します
set, inc(integer), push(array), pull(array)
そのままだと全件更新されて使い勝手があまり良くないので、
第３引数を無名関数にして from文で絞り込み等を行う機会が多そうです

delete

post = MyApp.Repo.get!(Post, 1)
Ecto.Multi.new()
|> Ecto.Multi.delete(:delete, post)
|> MyApp.Repo.transaction()

単純に１件消したり、無名関数でいろいろしたあとに消したりするときに使います

delete_all

queryable = from(p in Post, where: p.id < 5)
Ecto.Multi.new()
|> Ecto.Multi.delete_all(:delete_all, queryable)
|> MyApp.Repo.transaction()

fromで絞り込んで消す

inspect

IO.inspectだと最終的なデータが
Multi.inspect()ではkeyと実行されたオペレーションが表示されます
エラーのときにはMulti.inspect()の方がデバッグがしやすいです

# Ecto.Multi.inspect
%Ecto.Multi{
  names: #MapSet<[:insert]>,
  operations: [
    inspect: {:inspect, []},
    insert: {:changeset,
     #Ecto.Changeset<action: :insert, changes: %{}, errors: [],
      data: #MyApp.Books.Book<>, valid?: true>, []}
  ]
}
# IO.inspect
%{
  insert: %MyApp.Books.Book{
    __meta__: #Ecto.Schema.Metadata<:loaded, "books">, 
    id: 1,
    inserted_at: ~N[2022-02-23 17:11:18],
    price: 3000,
    title: "Programing Elixir",
    updated_at: ~N[2022-02-23 17:11:18]
  }
}

run

Ecto.Multi.run(multi, :write, fn _repo, %{image: image} ->
  with :ok <- File.write(image.name, image.contents) do
    {:ok, nil}
  end
end)

DB系に限らずいろいろ実行できます
上記の例だと fn %{write: result} -> end で他のMultiから参照すると
resultと{:ok, nil}が束縛されるので、 case文で処理を分けたりできます

ユースケース

実際の業務でありえるだろうユースケースを簡略化して実行してみます
こちらで作った画像つきブログを作成するアプリケーションをベースにしています

一度レコードを登録してからファイルをアップロードしたい!

ファイルをアップロードサンプルは以下みたいな流れですが、
save event -> upload gcs or aws or local -> insert record

実務だと insert時にバリデーションエラーが起こった際にゴミファイルがすでにアップロードされているといったことになるよで良くないです

なので先に insertを成功して、その後にアップロードするという流れが一般的です
それを以下に説明していきます

# assigns
%{
  __changed__: %{},
  action: :new,
  changeset: #Ecto.Changeset<
    action: :validate,
    changes: %{body: "hoge", title: "test"},
    errors: [eyecatch: {"can't be blank", [validation: :required]}],
    data: #Blog.Blogs.Post<>,
    valid?: false
  >,
  flash: %{},
  id: :new,
  myself: %Phoenix.LiveComponent.CID{cid: 1},
  post: %Blog.Blogs.Post{
    __meta__: #Ecto.Schema.Metadata<:built, "posts">,
    body: nil,
    eyecatch: nil,
    id: nil,
    inserted_at: nil,
    title: nil,
    updated_at: nil
  },
  return_to: "/posts",
  title: "New Post",
  uploads: %{
    __phoenix_refs_to_names__: %{"phx-FtakzruON8ievwEE" => :image},
    image: #Phoenix.LiveView.UploadConfig<
      accept: :any,
      auto_upload?: false,
      entries: [
        %Phoenix.LiveView.UploadEntry{
          cancelled?: false,
          client_last_modified: nil,
          client_name: "DyKnUXQUwAAUGfG.jpeg",
          client_size: 131516,
          client_type: "image/jpeg",
          done?: true,
          preflighted?: true,
          progress: 100,
          ref: "0",
          upload_config: :image, 
          upload_ref: "phx-FtakzruON8ievwEE",
          uuid: "a9c37604-ebff-456b-8db6-13610888cd22",
          valid?: true
        }
      ],
      errors: [],
      max_entries: 1,
      max_file_size: 8000000,
      name: :image,
      progress_event: nil,
      ref: "phx-FtakzruON8ievwEE",
      ...
    >
  }
}

save eventをformを入力して、画像をアップロードした状態で実行すると上記のようなassign(LiveView上の変数郡）が渡されます

• changeset　=> changes: %{body: "hoge", title: "test"}　がフォーム入力値
• post => %Blog.Blogs.Postが空の構造体
• uploads => entriesがアップロードしたファイル
  をそれぞれ表しています

  defp save_post(%{assigns: assigns} = socket, :new, post_params) do
    Ecto.Multi.new()
    |> Ecto.Multi.insert(:insert, fn _ ->
      post_params
      |> Map.put("eyecatch", "placeholder")
      |> then(&Blogs.change_post(assigns.post, &1))
    end)
    |> Ecto.Multi.update(:upload, fn %{insert: post} ->
      Blogs.change_post(post, upload_function(socket))
    end)
    |> Blog.Repo.transaction()
    |> IO.inspect()
    |> case  do
      {:ok, _post} ->
        {:noreply,
         socket
         |> put_flash(:info, "Post created successfully")
         |> push_redirect(to: socket.assigns.return_to)}

      {:error, _name ,%Ecto.Changeset{} = changeset, _} ->
        {:noreply, assign(socket, changeset: changeset)}
   end
  end

実際のEcto.Multiのコードです
Multi.insertで オペレーション名 :insertでpostを保存します
保存する際に画像パスを保存する eyecatchがnot null属性でそのままだとエラーになるので、空文字でもいいですがわかりやすくするためにplaceholderを入れておきます

次にMulti.updateで :insert オペレーションの実行結果を取得して、　オペレーション名:uploadで後述するupload_functionを実行してファイルアップロードを行い、そのファイルパスを change_postでeyecatchの値を変更して更新を行います

最後に case で条件分岐を行うのですが
{:ok,　_post} はMultiが返ってくるのでそのまま
{:error, changeset}は {:error, 失敗したオペレーション, changeset, エラーメッセージ}となるので変更を忘れないようにしましょう

  defp upload_function(socket) do
    upload_path =
      consume_uploaded_entries(socket, :image, fn %{path: path}, _entry ->
        dest = Path.join([:code.priv_dir(:blog), "uploads", Path.basename(path)])
        File.cp!(path, dest)
        Routes.static_path(socket, "/uploads/#{Path.basename(dest)}")
      end)
      |> List.first
    %{"eyecatch" => upload_path}
  end

以下成功時のログになります
Multiにはinsert時の結果とupdate時の結果が両方保持されているのでテストの際に変更されたか？をチェックするのが楽です

[debug] QUERY OK db=0.4ms idle=1614.8ms
begin []
[debug] QUERY OK db=1.3ms
INSERT INTO "posts" ("body","eyecatch","title","inserted_at","updated_at") VALUES ($1,$2,$3,$4,$5) RETURNING "id" ["hoge", "placeholder", "test", ~N[2022-02-24 07:10:34], ~N[2022-02-24 07:10:34]]
[debug] QUERY OK db=0.1ms
UPDATE "posts" SET "eyecatch" = $1, "updated_at" = $2 WHERE "id" = $3 ["/uploads/live_view_upload-1645686634-353389780394486-1", ~N[2022-02-24 07:10:34], 1]
[debug] QUERY OK db=0.2ms
commit []
{:ok,
 %{
   insert: %Blog.Blogs.Post{
     __meta__: #Ecto.Schema.Metadata<:loaded, "posts">,
     body: "hoge",
     eyecatch: "placeholder",
     id: 1,
     inserted_at: ~N[2022-02-24 07:10:34],
     title: "test",
     updated_at: ~N[2022-02-24 07:10:34]
   },
   upload: %Blog.Blogs.Post{
     __meta__: #Ecto.Schema.Metadata<:loaded, "posts">,
     body: "hoge",
     eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1",
     id: 1,
     inserted_at: ~N[2022-02-24 07:10:34],
     title: "test",
     updated_at: ~N[2022-02-24 07:10:34]
   }
 }}

失敗時のログです、失敗したオペレーションとchangesetが返ってきてます、validation エラーなのでchangesetのみでそれ以外のエラーメッセージはありません

begin []
[debug] QUERY OK db=0.2ms
rollback []
{:error, :insert,
 #Ecto.Changeset<
   action: :insert,
   changes: %{eyecatch: "placeholder", title: "test"},
   errors: [body: {"can't be blank", [validation: :required]}],
   data: #Blog.Blogs.Post<>,
   valid?: false
 >, %{}}

CSVでアップロードして一括で登録したい！

次に業務でよくあるのがデータをCSVアップロードで一括登録したいですね、
１つでもコケたら失敗する、失敗したらどの箇所が失敗したかを表示とかあるあるです

uploadとパース部分は飛ばして、以下のようなデータになるとします

parsed_data = [
 %{title: "title1", body: "body1", eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1"},
 %{title: "title2", body: "body2", eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1"},
 %{title: "title3", body: "body3", eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1"},
 %{title: "title4", body: "body4", eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1"}
]

Ecto.Multi.insert_allの場合

上記のMapのままでは失敗するので、データを加工します

  def build_post(data) do
    now = NaiveDateTime.local_now()

    Enum.map(data, fn post ->
      %{
        title: post.title,
        body: post.body,
        eyecatch: post.eyecatch,
        inserted_at: now,
        updated_at: now
      }
    end)
  end

実行して成功すると以下のようになり　何件成功したかが{4, nil}と返ってきてます

Ecto.Multi.new()
|> Ecto.Multi.insert_all(:insert_all, Post, Blogs.build_post(parsed_data))
|> Ecto.Multi.inspect
|> Repo.transaction() 

# log
[debug] QUERY OK db=1.8ms idle=1891.9ms
begin []
[debug] QUERY OK db=0.9ms
INSERT INTO "posts" ("body","eyecatch","inserted_at","title","updated_at") VALUES ($1,$2,$3,$4,$5),($6,$7,$8,$9,$10),($11,$12,$13,$14,$15),($16,$17,$18,$19,$20) ["body1", "/uploads/live_view_upload-1645686634-353389780394486-1", ~N[2022-02-24 16:54:30], "title1", ~N[2022-02-24 16:54:30], "body2", "/uploads/live_view_upload-1645686634-353389780394486-1", ~N[2022-02-24 16:54:30], "title2", ~N[2022-02-24 16:54:30], "body3", "/uploads/live_view_upload-1645686634-353389780394486-1", ~N[2022-02-24 16:54:30], "title3", ~N[2022-02-24 16:54:30], "body4", "/uploads/live_view_upload-1645686634-353389780394486-1", ~N[2022-02-24 16:54:30], "title4", ~N[2022-02-24 16:54:30]]
%{insert_all: {4, nil}}
[debug] QUERY OK db=0.8ms
commit []
{:ok, %{insert_all: {4, nil}}}

失敗例でデータの加工をしなかった場合は、null falseなinserted_atとupdated_atが空なので例外が発生し失敗します

Ecto.Multi.new()
|> Ecto.Multi.insert_all(:insert_all, Post,parsed_data) 
|> Blog.Repo.transaction()

# log
[debug] QUERY OK db=0.7ms queue=0.1ms idle=1824.5ms
begin []
[debug] QUERY ERROR db=3.1ms
INSERT INTO "posts" ("body","eyecatch","title") VALUES ($1,$2,$3),($4,$5,$6),($7,$8,$9),($10,$11,$12) ["body1", "/uploads/live_view_upload-1645686634-353389780394486-1", "title1", "body2", "/uploads/live_view_upload-1645686634-353389780394486-1", "title2", "body3", "/uploads/live_view_upload-1645686634-353389780394486-1", "title3", "body4", "/uploads/live_view_upload-1645686634-353389780394486-1", "title4"]
[debug] QUERY OK db=0.9ms
rollback []
** (Postgrex.Error) ERROR 23502 (not_null_violation) null value in column "inserted_at" of relation "posts" violates not-null constraint

    table: posts
    column: inserted_at

Failing row contains (6, title1, body1, /uploads/live_view_upload-1645686634-353389780394486-1, null, null).
...

Ecto.Multi.insert と Enum.reduceのあわせ技

上記のbuild_post無しでやりたい、バリデーションや値の加工処理も挟みたいという場合には
insertをEnum.reduceで回すことで実現できます

Enum.zip(1..length(parsed_data), parsed_data) 
|> Enum.reduce(Ecto.Multi.new(), fn {index, post}, multi -> 
  Ecto.Multi.insert(
    multi,
    "post_#{index}", 
    Blogs.change_post(%Post{}, post)) 
end) 
|> Repo.transaction()

# log
[debug] QUERY OK db=0.5ms idle=1008.6ms
begin []
[debug] QUERY OK db=0.8ms
INSERT INTO "posts" ("body","eyecatch","title","inserted_at","updated_at") VALUES ($1,$2,$3,$4,$5) RETURNING "id" ["body1", "/uploads/live_view_upload-1645686634-353389780394486-1", "title1", ~N[2022-02-24 08:04:34], ~N[2022-02-24 08:04:34]]
[debug] QUERY OK db=0.2ms
INSERT INTO "posts" ("body","eyecatch","title","inserted_at","updated_at") VALUES ($1,$2,$3,$4,$5) RETURNING "id" ["body2", "/uploads/live_view_upload-1645686634-353389780394486-1", "title2", ~N[2022-02-24 08:04:34], ~N[2022-02-24 08:04:34]]
[debug] QUERY OK db=0.3ms
INSERT INTO "posts" ("body","eyecatch","title","inserted_at","updated_at") VALUES ($1,$2,$3,$4,$5) RETURNING "id" ["body3", "/uploads/live_view_upload-1645686634-353389780394486-1", "title3", ~N[2022-02-24 08:04:34], ~N[2022-02-24 08:04:34]]
[debug] QUERY OK db=0.2ms
INSERT INTO "posts" ("body","eyecatch","title","inserted_at","updated_at") VALUES ($1,$2,$3,$4,$5) RETURNING "id" ["body4", "/uploads/live_view_upload-1645686634-353389780394486-1", "title4", ~N[2022-02-24 08:04:34], ~N[2022-02-24 08:04:34]]
[debug] QUERY OK db=0.8ms
commit []
{:ok,
 %{
   "post_1" => %Blog.Blogs.Post{
     __meta__: #Ecto.Schema.Metadata<:loaded, "posts">,
     body: "body1",
     eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1",
     id: 7,
     inserted_at: ~N[2022-02-24 08:04:34],
     title: "title1",
     updated_at: ~N[2022-02-24 08:04:34]
   },
   "post_2" => %Blog.Blogs.Post{
     __meta__: #Ecto.Schema.Metadata<:loaded, "posts">,
     body: "body2",
     eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1",
     id: 8,
     inserted_at: ~N[2022-02-24 08:04:34],
     title: "title2",
     updated_at: ~N[2022-02-24 08:04:34]
   },
   "post_3" => %Blog.Blogs.Post{
     __meta__: #Ecto.Schema.Metadata<:loaded, "posts">,
     body: "body3",
     eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1",
     id: 9,
     inserted_at: ~N[2022-02-24 08:04:34],
     title: "title3",
     updated_at: ~N[2022-02-24 08:04:34]
   },
   "post_4" => %Blog.Blogs.Post{
     __meta__: #Ecto.Schema.Metadata<:loaded, "posts">,
     body: "body4",
     eyecatch: "/uploads/live_view_upload-1645686634-353389780394486-1",
     id: 10,
     inserted_at: ~N[2022-02-24 08:04:34],
     title: "title4",
     updated_at: ~N[2022-02-24 08:04:34]
   }
 }}

オペレーション名ですがユニークである必要があるので、 Enum.zipでindex付きでreduceを実行します

第２引数にEcto.Multi.new()をセットすることで複数のMulti.insertの結果を集約できます
あとは第３引数の無名関数で {index, post}のタプルでMulti.insertを実行していきます

SQLが一つずつ走るので効率は良くはないですが、どこが失敗してどんなエラーになっているのかが追いやすくなるかと思います