Packages
surgex
5.1.1
6.0.1
6.0.0
5.1.1
5.1.0
5.0.0
5.0.0-git-015aa39a
4.15.2
4.15.1
4.15.1-git-98b3
4.15.0
4.15.0-git-943e
4.15.0-git-5806
4.14.1
4.14.0
4.13.1
4.13.0
4.12.0
4.11.0
4.11.0-git-97d4
4.11.0-git-14c8
4.10.0
4.10.0-git-37f2
4.9.0
4.9.0-git-9d5f
4.8.1-git-80a7
4.8.0
4.8.0-git-fbda
4.8.0-git-e058
4.7.0
4.7.0-git-5414
4.6.1
4.6.0
4.5.0
4.4.0
4.3.0
4.2.1
4.2.0
4.1.1
4.1.1-git-9f6b
4.1.1-git-6f0cd
4.1.0
4.1.0-git-e934
4.1.0-git-8c28
4.1.0-git-56a9
4.1.0-git-55fd
4.0.0
3.2.8
3.2.7
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
retired
3.1.0
3.0.0
2.24.1
2.24.0
retired
2.23.0
2.22.0
2.21.0
2.20.1
2.20.0
2.19.0
2.18.1
2.18.0
2.17.0
2.16.0
2.15.0
2.14.0
2.13.0
2.12.1
2.12.0
2.11.0
2.10.0
2.9.0
2.8.0
2.7.0
2.6.0
2.5.1
2.5.0
2.4.0
retired
2.3.0
2.2.1
2.2.0
2.1.1
2.1.0
2.0.0
1.6.0
1.5.2
1.5.1
1.5.0
1.4.0
1.3.1
retired
1.3.0
retired
1.2.1
1.2.0
1.1.0
1.0.0
0.5.2
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.0
0.1.0
All Things Elixir @ Surge Ventures Inc, the creators of Fresha
Current section
Files
Jump to
Current section
Files
lib/surgex/data_pipe/table_sync.ex
case Code.ensure_loaded(Ecto) do
{:module, _} ->
defmodule Surgex.DataPipe.TableSync do
@moduledoc """
Extracts and transforms data from one PostgreSQL table into another.
## Usage
Refer to `Surgex.DataPipe` for a complete data pipe example.
"""
import Ecto.Query
alias Ecto.Adapters.SQL
@doc """
Synchronizes the given repository's table with data fetched using a specified query.
The synchronization is done via a single SQL query by utilizing the `WITH` statement. It first
executes `INSERT .. ON CONFLICT` (called "upserting") to insert and update new rows, followed by
`DELETE .. WHERE` that removes old entries that didn't appear in the input query.
Returns a tuple with a number of upserts (inserts + updates) and a number of deletions.
"""
def call(repo, source, target, opts \\ [])
def call(repo, source, target, opts) do
table =
case target do
name when is_binary(name) -> name
schema -> schema.__schema__(:source)
end
columns =
Keyword.get_lazy(opts, :columns, fn ->
target.__schema__(:fields)
end)
conflict_target =
Keyword.get_lazy(opts, :conflict_target, fn ->
target.__schema__(:primary_key)
end)
query =
case(source) do
"SELECT " <> _ -> source
%{select: select} when not is_nil(select) -> source
_ -> select(source, ^columns)
end
default_opts = [
on_conflict: :replace_all,
conflict_target: conflict_target
]
do_sync(repo, table, columns, query, Keyword.merge(default_opts, opts))
end
defp do_sync(repo, table, columns, query, opts) do
delete_query_sql = "id NOT IN (SELECT id FROM upserts)"
input_scope = Keyword.get(opts, :scope)
delete_scope = Keyword.get(opts, :delete_scope)
scoped_query = apply_query_scope(query, input_scope)
scoped_delete_query_sql =
apply_delete_sql_scope(delete_query_sql, delete_scope || input_scope)
columns_sql = list_to_sql(columns)
{scoped_query_sql, params} = query_to_sql(repo, scoped_query)
on_conflict =
parse_on_conflict(
Keyword.get(opts, :on_conflict),
columns,
Keyword.get(opts, :conflict_target)
)
sql =
"WITH upserts AS (" <>
"INSERT INTO #{table} (#{columns_sql}) (#{scoped_query_sql}) #{on_conflict} RETURNING id" <>
"), deletions AS (" <>
"DELETE FROM #{table} WHERE #{scoped_delete_query_sql} RETURNING id" <>
") SELECT " <> "(SELECT COUNT(id) FROM upserts), (SELECT COUNT(id) FROM deletions)"
%{rows: [[upserts, deletions]]} = repo.query!(sql, params)
{upserts, deletions}
end
defp apply_query_scope(query, nil), do: query
defp apply_query_scope(query = %{}, scope) when is_list(scope), do: where(query, ^scope)
defp apply_delete_sql_scope(delete_sql, nil), do: delete_sql
defp apply_delete_sql_scope(delete_sql, scope) when is_binary(scope) do
delete_sql <> " AND #{scope}"
end
defp apply_delete_sql_scope(delete_sql, scope) when is_list(scope) do
delete_sql <> Enum.map_join(scope, fn {col, val} -> " AND #{col} = #{val}" end)
end
defp parse_on_conflict(nil, _, _), do: nil
defp parse_on_conflict(:replace_all, columns, conflict_target) do
setters = Enum.map(columns, fn col -> "#{col} = excluded.#{col}" end)
"ON CONFLICT (#{list_to_sql(conflict_target)}) DO UPDATE SET #{list_to_sql(setters)}"
end
defp query_to_sql(_repo, sql) when is_binary(sql), do: {sql, []}
defp query_to_sql(repo, query) do
SQL.to_sql(:all, repo, query)
end
defp list_to_sql(list), do: Enum.join(list, ", ")
end
_ ->
nil
end