Packages
ash
3.0.0-rc.35
3.29.3
3.29.2
3.29.1
3.29.0
3.28.0
3.27.8
3.27.7
3.27.6
3.27.5
3.27.4
3.27.3
3.27.2
3.27.1
3.27.0
3.26.0
3.25.2
3.25.1
3.25.0
retired
3.24.7
3.24.6
3.24.5
3.24.4
3.24.3
3.24.2
3.24.1
3.24.0
3.23.1
3.23.0
3.22.2
3.22.1
3.22.0
3.21.3
3.21.2
3.21.1
3.21.0
3.20.0
3.19.3
3.19.2
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.0
3.15.0
3.14.1
3.14.0
retired
3.13.2
3.13.1
3.13.0
3.12.0
3.11.3
3.11.2
3.11.1
3.11.0
3.10.1
3.10.0
3.9.0
3.8.0
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
retired
3.6.3
retired
3.6.2
3.6.1
3.6.0
3.5.43
3.5.42
3.5.41
3.5.40
3.5.39
3.5.38
3.5.37
3.5.36
3.5.35
3.5.34
3.5.33
3.5.32
3.5.31
3.5.30
3.5.29
3.5.28
3.5.27
3.5.26
3.5.25
3.5.24
3.5.23
3.5.22
3.5.21
3.5.20
3.5.19
3.5.18
3.5.17
3.5.16
3.5.15
3.5.14
3.5.13
3.5.12
3.5.11
3.5.10
3.5.9
3.5.8
3.5.7
3.5.6
3.5.5
3.5.4
3.5.3
3.5.2
3.5.1
3.5.0
3.4.74
retired
3.4.73
3.4.72
3.4.71
3.4.70
3.4.69
3.4.68
3.4.67
3.4.66
3.4.65
3.4.64
3.4.63
3.4.62
3.4.61
3.4.60
3.4.59
3.4.58
3.4.57
3.4.56
3.4.55
3.4.54
3.4.53
3.4.52
3.4.51
3.4.50
3.4.49
3.4.48
3.4.47
3.4.46
3.4.45
3.4.44
3.4.43
3.4.42
3.4.41
3.4.40
3.4.39
3.4.38
3.4.37
3.4.36
3.4.35
3.4.34
3.4.33
3.4.32
3.4.31
3.4.30
3.4.29
3.4.28
3.4.27
3.4.26
3.4.25
3.4.24
3.4.23
3.4.22
3.4.21
3.4.20
3.4.19
3.4.18
3.4.17
3.4.16
3.4.15
3.4.14
3.4.13
3.4.12
3.4.11
3.4.10
3.4.9
3.4.8
3.4.7
3.4.6
3.4.5
3.4.4
3.4.3
3.4.2
3.4.1
3.4.0
3.3.3
3.3.2
3.3.1
3.3.0
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.16
3.0.15
3.0.14
3.0.13
3.0.12
3.0.11
3.0.10
3.0.9
3.0.8
3.0.7
3.0.6
3.0.5
3.0.4
3.0.3
3.0.2
3.0.1
3.0.0
3.0.0-rc.46
3.0.0-rc.45
3.0.0-rc.44
3.0.0-rc.43
3.0.0-rc.42
3.0.0-rc.41
3.0.0-rc.40
3.0.0-rc.39
3.0.0-rc.38
3.0.0-rc.37
3.0.0-rc.36
3.0.0-rc.35
3.0.0-rc.34
3.0.0-rc.33
3.0.0-rc.32
3.0.0-rc.31
3.0.0-rc.29
3.0.0-rc.28
3.0.0-rc.27
3.0.0-rc.26
3.0.0-rc.25
3.0.0-rc.24
3.0.0-rc.23
3.0.0-rc.22
3.0.0-rc.21
3.0.0-rc.20
3.0.0-rc.19
3.0.0-rc.18
3.0.0-rc.17
3.0.0-rc.16
3.0.0-rc.15
3.0.0-rc.14
3.0.0-rc.13
3.0.0-rc.12
3.0.0-rc.11
3.0.0-rc.10
3.0.0-rc.9
3.0.0-rc.8
3.0.0-rc.7
3.0.0-rc.6
3.0.0-rc.5
3.0.0-rc.4
3.0.0-rc.3
3.0.0-rc.1
3.0.0-rc.0
2.21.15
2.21.14
2.21.13
2.21.12
2.21.11
2.21.10
2.21.9
2.21.8
2.21.7
2.21.6
2.21.5
2.21.4
2.21.3
2.21.2
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.14
2.19.13
2.19.12
2.19.11
2.19.10
2.19.9
2.19.8
2.19.7
2.19.6
2.19.5
2.19.4
2.19.3
retired
2.19.2
retired
2.19.1
retired
2.19.0
retired
2.18.2
2.18.1
2.18.0
2.17.24
2.17.23
2.17.22
2.17.21
2.17.20
2.17.19
2.17.18
2.17.17
2.17.16
2.17.15
2.17.14
2.17.13
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.1
2.16.0
2.15.20
2.15.19
2.15.18
2.15.17
2.15.16
2.15.15
2.15.14
2.15.13
2.15.12
2.15.11
2.15.10
2.15.9
2.15.8
2.15.7
2.15.6
2.15.5
2.15.4
2.15.2
2.15.1
2.15.0
2.14.21
2.14.20
2.14.19
2.14.18
2.14.17
2.14.16
2.14.15
2.14.14
2.14.13
2.14.12
2.14.11
2.14.10
2.14.9
2.14.8
2.14.7
2.14.6
2.14.5
2.14.4
2.14.3
2.14.2
2.14.1
2.14.0
2.13.4
retired
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.11
2.11.10
2.11.9
2.11.8
2.11.7
2.11.6
2.11.5
2.11.4
2.11.3
2.11.2
2.11.1
2.11.0
2.11.0-rc.3
2.11.0-rc.2
2.11.0-rc.1
2.11.0-rc.0
2.10.2
2.10.1
2.10.0
2.9.29
2.9.28
2.9.27
2.9.26
2.9.25
2.9.24
2.9.23
2.9.22
2.9.21
2.9.20
2.9.19
2.9.18
2.9.17
2.9.16
2.9.15
2.9.14
2.9.13
2.9.12
2.9.11
2.9.10
2.9.9
2.9.8
2.9.7
2.9.6
2.9.5
2.9.4
2.9.3
2.9.2
2.9.1
2.9.0
2.8.1
2.8.0
2.7.1
2.7.0
2.6.31
2.6.30
2.6.29
2.6.28
2.6.27
2.6.26
2.6.25
2.6.24
2.6.23
2.6.22
2.6.21
2.6.20
2.6.19
2.6.18
2.6.17
2.6.16
2.6.15
2.6.14
2.6.13
2.6.11
2.6.10
2.6.9
2.6.8
2.6.7
2.6.6
2.6.5
2.6.4
2.6.3
2.6.2
2.6.1
2.6.0
2.5.16
2.5.15
2.5.14
2.5.13
2.5.12
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.5.0-rc.6
2.5.0-rc.5
2.5.0-rc.4
2.5.0-rc.3
2.5.0-rc.2
2.5.0-rc.1
2.5.0-rc.0
2.4.30
2.4.29
2.4.28
2.4.27
2.4.26
2.4.25
2.4.24
2.4.23
2.4.22
2.4.21
2.4.20
2.4.19
2.4.18
2.4.17
2.4.16
2.4.15
2.4.14
2.4.13
2.4.12
2.4.11
2.4.10
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.15
2.0.0-rc.14
2.0.0-rc.13
2.0.0-rc.12
2.0.0-rc.11
2.0.0-rc.10
2.0.0-rc.9
2.0.0-rc.8
2.0.0-rc.7
2.0.0-rc.6
2.0.0-rc.5
2.0.0-rc.4
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
2.0.0-pre.8
2.0.0-pre.7
2.0.0-pre.6
2.0.0-pre.5
2.0.0-pre.4
2.0.0-pre.3
2.0.0-pre.2
2.0.0-pre.1
2.0.0-pre.0
1.53.3
1.53.2
1.53.0
1.52.0-rc.22
1.52.0-rc.21
1.52.0-rc.20
1.52.0-rc.19
1.52.0-rc.18
1.52.0-rc.17
1.52.0-rc.16
1.52.0-rc.15
1.52.0-rc.14
1.52.0-rc.13
1.52.0-rc.12
1.52.0-rc.11
1.52.0-rc.10
1.52.0-rc.9
1.52.0-rc.8
1.52.0-rc.7
1.52.0-rc.6
1.52.0-rc.5
1.52.0-rc.4
1.52.0-rc.3
1.52.0-rc.2
1.52.0-rc.1
1.52.0-rc.0
1.51.2
1.51.1
retired
1.51.0
1.50.21
1.50.20
1.50.19
1.50.18
1.50.17
1.50.16
1.50.15
1.50.14
1.50.13
1.50.12
1.50.11
1.50.10
1.50.9
1.50.8
1.50.7
1.50.6
1.50.5
1.50.4
1.50.3
1.50.2
1.50.1
1.50.0
1.49.0
1.48.0-rc.30
1.48.0-rc.29
1.48.0-rc.28
1.48.0-rc.27
1.48.0-rc.26
1.48.0-rc.25
1.48.0-rc.24
1.48.0-rc.23
1.48.0-rc.22
1.48.0-rc.21
1.48.0-rc.20
1.48.0-rc.19
1.48.0-rc.18
1.48.0-rc.17
1.48.0-rc.16
1.48.0-rc.15
1.48.0-rc.14
1.48.0-rc.13
1.48.0-rc.12
1.48.0-rc.11
1.48.0-rc.10
1.48.0-rc.9
1.48.0-rc.8
1.48.0-rc.7
1.48.0-rc.6
1.48.0-rc.5
1.48.0-rc.4
1.48.0-rc.3
1.48.0-rc.2
1.48.0-rc.1
1.48.0-rc.0
1.47.12
1.47.11
1.47.10
1.47.9
1.47.8
1.47.7
1.47.6
1.47.5
1.47.4
1.47.3
1.47.2
1.47.1
1.47.0
1.46.13
1.46.12
1.46.11
1.46.10
1.46.9
1.46.8
1.46.7
1.46.6
1.46.5
1.46.4
1.46.3
1.46.2
1.46.1
1.46.0
1.45.0-rc9
1.45.0-rc8
1.45.0-rc7
1.45.0-rc6
1.45.0-rc5
1.45.0-rc4
1.45.0-rc3
1.45.0-rc20
1.45.0-rc2
1.45.0-rc19
1.45.0-rc18
1.45.0-rc17
1.45.0-rc16
1.45.0-rc15
1.45.0-rc14
1.45.0-rc13
1.45.0-rc12
1.45.0-rc11
1.45.0-rc10
1.45.0-rc1
1.45.0-rc0
1.44.13
1.44.12
1.44.11
1.44.10
1.44.9
1.44.8
1.44.7
1.44.6
1.44.5
1.44.4
1.44.3
1.44.2
1.44.1
1.44.0
1.43.12
1.43.11
1.43.10
1.43.9
1.43.8
1.43.7
1.43.6
1.43.5
1.43.4
1.43.3
1.43.2
1.43.1
1.43.0
1.42.0
1.41.12
1.41.11
1.41.10
1.41.9
1.41.8
1.41.7
1.41.6
1.41.5
1.41.4
1.41.3
1.41.2
1.41.1
1.41.0
1.40.0
1.39.7
1.39.6
1.39.5
1.39.4
1.39.3
1.39.2
1.39.1
1.39.0
1.38.0
1.37.2
1.37.1
1.37.0
1.36.22
1.36.21
1.36.19
1.36.18
1.36.17
1.36.16
1.36.15
1.36.14
1.36.13
1.36.12
1.36.11
1.36.10
1.36.9
1.36.8
1.36.7
1.36.6
1.36.5
1.36.4
1.36.3
1.36.2
1.36.0
1.35.1
1.35.0
1.34.9
1.34.8
1.34.7
1.34.6
1.34.5
1.34.4
1.34.3
1.34.2
1.34.1
1.34.0
1.33.0
1.32.2
1.32.1
1.32.0
1.31.1
1.31.0
1.30.2
1.30.1
1.29.0-rc1
1.29.0-rc0
1.28.1
1.28.0
1.27.1
1.27.0
1.26.13
1.26.12
1.26.11
1.26.10
1.26.9
1.26.8
1.26.7
1.26.6
1.26.5
1.26.4
1.26.2
1.26.1
1.26.0
1.25.8
1.25.7
1.25.6
1.25.5
1.25.4
1.25.3
1.25.2
1.25.1
1.25.0
1.24.2
1.24.1
1.24.0
1.23.3
1.23.2
1.23.1
1.23.0
1.22.1
1.22.0
1.20.1
1.20.0
1.19.1
1.19.0
1.18.1
1.18.0
1.17.1
1.17.0
1.16.2
1.15.1
1.15.0
1.14.0
1.13.4
1.13.3
1.13.2
1.13.1
1.13.0
1.12.0
1.11.1
1.11.0
1.10.0
1.9.0
1.8.0
1.7.0
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.1
1.4.0
1.3.1
1.3.0
1.2.1
1.2.0
1.1.3
1.1.2
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
0.13.1
0.13.0
0.12.0
0.10.0
0.9.1
0.9.0
0.8.0
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.0
0.3.0
0.2.0
0.1.9
0.1.8
0.1.3
0.1.1
0.1.0
A declarative, extensible framework for building Elixir applications.
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/ash/actions/create/bulk.ex
defmodule Ash.Actions.Create.Bulk do
@moduledoc false
@spec run(Ash.Domain.t(), Ash.Resource.t(), atom(), Enumerable.t(map), Keyword.t()) ::
:ok
| {:ok, [Ash.Resource.record()]}
| {:ok, [Ash.Resource.record()], [Ash.Notifier.Notification.t()]}
| {:error, term}
def run(domain, resource, action, inputs, opts) do
action = Ash.Resource.Info.action(resource, action)
if !action do
raise Ash.Error.Invalid.NoSuchAction, resource: resource, action: action, type: :create
end
if opts[:transaction] == :all && opts[:return_stream?] do
raise ArgumentError,
"Cannot specify `transaction: :all` and `return_stream?: true` together"
end
if opts[:return_stream?] && opts[:sorted?] do
raise ArgumentError, "Cannot specify `sorted?: true` and `return_stream?: true` together"
end
opts =
Keyword.put_new_lazy(opts, :select, fn ->
resource |> Ash.Resource.Info.attributes() |> Enum.map(& &1.name)
end)
if opts[:transaction] == :all &&
Ash.DataLayer.data_layer_can?(resource, :transact) do
notify? =
if opts[:notify?] do
if Process.get(:ash_started_transaction?) do
false
else
Process.put(:ash_started_transaction?, true)
true
end
else
false
end
Ash.DataLayer.transaction(
List.wrap(resource) ++ action.touches_resources,
fn ->
do_run(domain, resource, action, inputs, opts)
end,
opts[:timeout],
%{
type: :bulk_create,
metadata: %{
resource: resource,
action: action.name,
actor: opts[:actor]
},
data_layer_context: opts[:data_layer_context] || %{}
}
)
|> case do
{:ok, bulk_result} ->
bulk_result =
if notify? do
%{
bulk_result
| notifications:
bulk_result.notifications ++ Process.delete(:ash_notifications) || []
}
else
bulk_result
end
handle_bulk_result(bulk_result, resource, action, opts)
{:error, error} ->
{:error, error}
end
else
domain
|> do_run(resource, action, inputs, opts)
|> handle_bulk_result(resource, action, opts)
end
end
def do_run(domain, resource, action, inputs, opts) do
opts = Ash.Actions.Helpers.set_opts(opts, domain)
upsert? = opts[:upsert?] || action.upsert?
upsert_fields = opts[:upsert_fields] || action.upsert_fields
if upsert? && !upsert_fields do
raise ArgumentError,
"For bulk actions, `upsert_fields` must be specified if upsert? is set to true`"
end
{_, opts} =
Ash.Actions.Helpers.set_context_and_get_opts(domain, Ash.Changeset.new(resource), opts)
manual_action_can_bulk? =
case action.manual do
{mod, _opts} ->
function_exported?(mod, :bulk_create, 3)
_ ->
false
end
data_layer_can_bulk? = Ash.DataLayer.data_layer_can?(resource, :bulk_create)
batch_size =
if data_layer_can_bulk? || manual_action_can_bulk? do
opts[:batch_size] || 100
else
1
end
ref = make_ref()
lazy_matching_default_values = lazy_matching_default_values(resource)
base_changeset = base_changeset(resource, domain, opts, action)
all_changes =
pre_template_all_changes(action, resource, action.type, base_changeset, opts[:actor])
argument_names = Enum.map(action.arguments, & &1.name)
changeset_stream =
inputs
|> Stream.with_index()
|> Stream.chunk_every(batch_size)
|> map_batches(
resource,
opts,
ref,
fn batch ->
try do
batch
|> Enum.map(
&setup_changeset(
&1,
action,
opts,
lazy_matching_default_values,
base_changeset,
argument_names
)
)
|> reject_and_maybe_store_errors(ref, opts)
|> handle_batch(
domain,
resource,
action,
all_changes,
data_layer_can_bulk?,
opts,
ref
)
after
if opts[:notify?] && !opts[:return_notifications?] do
Ash.Notifier.notify(Process.delete({:bulk_create_notifications, ref}))
end
end
end
)
if opts[:return_stream?] do
Stream.concat(changeset_stream)
else
try do
records =
if opts[:return_records?] do
Enum.to_list(Stream.concat(changeset_stream))
else
Stream.run(changeset_stream)
[]
end
notifications =
if opts[:notify?] && opts[:return_notifications?] do
Process.delete({:bulk_create_notifications, ref})
else
if opts[:notify?] do
Ash.Notifier.notify(Process.delete({:bulk_create_notifications, ref}))
end
[]
end
{errors, error_count} = Process.get({:bulk_create_errors, ref}) || {[], 0}
bulk_result = %Ash.BulkResult{
records: records,
errors: errors,
notifications: notifications,
error_count: error_count
}
case bulk_result do
%{records: _, error_count: 0} -> %{bulk_result | status: :success}
%{records: [], error_count: _} -> %{bulk_result | status: :error}
_ -> %{bulk_result | status: :partial_success}
end
catch
{:error, error, batch_number} ->
status =
if batch_number > 1 do
:partial_success
else
:error
end
result = %Ash.BulkResult{
status: status,
notifications: Process.delete({:bulk_create_notifications, ref})
}
{error_count, errors} = errors(result, error, opts)
%{result | errors: errors, error_count: error_count}
after
Process.delete({:bulk_create_errors, ref})
Process.delete({:bulk_create_notifications, ref})
end
end
end
defp pre_template_all_changes(action, resource, :create, base, actor) do
action.changes
|> Enum.concat(Ash.Resource.Info.validations(resource, action.type))
|> Enum.concat(Ash.Resource.Info.changes(resource, action.type))
|> Enum.map(fn
%{change: {module, opts}} = change ->
%{change | change: {module, pre_template(opts, base, actor)}}
%{validation: {module, opts}} = validation ->
%{validation | validation: {module, pre_template(opts, base, actor)}}
end)
|> Enum.map(fn
%{where: where} = change ->
new_where =
if where do
where
|> List.wrap()
|> Enum.map(fn {module, opts} -> {module, pre_template(opts, base, actor)} end)
end
%{change | where: new_where}
other ->
other
end)
|> Enum.with_index()
end
defp pre_template(opts, changeset, actor) do
if Ash.Expr.template_references_argument?(opts) ||
Ash.Expr.template_references_context?(opts) do
opts
else
{:templated,
Ash.Expr.fill_template(
opts,
actor,
%{},
changeset.context
)}
end
end
defp base_changeset(resource, domain, opts, action) do
resource
|> Ash.Changeset.new()
|> Map.put(:domain, domain)
|> Map.put(:context, %{
private: %{
upsert?: opts[:upsert?] || action.upsert? || false,
upsert_identity: opts[:upsert_identity] || action.upsert_identity,
upsert_fields:
Ash.Changeset.expand_upsert_fields(
opts[:upsert_fields] || action.upsert_fields,
resource
)
}
})
|> Ash.Actions.Helpers.add_context(opts)
|> Ash.Changeset.set_context(opts[:context] || %{})
|> Ash.Changeset.prepare_changeset_for_action(action, opts)
end
defp lazy_matching_default_values(resource) do
resource
|> Ash.Resource.Info.lazy_matching_default_attributes(:create)
|> Enum.group_by(& &1.default)
|> Enum.reduce(%{}, fn {fun, attributes}, lazy_matching_default_values ->
default =
case fun do
{m, f, a} -> apply(m, f, a)
fun -> fun.()
end
Enum.reduce(attributes, lazy_matching_default_values, fn attribute,
lazy_matching_default_values ->
Map.put(lazy_matching_default_values, attribute.name, default)
end)
end)
end
defp error_stream(ref) do
Stream.resource(
fn -> Process.delete({:bulk_create_errors, ref}) end,
fn
{errors, _count} ->
{Stream.map(errors || [], &{:error, &1}), []}
_ ->
{:halt, []}
end,
fn _ -> :ok end
)
end
defp notification_stream(ref) do
Stream.resource(
fn -> Process.delete({:bulk_create_notifications, ref}) end,
fn
[] ->
{:halt, []}
notifications ->
{Stream.map(notifications || [], &{:notification, &1}), []}
end,
fn _ -> :ok end
)
end
defp handle_batch(
batch,
domain,
resource,
action,
all_changes,
data_layer_can_bulk?,
opts,
ref
) do
if opts[:transaction] == :batch &&
Ash.DataLayer.data_layer_can?(resource, :transact) do
context = batch |> Enum.at(0) |> Kernel.||(%{}) |> Map.get(:context)
notify? =
if opts[:notify?] do
if Process.get(:ash_started_transaction?) do
false
else
Process.put(:ash_started_transaction?, true)
true
end
else
false
end
try do
Ash.DataLayer.transaction(
List.wrap(resource) ++ action.touches_resources,
fn ->
do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
data_layer_can_bulk?,
ref
)
end,
opts[:timeout],
%{
type: :bulk_create,
metadata: %{
resource: resource,
action: action.name,
actor: opts[:actor]
},
data_layer_context: opts[:data_layer_context] || context
}
)
|> case do
{:ok, result} ->
result
{:error, error} ->
store_error(ref, error, opts)
[]
end
after
if notify? do
notifications = Process.get(:ash_notifications, [])
remaining_notifications = Ash.Notifier.notify(notifications)
Process.delete(:ash_notifications) || []
Ash.Actions.Helpers.warn_missed!(resource, action, %{
resource_notifications: remaining_notifications
})
end
end
else
do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
data_layer_can_bulk?,
ref
)
end
end
defp do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
data_layer_can_bulk?,
ref
) do
must_return_records? =
opts[:notify?] ||
Enum.any?(batch, fn item ->
item.after_action != []
end)
%{
must_return_records?: must_return_records_for_changes?,
batch: batch,
changes: changes
} =
run_action_changes(
batch,
all_changes,
action,
opts[:actor],
opts[:authorize?],
opts[:tracer],
opts[:tenant]
)
batch =
batch
|> authorize(opts)
|> run_bulk_before_batches(
changes,
all_changes,
opts,
ref
)
changesets_by_index = index_changesets(batch)
run_batch(
resource,
batch,
action,
opts,
must_return_records?,
must_return_records_for_changes?,
data_layer_can_bulk?,
domain,
ref
)
|> run_after_action_hooks(opts, domain, ref, changesets_by_index)
|> process_results(
changes,
all_changes,
opts,
ref,
changesets_by_index,
batch,
domain,
resource
)
|> then(fn stream ->
if opts[:return_stream?] do
stream
|> Stream.map(&{:ok, &1})
|> Stream.concat(error_stream(ref))
|> Stream.concat(notification_stream(ref))
else
stream
end
end)
end
defp setup_changeset(
{input, index},
action,
opts,
lazy_matching_default_values,
base,
argument_names
) do
base
|> Ash.Changeset.put_context(:bulk_create, %{index: index})
|> handle_params(
Keyword.get(opts, :assume_casted?, false),
action,
opts,
input,
argument_names
)
|> set_lazy_non_matching_defaults()
|> set_lazy_matching_defaults(lazy_matching_default_values)
|> set_tenant()
end
defp set_tenant(changeset) do
if changeset.tenant &&
Ash.Resource.Info.multitenancy_strategy(changeset.resource) == :attribute do
attribute = Ash.Resource.Info.multitenancy_attribute(changeset.resource)
{m, f, a} = Ash.Resource.Info.multitenancy_parse_attribute(changeset.resource)
attribute_value = apply(m, f, [changeset.to_tenant | a])
Ash.Changeset.force_change_attribute(changeset, attribute, attribute_value)
else
changeset
end
end
defp handle_params(changeset, false, action, opts, input, _argument_names) do
Ash.Changeset.handle_params(changeset, action, input, opts)
end
defp handle_params(changeset, true, action, opts, input, argument_names) do
{args, attrs} =
Map.split(input, argument_names)
%{changeset | arguments: args, attributes: attrs}
|> Ash.Changeset.handle_params(action, input, Keyword.put(opts, :cast_params?, false))
end
defp map_batches(stream, resource, opts, ref, callback) do
max_concurrency = opts[:max_concurrency]
max_concurrency =
if max_concurrency && max_concurrency > 1 && Ash.DataLayer.can?(:async_engine, resource) do
max_concurrency
else
0
end
if max_concurrency && max_concurrency > 1 do
Task.async_stream(
stream,
fn batch ->
try do
Process.put(:ash_started_transaction?, true)
batch_result = callback.(batch)
{errors, _} = Process.get({:bulk_create_errors, ref}) || {[], 0}
notifications =
if opts[:notify?] do
process_notifications = Process.get(:ash_notifications, [])
bulk_notifications = Process.get({:bulk_create_notifications, ref}) || []
if opts[:return_notifications?] do
process_notifications ++ bulk_notifications
else
if opts[:transaction] && opts[:transaction] != :all do
Ash.Notifier.notify(bulk_notifications)
Ash.Notifier.notify(process_notifications)
end
[]
end
end
{batch_result, notifications, errors}
catch
value ->
{:throw, value}
end
end,
timeout: :infinity,
max_concurrency: max_concurrency
)
|> Stream.map(fn
{:ok, {:throw, value}} ->
throw(value)
{:ok, {result, notifications, errors}} ->
store_notification(ref, notifications, opts)
store_error(ref, errors, opts)
result
{:exit, error} ->
store_error(ref, error, opts)
[]
end)
else
Stream.map(stream, callback)
end
end
defp set_lazy_non_matching_defaults(changeset) do
changeset.resource
|> Ash.Resource.Info.lazy_non_matching_default_attributes(:create)
|> Enum.reduce(changeset, fn attribute, changeset ->
if Ash.Changeset.changing_attribute?(changeset, attribute.name) do
changeset
else
Ash.Changeset.force_change_attribute(
changeset,
attribute.name,
default(attribute)
)
end
end)
end
defp set_lazy_matching_defaults(changeset, values) do
Enum.reduce(values, changeset, fn {key, value}, changeset ->
if Ash.Changeset.changing_attribute?(changeset, key) do
changeset
else
Ash.Changeset.force_change_attribute(
changeset,
key,
value
)
end
end)
end
defp index_changesets(batch) do
Enum.reduce(batch, %{}, fn changeset, changesets_by_index ->
Map.put(
changesets_by_index,
changeset.context.bulk_create.index,
changeset
)
end)
end
defp default(%{default: {mod, func, args}}), do: apply(mod, func, args)
defp default(%{default: function}) when is_function(function, 0), do: function.()
defp default(%{default: value}), do: value
defp errors(result, invalid, opts) when is_list(invalid) do
Enum.reduce(invalid, {result.error_count, result.errors}, fn invalid, {error_count, errors} ->
errors(%{result | error_count: error_count, errors: errors}, invalid, opts)
end)
end
defp errors(result, nil, _opts) do
{result.error_count + 1, []}
end
defp errors(result, {:error, error}, opts) do
if opts[:return_errors?] do
{result.error_count + 1, [error | result.errors]}
else
{result.error_count + 1, []}
end
end
defp errors(result, invalid, opts) do
if Enumerable.impl_for(invalid) do
invalid = Enum.to_list(invalid)
errors(result, invalid, opts)
else
errors(result, {:error, invalid}, opts)
end
end
defp run_bulk_before_batches(
batch,
changes,
all_changes,
opts,
ref
) do
all_changes
|> Enum.filter(fn
{%{change: {module, _opts}}, _} ->
function_exported?(module, :batch_change, 3) &&
function_exported?(module, :before_batch, 3)
_ ->
false
end)
|> Enum.reduce(batch, fn {%{change: {module, change_opts}}, index}, batch ->
if changes[index] == :all do
module.before_batch(
batch,
change_opts,
struct(Ash.Resource.Change.Context, %{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
})
)
else
{matches, non_matches} =
batch
|> Enum.split_with(fn
%{valid?: false} ->
false
changeset ->
changeset.context.bulk_create.index in List.wrap(changes[index])
end)
before_batch_results =
module.before_batch(
matches,
change_opts,
struct(Ash.Resource.Change.Context, %{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
})
)
Enum.concat([before_batch_results, non_matches])
end
end)
|> Enum.reject(fn
%Ash.Notifier.Notification{} = notification ->
store_notification(ref, notification, opts)
true
_changeset ->
false
end)
end
defp reject_and_maybe_store_errors(stream, ref, opts) do
Enum.reject(stream, fn changeset ->
if changeset.valid? do
false
else
store_error(ref, changeset, opts)
true
end
end)
end
defp store_error(_ref, empty, _opts) when empty in [[], nil], do: :ok
defp store_error(ref, error, opts) do
if opts[:stop_on_error?] && !opts[:return_stream?] do
throw({:error, Ash.Error.to_error_class(error), 0, []})
else
if opts[:return_errors?] do
{errors, count} = Process.get({:bulk_create_errors, ref}) || {[], 0}
error =
case error do
%Ash.Changeset{} = changeset ->
changeset
other ->
Ash.Error.to_ash_error(other)
end
Process.put(
{:bulk_create_errors, ref},
{[error | errors], count + 1}
)
else
{errors, count} = Process.get({:bulk_create_errors, ref}) || {[], 0}
Process.put({:bulk_create_errors, ref}, {errors, count + 1})
end
end
end
defp store_notification(_ref, empty, _opts) when empty in [[], nil], do: :ok
defp store_notification(ref, notification, opts) do
if opts[:notify?] || opts[:return_notifications?] do
notifications = Process.get({:bulk_create_notifications, ref}) || []
new_notifications =
if is_list(notification) do
notification ++ notifications
else
[notification | notifications]
end
Process.put({:bulk_create_notifications, ref}, new_notifications)
end
end
defp authorize(batch, opts) do
if opts[:authorize?] do
Enum.map(batch, fn changeset ->
if changeset.valid? do
case Ash.can(changeset, opts[:actor],
return_forbidden_error?: true,
run_queries?: false,
maybe_is: false,
alter_source?: true
) do
{:ok, true} ->
changeset
{:ok, true, changeset} ->
changeset
{:ok, false, error} ->
Ash.Changeset.add_error(changeset, error)
{:error, error} ->
Ash.Changeset.add_error(changeset, error)
end
else
changeset
end
end)
else
batch
end
end
defp handle_bulk_result(%Ash.BulkResult{} = bulk_result, _resource, _action, opts) do
bulk_result
|> sort(opts)
|> ensure_records_return_type(opts)
|> ensure_errors_return_type(opts)
end
# for when we return a stream
defp handle_bulk_result(stream, _, _, _), do: stream
defp ensure_records_return_type(result, opts) do
if opts[:return_records?] do
%{result | records: result.records || []}
else
%{result | records: nil}
end
end
defp ensure_errors_return_type(result, opts) do
if opts[:return_errors?] do
%{result | errors: result.errors || []}
else
%{result | errors: nil}
end
end
defp sort(%{records: records} = result, opts) when is_list(records) do
if opts[:sorted?] do
%{result | records: Enum.sort_by(records, & &1.__metadata__.bulk_create_index)}
else
result
end
end
defp sort(result, _), do: result
defp run_batch(
resource,
batch,
action,
opts,
must_return_records?,
must_return_records_for_changes?,
data_layer_can_bulk?,
domain,
ref
) do
batch
|> Enum.map(fn changeset ->
if changeset.valid? do
{changeset, %{notifications: new_notifications}} =
Ash.Changeset.run_before_actions(changeset)
new_notifications = store_notification(ref, new_notifications, opts)
{changeset, manage_notifications} =
if changeset.valid? do
case Ash.Actions.ManagedRelationships.setup_managed_belongs_to_relationships(
changeset,
opts[:actor],
authorize?: opts[:authorize?],
actor: opts[:actor],
tenant: opts[:tenant]
) do
{:error, error} ->
{Ash.Changeset.add_error(changeset, error), new_notifications}
{changeset, manage_instructions} ->
{changeset, manage_instructions.notifications}
end
else
{changeset, []}
end
store_notification(ref, manage_notifications, opts)
changeset
else
changeset
end
end)
|> Enum.reject(fn
%{valid?: false} = changeset ->
store_error(ref, changeset, opts)
true
_changeset ->
false
end)
|> case do
[] ->
[]
batch ->
upsert_keys =
if opts[:upsert?] || action.upsert? do
case opts[:upsert_identity] || action.upsert_identity do
nil ->
Ash.Resource.Info.primary_key(resource)
identity ->
keys =
resource
|> Ash.Resource.Info.identities()
|> Enum.find(&(&1.name == identity))
|> Kernel.||(
raise Ash.Error.Invalid.NoIdentityFound,
resource: resource,
identity: identity
)
|> Map.get(:keys)
if opts[:tenant] &&
Ash.Resource.Info.multitenancy_strategy(resource) == :attribute do
[Ash.Resource.Info.multitenancy_attribute(resource) | keys]
else
keys
end
end
end
batch
|> Enum.group_by(&{&1.atomics, &1.filter})
|> Enum.flat_map(fn {_atomics, batch} ->
result =
case action.manual do
{mod, opts} ->
if function_exported?(mod, :bulk_create, 3) do
mod.bulk_create(batch, opts, %Ash.Resource.ManualCreate.Context{
actor: opts[:actor],
select: opts[:select],
batch_size: opts[:batch_size],
authorize?: opts[:authorize?],
tracer: opts[:tracer],
domain: domain,
upsert?: opts[:upsert?] || action.upsert?,
upsert_keys: upsert_keys,
upsert_fields:
Ash.Changeset.expand_upsert_fields(
opts[:upsert_fields] || action.upsert_fields,
resource
),
return_records?:
opts[:return_records?] || must_return_records? ||
must_return_records_for_changes?,
tenant: opts[:tenant]
})
|> Ash.Actions.Helpers.rollback_if_in_transaction(resource, nil)
|> case do
{:ok, results} ->
results
:ok ->
if opts[:return_records?] do
raise "`#{inspect(mod)}.bulk_create/3` returned :ok without a result when `return_records?` is true"
else
[]
end
{:error, error} ->
store_error(ref, error, opts)
[]
{:notifications, notifications} ->
store_notification(ref, notifications, opts)
[]
end
else
[changeset] = batch
result =
mod.create(changeset, opts, %Ash.Resource.ManualCreate.Context{
select: opts[:select],
actor: opts[:actor],
tenant: opts[:tenant],
authorize?: opts[:authorize?],
tracer: opts[:tracer],
domain: domain
})
|> Ash.Actions.Helpers.rollback_if_in_transaction(resource, nil)
case result do
{:ok, result} ->
{:ok,
[
Ash.Resource.put_metadata(
result,
:bulk_create_index,
changeset.context.bulk_create.index
)
]}
{:error, error} ->
{:error, error}
end
end
_ ->
if data_layer_can_bulk? do
Ash.DataLayer.bulk_create(
resource,
Stream.map(batch, fn changeset ->
%{changeset | tenant: changeset.to_tenant}
end),
%{
select: opts[:select],
batch_size: opts[:batch_size],
return_records?:
opts[:return_records?] || must_return_records? ||
must_return_records_for_changes?,
upsert?: opts[:upsert?] || action.upsert? || false,
upsert_keys: upsert_keys,
upsert_fields:
Ash.Changeset.expand_upsert_fields(
opts[:upsert_fields] || action.upsert_fields,
resource
),
tenant: Ash.ToTenant.to_tenant(opts[:tenant], resource)
}
)
|> Ash.Actions.Helpers.rollback_if_in_transaction(resource, nil)
else
[changeset] = batch
upsert? = opts[:upsert?] || action.upsert? || false
result =
if upsert? do
Ash.DataLayer.upsert(resource, changeset, upsert_keys)
else
Ash.DataLayer.create(resource, changeset)
end
case result do
{:ok, result} ->
{:ok,
[
Ash.Resource.put_metadata(
result,
:bulk_create_index,
changeset.context.bulk_create.index
)
]}
{:error, :no_rollback, error} ->
{:error, :no_rollback, error}
{:error, error} ->
{:error, error}
end
end
end
case result do
{:ok, result} ->
result
:ok ->
[]
{:error, :no_rollback, error} ->
store_error(ref, error, opts)
{:error, error} ->
store_error(ref, error, opts)
[]
end
end)
end
end
defp manage_relationships(created, domain, changeset, engine_opts) do
with {:ok, loaded} <-
Ash.Actions.ManagedRelationships.load(domain, created, changeset, engine_opts),
{:ok, with_relationships, new_notifications} <-
Ash.Actions.ManagedRelationships.manage_relationships(
loaded,
changeset,
engine_opts[:actor],
engine_opts
) do
{:ok, with_relationships, %{notifications: new_notifications, new_changeset: changeset}}
end
end
defp run_after_action_hooks(
batch_results,
opts,
domain,
ref,
changesets_by_index
) do
Enum.flat_map(batch_results, fn result ->
changeset = changesets_by_index[result.__metadata__.bulk_create_index]
case manage_relationships(result, domain, changeset,
actor: opts[:actor],
authorize?: opts[:authorize?]
) do
{:ok, result, %{notifications: new_notifications, new_changeset: changeset}} ->
store_notification(ref, new_notifications, opts)
case Ash.Changeset.run_after_actions(result, changeset, []) do
{:error, error} ->
store_error(ref, error, opts)
[]
{:ok, result, _changeset, %{notifications: more_new_notifications}} ->
store_notification(ref, more_new_notifications, opts)
[result]
end
{:error, error} ->
store_error(ref, error, opts)
[]
end
end)
end
defp process_results(
batch,
changes,
all_changes,
opts,
ref,
changesets_by_index,
changesets,
domain,
resource
) do
results =
Enum.flat_map(batch, fn result ->
changeset = changesets_by_index[result.__metadata__.bulk_create_index]
if opts[:notify?] || opts[:return_notifications?] do
store_notification(ref, notification(changeset, result, opts), opts)
end
try do
case Ash.Changeset.run_after_transactions(
{:ok, result},
changeset
) do
{:ok, result} ->
if opts[:return_records?] do
[result]
else
[]
end
{:error, error} ->
store_error(ref, error, opts)
[]
end
rescue
e ->
store_error(ref, e, opts)
[]
end
end)
changes
|> run_bulk_after_changes(all_changes, results, changesets_by_index, changesets, opts, ref)
|> then(fn records ->
select =
if opts[:select] do
List.wrap(opts[:select])
else
resource |> Ash.Resource.Info.public_attributes() |> Enum.map(& &1.name)
end
case Ash.load(
records,
select,
reuse_values?: true,
domain: domain,
actor: opts[:actor],
authorize?: opts[:authorize?],
tracer: opts[:tracer]
) do
{:ok, records} ->
Ash.load(
records,
List.wrap(opts[:load]),
domain: domain,
actor: opts[:actor],
authorize?: opts[:authorize?],
tracer: opts[:tracer]
)
|> case do
{:ok, records} ->
{:ok, Enum.reject(records, & &1.__metadata__[:private][:missing_from_data_layer])}
{:error, error} ->
{:error, error}
end
other ->
other
end
end)
|> case do
{:ok, records} ->
records
{:error, error} ->
store_error(ref, error, opts)
[]
end
end
defp run_bulk_after_changes(
changes,
all_changes,
results,
changesets_by_index,
changesets,
opts,
ref
) do
context =
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
all_changes
|> Enum.filter(fn
{%{change: {module, change_opts}}, _} ->
function_exported?(module, :batch_change, 3) &&
function_exported?(module, :after_batch, 3) &&
module.batch_callbacks?(changesets, change_opts, context)
_ ->
false
end)
|> Enum.reduce(results, fn {%{change: {module, change_opts}}, index}, results ->
records = results
if changes[index] == :all do
results =
Enum.map(results, fn result ->
{changesets_by_index[result.__metadata__.bulk_create_index], result}
end)
module.after_batch(
results,
change_opts,
context
)
|> handle_after_batch_results(records, ref, opts)
else
{matches, non_matches} =
results
|> Enum.split_with(fn
{:ok, result} ->
result.__metadata__.bulk_create_index in List.wrap(changes[index])
_ ->
false
end)
matches =
Enum.map(matches, fn match ->
{changesets_by_index[match.__metadata__.bulk_create_index], match}
end)
after_batch_results =
module.after_batch(
matches,
change_opts,
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
)
|> handle_after_batch_results(matches, ref, opts)
Enum.concat([after_batch_results, non_matches])
end
end)
end
defp handle_after_batch_results(:ok, matches, _, _), do: matches
defp handle_after_batch_results(results, _matches, ref, options) do
Enum.flat_map(
results,
fn
%Ash.Notifier.Notification{} = notification ->
store_notification(ref, notification, options)
{:ok, result} ->
[result]
{:error, error} ->
store_error(ref, error, options)
[]
end
)
end
defp notification(changeset, result, opts) do
%Ash.Notifier.Notification{
resource: changeset.resource,
domain: changeset.domain,
actor: opts[:actor],
action: changeset.action,
for: Ash.Resource.Info.notifiers(changeset.resource) ++ changeset.action.notifiers,
data: result,
changeset: changeset
}
end
defp templated_opts({:templated, opts}, _actor, _arguments, _context), do: opts
defp templated_opts(opts, actor, arguments, context) do
Ash.Expr.fill_template(
opts,
actor,
arguments,
context
)
end
defp run_action_changes(batch, all_changes, _action, actor, authorize?, tracer, tenant) do
context = %{
actor: actor,
authorize?: authorize? || false,
tracer: tracer,
tenant: tenant
}
Enum.reduce(
all_changes,
%{must_return_records?: false, batch: batch, changes: %{}},
fn
{%{validation: {module, opts}} = validation, _change_index}, %{batch: batch} = state ->
batch =
Enum.map(batch, fn changeset ->
cond do
validation.only_when_valid? && !changeset.valid? ->
changeset
Enum.all?(validation.where || [], fn {module, opts} ->
opts = templated_opts(opts, actor, changeset.arguments, changeset.context)
{:ok, opts} = module.init(opts)
module.validate(
changeset,
opts,
struct(Ash.Resource.Validation.Context, context)
) == :ok
end) ->
opts = templated_opts(opts, actor, changeset.arguments, changeset.context)
{:ok, opts} = module.init(opts)
case module.validate(
changeset,
opts,
struct(
Ash.Resource.Validation.Context,
Map.put(context, :message, validation.message)
)
) do
:ok ->
changeset
{:error, error} ->
Ash.Changeset.add_error(changeset, error)
end
true ->
changeset
end
end)
%{
state
| must_return_records?: state.must_return_records?,
batch: batch,
changes: state.changes
}
{%{change: {module, change_opts}} = change, change_index}, %{batch: batch} = state ->
# could track if any element in the batch is invalid, and if not use the fast version
if Enum.empty?(change.where) && !change.only_when_valid? do
batch = batch_change(module, batch, change_opts, context, actor)
must_return_records? =
state.must_return_records? ||
Enum.any?(batch, fn item ->
item.relationships not in [nil, %{}] || !Enum.empty?(item.after_action)
end) ||
(function_exported?(module, :batch_change, 3) &&
function_exported?(module, :after_batch, 3) &&
module.batch_callbacks?(batch, change_opts, context))
%{
state
| must_return_records?: must_return_records?,
batch: batch,
changes: Map.put(state.changes, change_index, :all)
}
else
{matches, non_matches} =
batch
|> Enum.split_with(fn changeset ->
applies_from_where? =
Enum.all?(change.where || [], fn {module, opts} ->
opts = templated_opts(opts, actor, changeset.arguments, changeset.context)
{:ok, opts} = module.init(opts)
module.validate(
changeset,
opts,
struct(Ash.Resource.Validation.Context, context)
) == :ok
end)
applies_from_only_when_valid? =
if change.only_when_valid? do
changeset.valid?
else
true
end
applies_from_where? and applies_from_only_when_valid?
end)
if Enum.empty?(matches) do
%{
state
| must_return_records?: state.must_return_records?,
batch: non_matches,
changes: state.changes
}
else
matches = batch_change(module, matches, change_opts, context, actor)
must_return_records? =
state.must_return_records? ||
Enum.any?(batch, fn item ->
item.relationships not in [nil, %{}] || !Enum.empty?(item.after_action)
end) ||
(function_exported?(module, :batch_change, 3) &&
function_exported?(module, :after_batch, 3) &&
module.batch_callbacks?(batch, change_opts, context))
%{
state
| must_return_records?: must_return_records?,
batch: Enum.concat(matches, non_matches),
changes:
Map.put(
state.changes,
change_index,
Enum.map(matches, & &1.context.bulk_create.index)
)
}
end
end
end
)
end
defp batch_change(module, batch, change_opts, context, actor) do
case change_opts do
{:templated, change_opts} ->
if function_exported?(module, :batch_change, 4) do
module.batch_change(
batch,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true),
actor
)
else
Enum.map(batch, fn changeset ->
{:ok, change_opts} = module.init(change_opts)
module.change(
changeset,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
end)
end
change_opts ->
Enum.map(batch, fn changeset ->
change_opts = templated_opts(change_opts, actor, changeset.arguments, changeset.context)
{:ok, change_opts} = module.init(change_opts)
module.change(
changeset,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
end)
end
end
end