diff --git a/lib/music_library/artists/batch.ex b/lib/music_library/artists/batch.ex index b3213db1..88bba49f 100644 --- a/lib/music_library/artists/batch.ex +++ b/lib/music_library/artists/batch.ex @@ -3,52 +3,23 @@ defmodule MusicLibrary.Artists.Batch do alias MusicLibrary.Artists alias MusicLibrary.Artists.ArtistInfo - alias MusicLibrary.Repo - - require Logger + alias MusicLibrary.Batch def refresh_musicbrainz_data do - run_on_all_artist_infos(fn artist_info -> + Batch.run_on_all(from(r in ArtistInfo), "artist_info", fn artist_info -> Artists.refresh_musicbrainz_data_async(artist_info) end) end def refresh_discogs_data do - run_on_all_artist_infos(fn artist_info -> + Batch.run_on_all(from(r in ArtistInfo), "artist_info", fn artist_info -> Artists.refresh_discogs_data_async(artist_info) end) end def refresh_wikipedia_data do - run_on_all_artist_infos(fn artist_info -> + Batch.run_on_all(from(r in ArtistInfo), "artist_info", fn artist_info -> Artists.refresh_wikipedia_data_async(artist_info) end) end - - defp run_on_all_artist_infos(fun) do - q = from(r in ArtistInfo) - stream = Repo.stream(q, max_rows: 50) - - Repo.transaction( - fn -> - Enum.reduce(stream, [], fn artist_info, acc -> - case fun.(artist_info) do - {:error, reason} -> - Logger.error( - "Failed to run function on artist_info #{artist_info.id} with #{inspect(reason)}" - ) - - [artist_info.id | acc] - - :ok -> - acc - - {:ok, _artist_info} -> - acc - end - end) - end, - timeout: :infinity - ) - end end diff --git a/lib/music_library/batch.ex b/lib/music_library/batch.ex new file mode 100644 index 00000000..f474acb7 --- /dev/null +++ b/lib/music_library/batch.ex @@ -0,0 +1,31 @@ +defmodule MusicLibrary.Batch do + alias MusicLibrary.Repo + + require Logger + + def run_on_all(queryable, label, fun) do + stream = Repo.stream(queryable, max_rows: 50) + + Repo.transaction( + fn -> + Enum.reduce(stream, [], fn record, acc -> + case fun.(record) do + {:error, reason} -> + Logger.error( + "Failed to run function on #{label} #{record.id} with #{inspect(reason)}" + ) + + [record.id | acc] + + :ok -> + acc + + {:ok, _result} -> + acc + end + end) + end, + timeout: :infinity + ) + end +end diff --git a/lib/music_library/records/batch.ex b/lib/music_library/records/batch.ex index b45e4217..fbb59a3e 100644 --- a/lib/music_library/records/batch.ex +++ b/lib/music_library/records/batch.ex @@ -1,48 +1,19 @@ defmodule MusicLibrary.Records.Batch do import Ecto.Query + alias MusicLibrary.Batch alias MusicLibrary.Records alias MusicLibrary.Records.Record - alias MusicLibrary.Repo - - require Logger def refresh_musicbrainz_data do - run_on_all_records(fn record -> + Batch.run_on_all(from(r in Record), "record", fn record -> Records.refresh_musicbrainz_data_async(record) end) end def generate_embeddings do - run_on_all_records(fn record -> + Batch.run_on_all(from(r in Record), "record", fn record -> Records.generate_embedding_async(record) end) end - - defp run_on_all_records(fun) do - q = from(r in Record) - stream = Repo.stream(q, max_rows: 50) - - Repo.transaction( - fn -> - Enum.reduce(stream, [], fn record, acc -> - case fun.(record) do - {:error, reason} -> - Logger.error( - "Failed to run function on record #{record.id} with #{inspect(reason)}" - ) - - [record.id | acc] - - :ok -> - acc - - {:ok, _record} -> - acc - end - end) - end, - timeout: :infinity - ) - end end