Packages
ash
3.2.6
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_name, inputs, opts) do
action = Ash.Resource.Info.action(resource, action_name)
opts =
if opts[:return_notifications?] do
Keyword.put(opts, :notify?, true)
else
opts
end
if !action do
raise Ash.Error.Invalid.NoSuchAction, resource: resource, action: action_name, 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)
belongs_to_attrs =
resource
|> Ash.Resource.Info.relationships()
|> Enum.filter(&(&1.type == :belongs_to))
|> Enum.map(& &1.source_attribute)
required_attrs_list =
resource
|> Ash.Resource.Info.attributes()
|> Enum.reject(&(&1.allow_nil? || &1.generated? || &1.name in belongs_to_attrs))
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
)
)
|> handle_batch(
domain,
resource,
action,
all_changes,
data_layer_can_bulk?,
opts,
ref,
required_attrs_list
)
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}))
else
[]
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,
required_attrs_list
) do
%{
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, must_be_simple} =
batch
|> Stream.map(fn changeset ->
Ash.Changeset.require_values(
changeset,
:create,
true,
required_attrs_list
)
end)
|> Enum.reduce({[], []}, fn changeset, {batch, must_be_simple} ->
if changeset.after_transaction in [[], nil] do
changeset = Ash.Changeset.run_before_transaction_hooks(changeset)
{[changeset | batch], must_be_simple}
else
{batch, [%{changeset | __validated_for_action__: action.name} | must_be_simple]}
end
end)
must_be_simple_results =
Enum.flat_map(must_be_simple, fn changeset ->
case Ash.Actions.Create.run(domain, changeset, action, opts) do
{:ok, result} ->
[
Ash.Resource.set_metadata(result, %{
bulk_create_index: changeset.context.bulk_create.index
})
]
{:error, error} ->
store_error(ref, error, opts)
[]
end
end)
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 ->
tmp_ref = make_ref()
result =
do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
data_layer_can_bulk?,
ref,
changes,
must_return_records_for_changes?,
must_be_simple_results
)
{new_errors, new_error_count} =
Process.delete({:bulk_create_errors, tmp_ref}) || {[], 0}
store_error(ref, new_errors, new_error_count)
result
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,
changes,
must_return_records_for_changes?,
must_be_simple_results
)
end
end
defp do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
data_layer_can_bulk?,
ref,
changes,
must_return_records_for_changes?,
must_be_simple_results
) do
must_return_records? =
opts[:notify?] ||
Enum.any?(batch, fn item ->
item.after_action != []
end)
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
)
|> Stream.concat(must_be_simple_results)
|> 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)
else
[]
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}}, _} ->
module.has_batch_change?() &&
module.has_before_batch?()
_ ->
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 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}
new_errors =
error
|> List.wrap()
|> Enum.map(&Ash.Error.to_ash_error/1)
Process.put(
{:bulk_create_errors, ref},
{new_errors ++ errors, count + Enum.count(new_errors)}
)
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,
pre_flight?: 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)
changeset =
changeset
|> Ash.Changeset.hydrate_atomic_refs(opts[:actor])
|> Ash.Changeset.apply_atomic_constraints(opts[:actor])
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, _} =
Ash.Actions.ManagedRelationships.validate_required_belongs_to({changeset, []})
{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,
identity:
(opts[:upsert_identity] || action.upsert_identity) &&
Ash.Resource.Info.identity(
resource,
opts[:upsert_identity] || action.upsert_identity
),
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],
identity:
(opts[:upsert_identity] || action.upsert_identity) &&
Ash.Resource.Info.identity(
resource,
opts[:upsert_identity] || action.upsert_identity
),
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} ->
if opts[:transaction] && opts[:rollback_on_error?] do
if Ash.DataLayer.in_transaction?(changeset.resource) do
Ash.DataLayer.rollback(
changeset.resource,
error
)
end
end
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,
resource,
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,
tenant: opts[:tenant],
actor: opts[:actor],
authorize?: opts[:authorize?],
tracer: opts[:tracer]
) do
{:ok, records} ->
Ash.load(
records,
List.wrap(opts[:load]),
domain: domain,
tenant: opts[:tenant],
reuse_values?: true,
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,
resource,
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}}, _} ->
module.has_batch_change?() &&
module.has_after_batch?() &&
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, resource, 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, resource, opts)
Enum.concat([after_batch_results, non_matches])
end
end)
end
defp handle_after_batch_results(:ok, matches, _, _resource, _), do: matches
defp handle_after_batch_results(results, _matches, ref, resource, opts) do
Enum.flat_map(
results,
fn
%Ash.Notifier.Notification{} = notification ->
store_notification(ref, notification, opts)
{:ok, result} ->
[result]
{:error, error} ->
if opts[:transaction] && opts[:rollback_on_error?] do
if Ash.DataLayer.in_transaction?(resource) do
Ash.DataLayer.rollback(
resource,
error
)
end
end
store_error(ref, error, opts)
[]
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} ->
error = Ash.Error.to_ash_error(error)
if validation.message do
error =
Ash.Changeset.override_validation_message(error, validation.message)
Ash.Changeset.add_error(changeset, error)
else
Ash.Changeset.add_error(changeset, error)
end
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) ||
(module.has_batch_change?() &&
module.has_after_batch?() &&
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) ||
(module.has_batch_change?() &&
module.has_after_batch?() &&
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 module.has_batch_change?() do
module.batch_change(
batch,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
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