Packages
ash
3.6.3
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.
Retired package: Security issue - CVE-2025-48044
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/ash/actions/create/bulk.ex
# SPDX-FileCopyrightText: 2019 ash contributors <https://github.com/ash-project/ash/graphs.contributors>
#
# SPDX-License-Identifier: MIT
defmodule Ash.Actions.Create.Bulk do
alias Ash.Actions.BulkManualActionHelpers
@moduledoc false
@spec run(Ash.Domain.t(), Ash.Resource.t(), atom(), Enumerable.t(map), Keyword.t()) ::
Ash.BulkResult.t()
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? = opts[:notify?] && !Process.put(:ash_started_transaction?, true)
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] || %{}
},
rollback_on_error?: false
)
|> 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) | action: action},
opts
)
manual_action_can_bulk? =
case action.manual do
{mod, _opts} ->
function_exported?(mod, :bulk_create, 3)
_ ->
false
end
base_changeset = base_changeset(resource, domain, opts, action)
data_layer_can_bulk? =
if Ash.DataLayer.data_layer_can?(resource, :bulk_create) do
# If upserting with return_skipped_upsert? and data layer can't handle it in bulk, fall back to single inserts
!upsert? || !base_changeset.context[:private][:return_skipped_upsert?] ||
Ash.DataLayer.data_layer_can?(resource, :bulk_upsert_return_skipped)
else
false
end
batch_size =
cond do
action.manual != nil and manual_action_can_bulk? -> opts[:batch_size] || 100
action.manual == nil and data_layer_can_bulk? -> opts[:batch_size] || 100
true -> 1
end
all_changes =
pre_template_all_changes(
action,
resource,
action.type,
base_changeset,
opts[:actor],
base_changeset.to_tenant
)
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)
attrs_to_require =
resource
|> Ash.Resource.Info.attributes()
|> Enum.reject(&(&1.allow_nil? || &1.generated? || &1.name in belongs_to_attrs))
action_select =
if Ash.DataLayer.data_layer_can?(resource, :action_select) do
Enum.uniq(
Enum.concat(
Ash.Resource.Info.action_select(
resource,
action
),
List.wrap(
opts[:select] ||
MapSet.to_list(Ash.Resource.Info.selected_by_default_attribute_names(resource))
)
)
)
else
MapSet.to_list(Ash.Resource.Info.attribute_names(resource))
end
ref = make_ref()
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(resource),
base_changeset,
argument_names
)
)
|> handle_batch(
domain,
resource,
action,
all_changes,
data_layer_can_bulk?,
opts,
ref,
attrs_to_require,
action_select
)
after
if opts[:notify?] && !opts[:return_notifications?] do
Process.put(
{:bulk_notifications, ref},
Ash.Notifier.notify(Process.delete({:bulk_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)
nil
end
notifications =
if opts[:notify?] do
Process.delete({:bulk_notifications, ref}) || []
else
[]
end
# Unless return_notifications? is true, we always want to end up with nil.
notifications =
cond do
Enum.empty?(notifications) ->
if opts[:return_notifications?], do: [], else: nil
opts[:return_notifications?] ->
notifications
true ->
notifications =
if opts[:notify?] do
Ash.Notifier.notify(notifications)
else
notifications
end
if Process.get(:ash_started_transaction?) do
Process.put(
:ash_notifications,
Process.get(:ash_notifications, []) ++ notifications
)
end
nil
end
{errors, error_count} = Process.get({:bulk_errors, ref}) || {[], 0}
errors =
if opts[:return_errors?] do
Enum.map(errors, fn error ->
Ash.Error.to_ash_error(error, [],
bread_crumbs: [
"Exception raised in bulk create: #{inspect(resource)}.#{action.name}"
]
)
end)
else
nil
end
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}
_ ->
if Process.get({:any_success?, ref}) do
%{bulk_result | status: :partial_success}
else
%{bulk_result | status: :error}
end
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_notifications, ref})
}
{error_count, errors} = errors(result, error, opts)
%{result | errors: errors, error_count: error_count}
after
Process.delete({:bulk_errors, ref})
Process.delete({:bulk_notifications, ref})
end
end
rescue
e ->
reraise Ash.Error.to_error_class(e,
stacktrace: __STACKTRACE__,
bread_crumbs: [
"Exception raised in bulk create: #{inspect(resource)}.#{action.name}"
]
),
__STACKTRACE__
end
defp pre_template_all_changes(action, resource, :create, base, actor, tenant) do
action.changes
|> then(fn changes ->
if action.skip_global_validations? do
changes
else
Enum.concat(changes, Ash.Resource.Info.validations(resource, action.type))
end
end)
|> Enum.concat(Ash.Resource.Info.changes(resource, action.type))
|> Enum.map(fn
%{change: {module, opts}} = change ->
%{change | change: {module, pre_template(opts, base, actor, tenant)}}
%{validation: {module, opts}} = validation ->
%{validation | validation: {module, pre_template(opts, base, actor, tenant)}}
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, tenant)}
end)
end
%{change | where: new_where}
other ->
other
end)
|> Enum.with_index()
end
defp pre_template(opts, changeset, actor, tenant) do
if Ash.Expr.template_references_argument?(opts) ||
Ash.Expr.template_references_context?(opts) do
opts
else
{:templated,
Ash.Expr.fill_template(
opts,
actor: actor,
tenant: tenant,
args: %{},
context: changeset.context,
changeset: changeset
)}
end
end
defp base_changeset(resource, domain, opts, action) do
upsert_condition =
case opts[:upsert_condition] do
nil -> action && action.upsert_condition
other -> other
end
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
),
upsert_condition: upsert_condition,
return_skipped_upsert?:
opts[:return_skipped_upsert?] || (action && action.return_skipped_upsert?) || false
}
})
|> Ash.Actions.Helpers.add_context(opts)
|> Ash.Changeset.set_context(opts[:context] || %{})
|> Ash.Changeset.prepare_changeset_for_action(action, opts)
|> Ash.Changeset.set_private_arguments_for_action(opts[:private_arguments] || %{})
|> then(fn changeset ->
if opts[:after_action] do
Ash.Changeset.after_action(changeset, opts[:after_action])
else
changeset
end
end)
|> then(fn
changeset when upsert_condition != nil -> Ash.Changeset.filter(changeset, upsert_condition)
changeset -> changeset
end)
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_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_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,
attrs_to_require,
action_select
) do
%{
must_return_records?: must_return_records_for_changes?,
batch: batch,
re_sort?: re_sort?,
changes: changes
} =
run_action_changes(
batch,
all_changes,
action,
opts[:actor],
opts[:authorize?],
opts[:tracer],
opts[:tenant]
)
batch =
if re_sort? do
Enum.sort_by(batch, fn changeset ->
{index, _metadata_key} =
BulkManualActionHelpers.extract_bulk_metadata(changeset, :bulk_create)
index
end)
else
batch
end
{batch, must_be_simple_results} =
batch
|> Enum.map(fn changeset ->
Ash.Changeset.require_values(
changeset,
:create
)
|> Ash.Changeset.require_values(
:update,
false,
changeset.action.require_attributes
)
end)
|> authorize(opts)
|> Ash.Actions.Helpers.split_and_run_simple(
action,
opts,
changes,
all_changes,
:bulk_create,
fn changeset ->
case Ash.Actions.Create.run(domain, changeset, action, opts) do
{:ok, result} ->
Process.put({:any_success?, ref}, true)
{index, metadata_key} =
BulkManualActionHelpers.extract_bulk_metadata(changeset, :bulk_create)
[
Ash.Resource.put_metadata(result, metadata_key, index)
]
{:ok, result, notifications} ->
Process.put({:any_success?, ref}, true)
store_notification(ref, notifications, opts)
{index, metadata_key} =
BulkManualActionHelpers.extract_bulk_metadata(changeset, :bulk_create)
[
Ash.Resource.put_metadata(result, metadata_key, index)
]
{:error, error} ->
store_error(ref, error, opts)
[]
end
end
)
batch =
Enum.reject(batch, fn
%{valid?: false} = changeset ->
store_error(ref, changeset, opts)
true
_changeset ->
false
end)
if opts[:transaction] == :batch &&
Ash.DataLayer.data_layer_can?(resource, :transact) do
context = batch |> Enum.at(0) |> Kernel.||(%{}) |> Map.get(:context)
notify? = opts[:notify?] && !Process.put(:ash_started_transaction?, true)
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,
changes,
must_return_records_for_changes?,
must_be_simple_results,
attrs_to_require,
action_select
)
end,
opts[:timeout],
%{
type: :bulk_create,
metadata: %{
resource: resource,
action: action.name,
actor: opts[:actor]
},
data_layer_context: opts[:data_layer_context] || context
},
rollback_on_error?: false
)
|> 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)
Process.delete(:ash_started_transaction?)
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,
attrs_to_require,
action_select
)
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,
attrs_to_require,
action_select
) do
must_return_records? =
opts[:notify?] ||
Enum.any?(batch, fn item ->
item.after_action != []
end)
batch =
Ash.Actions.Update.Bulk.run_bulk_before_batches(
batch,
changes,
all_changes,
opts,
ref,
:bulk_create
)
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,
attrs_to_require,
action_select,
changesets_by_index
)
|> run_after_action_hooks(opts, domain, ref)
|> process_results(
changes,
all_changes,
opts,
ref,
changesets_by_index,
batch,
domain,
resource,
must_return_records_for_changes?,
action
)
|> 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
{context_key, _metadata_key} =
BulkManualActionHelpers.create_bulk_operation_keys(action.type)
base
|> Ash.Changeset.put_context(context_key, %{index: index})
|> Ash.Changeset.set_private_arguments_for_action(opts[:private_arguments] || %{})
|> 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
if is_nil(Ash.Resource.Info.multitenancy_strategy(changeset.resource)) ||
Ash.Resource.Info.multitenancy_global?(changeset.resource) || changeset.tenant do
changeset
else
Ash.Changeset.add_error(
changeset,
Ash.Error.Invalid.TenantRequired.exception(resource: changeset.resource)
)
end
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
ash_context = Ash.ProcessHelpers.get_context_for_transfer(opts)
Task.async_stream(
stream,
fn batch ->
Ash.ProcessHelpers.transfer_context(ash_context, opts)
try do
Process.put(:ash_started_transaction?, true)
batch_result = callback.(batch)
{errors, _} = Process.get({:bulk_errors, ref}) || {[], 0}
notifications =
if opts[:notify?] do
process_notifications = Process.delete(:ash_notifications) || []
bulk_notifications = Process.delete({:bulk_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, Process.get({:any_success?, ref})}
catch
value ->
{:throw, value}
end
end,
timeout: :infinity,
max_concurrency: max_concurrency
)
|> Stream.map(fn
{:ok, {:throw, value}} ->
throw(value)
{:ok, {result, notifications, errors, any_success?}} ->
Process.put({:any_success?, ref}, any_success?)
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 ->
{index, _metadata_key} =
BulkManualActionHelpers.extract_bulk_metadata(changeset, :bulk_create)
Map.put(
changesets_by_index,
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 store_error(ref, empty, opts, error_count \\ nil)
defp store_error(_ref, empty, _opts, _error_count) when empty in [[], nil], do: :ok
defp store_error(ref, error, opts, error_count) 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_errors, ref}) || {[], 0}
new_errors =
error
|> List.wrap()
|> Enum.map(&Ash.Error.to_ash_error/1)
Process.put(
{:bulk_errors, ref},
{new_errors ++ errors, count + (error_count || Enum.count(new_errors))}
)
else
{errors, count} = Process.get({:bulk_errors, ref}) || {[], 0}
Process.put({:bulk_errors, ref}, {errors, count + (error_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_notifications, ref}) || []
new_notifications =
if is_list(notification) do
notification ++ notifications
else
[notification | notifications]
end
Process.put({:bulk_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,
&BulkManualActionHelpers.get_bulk_index(&1, :bulk_create)
)
}
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,
attrs_to_require,
action_select,
changesets_by_index
) do
batch
|> Enum.map(fn changeset ->
with %{valid?: true} = changeset <- changeset,
{changeset, %{notifications: new_notifications}} <-
Ash.Changeset.run_before_actions(changeset),
%{valid?: true} = changeset <-
Ash.Changeset.require_values(
changeset,
:create,
true,
attrs_to_require
) do
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
{%Ash.Changeset{} = changeset, _} -> changeset
%Ash.Changeset{} = changeset -> changeset
end
end)
|> Enum.reject(fn
%{valid?: false} = changeset ->
store_error(ref, changeset, opts)
true
_changeset ->
false
end)
|> case do
[] ->
{[], changesets_by_index}
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
changesets_by_index = index_changesets(batch)
batch
|> Enum.group_by(&{&1.atomics, &1.filter})
|> Enum.flat_map(fn {_atomics, batch} ->
result =
case action.manual do
{mod, manual_opts} ->
source_context =
case batch do
[cs | _] ->
cs.context
_ ->
%{}
end
if function_exported?(mod, :bulk_create, 3) do
mod.bulk_create(batch, manual_opts, %Ash.Resource.ManualCreate.BulkContext{
actor: opts[:actor],
select: opts[:select],
batch_size: opts[:batch_size],
authorize?: opts[:authorize?],
source_context: source_context,
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?,
return_notifications?: opts[:return_notifications?] || false,
return_errors?: opts[:return_errors?] || false,
tenant: opts[:tenant]
})
else
ctx = %Ash.Resource.ManualCreate.Context{
actor: opts[:actor],
select: opts[:select],
authorize?: opts[:authorize?],
source_context: source_context,
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_notifications?: opts[:return_notifications?] || false,
tenant: opts[:tenant]
}
[changeset] = batch
[
mod.create(changeset, manual_opts, ctx)
|> BulkManualActionHelpers.process_non_bulk_result(
changeset,
:bulk_create
)
]
end
|> Ash.Actions.Helpers.rollback_if_in_transaction(resource, nil)
|> BulkManualActionHelpers.process_bulk_results(
mod,
:bulk_create,
&store_notification/3,
&store_error/3,
ref,
opts
)
_ ->
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],
action_select: action_select,
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
),
upsert_condition:
case opts[:upsert_condition] do
nil -> action && action.upsert_condition
other -> other
end,
return_skipped_upsert?:
case opts[:return_skipped_upsert?] do
nil -> action && action.return_skipped_upsert?
other -> other
end,
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 | action_select: action_select},
upsert_keys
)
else
Ash.DataLayer.create(resource, %{changeset | action_select: action_select})
end
case result do
{:ok, {:upsert_skipped, _query, callback}} ->
if changeset.context[:private][:return_skipped_upsert?] do
case callback.() do
{:ok, record} ->
{index, metadata_key} =
BulkManualActionHelpers.extract_bulk_metadata(
changeset,
:bulk_create
)
{:ok,
[
Ash.Resource.put_metadata(record, metadata_key, index)
]}
_ ->
[]
end
else
[]
end
{:ok, %{__metadata__: %{upsert_skipped: true}} = record} ->
if changeset.context[:private][:return_skipped_upsert?] do
{index, metadata_key} =
BulkManualActionHelpers.extract_bulk_metadata(
changeset,
:bulk_create
)
{:ok,
[
Ash.Resource.put_metadata(record, metadata_key, index)
]}
else
[]
end
{:ok, result} ->
{index, metadata_key} =
BulkManualActionHelpers.extract_bulk_metadata(
changeset,
:bulk_create
)
{:ok,
[
Ash.Resource.put_metadata(result, metadata_key, index)
]}
{:error, :no_rollback, error} ->
{:error, :no_rollback, error}
{:error, error} ->
{:error, error}
end
end
end
case result do
{:ok, result} ->
result =
if tenant = opts[:tenant] do
Enum.map(result, fn record ->
%{record | __metadata__: Map.put(record.__metadata__, :tenant, tenant)}
end)
else
result
end
Process.put({:any_success?, ref}, true)
Ash.Actions.Helpers.select(result, %{resource: resource, select: action_select})
:ok ->
Process.put({:any_success?, ref}, true)
[]
{:error, :no_rollback, error} ->
store_error(ref, error, opts)
{:error, error} ->
store_error(ref, error, opts)
[]
end
end)
|> then(&{&1, changesets_by_index})
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, changesets_by_index},
opts,
domain,
ref
) do
Enum.flat_map(batch_results, fn result ->
index = BulkManualActionHelpers.get_bulk_index(result, :bulk_create)
changeset = changesets_by_index[index]
case manage_relationships(result, domain, changeset,
upsert?: opts[:upsert?],
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,
must_return_records_for_changes?,
action
) do
{_context_key, metadata_key} = BulkManualActionHelpers.create_bulk_operation_keys(action.type)
results =
Enum.flat_map(batch, fn result ->
index = BulkManualActionHelpers.get_bulk_index(result, :bulk_create)
changeset = changesets_by_index[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?] || must_return_records_for_changes? do
[result]
else
[]
end
{:error, error} ->
store_error(ref, error, opts)
[]
end
rescue
e ->
store_error(ref, e, opts)
[]
end
end)
changes
|> Ash.Actions.Update.Bulk.run_bulk_after_changes(
all_changes,
results,
changesets_by_index,
changesets,
opts,
ref,
resource,
metadata_key
)
|> then(fn records ->
if opts[:return_records?] do
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,
context: %{private: %{just_created_by_action: action.name}},
reuse_values?: true,
domain: domain,
action: Ash.Resource.Info.primary_action(resource, :read) || action,
tenant: opts[:tenant],
actor: opts[:actor],
authorize?: opts[:authorize?],
tracer: opts[:tracer]
) do
{:ok, records} ->
Ash.load(
records,
List.wrap(opts[:load]),
context: %{private: %{just_created_by_action: action.name}},
domain: domain,
tenant: opts[:tenant],
action: Ash.Resource.Info.primary_action(resource, :read) || action,
reuse_values?: true,
actor: opts[:actor],
authorize?: opts[:authorize?],
tracer: opts[:tracer]
)
|> case do
{:ok, records} ->
Enum.reject(records, & &1.__metadata__[:private][:missing_from_data_layer])
{:error, error} ->
store_error(ref, error, opts)
[]
end
{:error, error} ->
store_error(ref, error, opts)
[]
end
else
[]
end
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, _tenant, _arguments, _context, _changeset),
do: opts
defp templated_opts(opts, actor, tenant, arguments, context, changeset) do
Ash.Expr.fill_template(
opts,
actor: actor,
tenant: tenant,
args: arguments,
context: context,
changeset: changeset
)
end
defp run_action_changes(batch, all_changes, _action, actor, authorize?, tracer, tenant) do
source_context =
case batch do
[cs | _] ->
cs.context
_ ->
%{}
end
context = %{
actor: actor,
source_context: source_context,
authorize?: authorize? || false,
tracer: tracer,
tenant: tenant
}
Enum.reduce(
all_changes,
%{must_return_records?: false, re_sort?: false, batch: batch, changes: %{}},
fn
{%{validation: {module, opts}} = validation, _change_index}, %{batch: batch} = state ->
batch =
Enum.map(batch, fn changeset ->
cond do
!module.has_validate?() ->
Ash.Changeset.add_error(
changeset,
Ash.Error.Framework.CanNotBeAtomic.exception(
resource: changeset.resource,
change: module,
reason: "Create actions cannot be made atomic"
)
)
invalid =
Enum.find(validation.where, fn {module, _} -> !module.has_validate?() end) ->
{module, _} = invalid
Ash.Changeset.add_error(
changeset,
Ash.Error.Framework.CanNotBeAtomic.exception(
resource: changeset.resource,
change: module,
reason: "Create actions cannot be made atomic"
)
)
validation.only_when_valid? && !changeset.valid? ->
changeset
Enum.all?(validation.where || [], fn {module, opts} ->
opts =
templated_opts(
opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, opts} = module.init(opts)
Ash.Resource.Validation.validate(
module,
changeset,
opts,
struct(Ash.Resource.Validation.Context, context)
) == :ok
end) ->
opts =
templated_opts(
opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, opts} = module.init(opts)
case Ash.Resource.Validation.validate(
module,
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.Error.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.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, opts} = module.init(opts)
Ash.Resource.Validation.validate(
module,
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),
re_sort?: true,
changes:
Map.put(
state.changes,
change_index,
Enum.map(matches, fn changeset ->
{index, _metadata_key} =
BulkManualActionHelpers.extract_bulk_metadata(
changeset,
:bulk_create
)
index
end)
)
}
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)
Ash.Resource.Change.change(
module,
changeset,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
end)
end
change_opts ->
if module.has_batch_change?() do
batch
|> Enum.map(fn changeset ->
change_opts =
templated_opts(
change_opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, change_opts} = module.init(change_opts)
module.batch_change(
[changeset],
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
end)
|> Enum.concat()
else
Enum.map(batch, fn changeset ->
change_opts =
templated_opts(
change_opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, change_opts} = module.init(change_opts)
Ash.Resource.Change.change(
module,
changeset,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
end)
end
end
end
end