Packages
ash
3.27.5
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/helpers/bulk.ex
# SPDX-FileCopyrightText: 2019 ash contributors <https://github.com/ash-project/ash/graphs/contributors>
#
# SPDX-License-Identifier: MIT
defmodule Ash.Actions.Helpers.Bulk do
@moduledoc false
require Logger
@typedoc """
Tagged result tuple carrying a record/error with its associated changeset.
Used throughout bulk operations to maintain changeset context for hook execution.
"""
@type tagged_result ::
{:ok, Ash.Resource.Record.t(), Ash.Changeset.t()}
| {:error, term(), Ash.Changeset.t()}
@typedoc """
Extended tagged result that includes hook completion markers.
The `_hooks_done` variants indicate after_transaction hooks have already run,
preventing double execution during result processing.
"""
@type tagged_result_with_hooks ::
tagged_result()
| {:ok_hooks_done, Ash.Resource.Record.t(), Ash.Changeset.t()}
| {:error_hooks_done, term(), Ash.Changeset.t()}
@doc """
Conditionally rolls back transaction for bulk operations.
Only rolls back if `opts[:transaction]` and `opts[:rollback_on_error?]` are set
and we're inside a transaction. Returns the error unchanged for piping.
"""
@spec maybe_rollback(error, Ash.Resource.t(), Keyword.t()) :: error when error: term()
def maybe_rollback(error, resource, opts) do
if opts[:transaction] && opts[:rollback_on_error?] do
if Ash.DataLayer.in_transaction?(resource) do
Ash.DataLayer.rollback(resource, error)
end
end
error
end
@doc """
Conditionally stops bulk operation on error.
Throws if `stop_on_error?` is set and `return_stream?` is false.
Otherwise returns the error unchanged for piping.
"""
@spec maybe_stop_on_error(error, Keyword.t()) :: error | no_return() when error: term()
def maybe_stop_on_error(error, opts) do
if stop_on_error?(opts) do
throw({:error, Ash.Error.to_error_class(error)})
end
error
end
@doc """
Conditionally stops bulk create operation on error.
In addition to the checks performed in from `maybe_stop_on_error/2`, we pass
the resource and check if the data layer supports partial success (via
`:bulk_create_with_partial_success`). If it does, we don't stop on error
because the data layer may have already persisted some records successfully.
Stopping would hide those successful results from the caller.
Returns the error unchanged for piping.
"""
@spec maybe_stop_on_bulk_create_error(error, Ash.Resource.t() | nil, Keyword.t()) ::
error | no_return()
when error: term()
def maybe_stop_on_bulk_create_error(error, resource, opts) do
if stop_on_error?(opts) &&
!Ash.DataLayer.data_layer_can?(resource, :bulk_create_with_partial_success) do
throw({:error, Ash.Error.to_error_class(error)})
end
error
end
@doc """
Conditionally stops bulk create operation after a batch with errors.
For data layers that support `:bulk_create_with_partial_success`, when
`stop_on_error?` is true, we stop at the end of the first batch containing
errors (instead of at the first individual error). This allows us to return
all results from that batch, including successfully persisted records.
When stopping, notifications accumulated in `{:bulk_notifications, ref}` are
handled via `build_batch_partial_success_result/3`, which calls
`return_notifications_or_notify/2` to either return them or send them
immediately depending on `notify?` and `return_notifications?` options.
Designed for use with `Stream.transform/3`. Returns `{[batch_results_list],
acc ++ batch_results_list}` to continue processing, or throws
`{:batch_partial_success, accumulated ++ batch_results_list, ref}` to signal
that processing should stop.
"""
@spec maybe_stop_on_bulk_create_batch_error(
Enumerable.t(),
Ash.Resource.t(),
Keyword.t(),
reference(),
Enumerable.t()
) :: {Enumerable.t(), Enumerable.t()} | no_return()
def maybe_stop_on_bulk_create_batch_error(batch_results, resource, opts, ref, acc) do
batch_results_list = Enum.to_list(batch_results)
if stop_on_error?(opts) and
Ash.DataLayer.data_layer_can?(resource, :bulk_create_with_partial_success) and
any_batch_errors?(batch_results) do
throw({:batch_partial_success, acc ++ batch_results_list, ref})
else
# Return the result in the format expected by Stream.transform since we're called by it
{[batch_results_list], acc ++ batch_results_list}
end
end
defp any_batch_errors?(batch_results) do
Enum.any?(batch_results, &match?({:error, _}, &1))
end
@doc """
Constructs a BulkResult from batch results after a batch partial success.
Called when `maybe_stop_on_bulk_create_batch_error` throws, to build the final result
with all records and errors from the batch that caused the stop.
"""
@spec build_batch_partial_success_result(
batch_results :: [Ash.Resource.Record.t() | {:error, term()}],
reference(),
Keyword.t()
) :: Ash.BulkResult.t()
def build_batch_partial_success_result(batch_results, ref, opts) do
{records, error_tuples} =
Enum.split_with(batch_results, fn
{:error, _} -> false
_ -> true
end)
errors = Enum.map(error_tuples, &(&1 |> elem(1) |> Ash.Error.to_ash_error()))
status = if Process.get({:any_success?, ref}), do: :partial_success, else: :error
notifications = return_notifications_or_notify(ref, opts)
result = %Ash.BulkResult{
status: status,
records: records,
notifications: notifications
}
{error_count, errors} = Ash.Actions.Helpers.Bulk.errors(result, errors, opts)
%{result | errors: errors, error_count: error_count}
end
@doc """
Handles notifications at the end of a bulk operation.
This function consolidates the notification handling logic that is used in
multiple places in bulk operations. It handles three cases:
1. `notify?: false` - Returns `nil`, notifications are discarded
2. `return_notifications?: true` - Returns the accumulated notifications from
`{:bulk_notifications, ref}` for the caller to handle
3. `notify?: true` and `return_notifications?: false` - Calls
`Ash.Notifier.notify/1` to send notifications immediately (or defer them
if in a transaction by storing in `:ash_notifications`), then returns `nil`
"""
@spec return_notifications_or_notify(reference(), Keyword.t()) :: list() | nil
def return_notifications_or_notify(ref, opts) do
cond do
opts[:return_notifications?] ->
# We return notifications
Process.delete({:bulk_notifications, ref}) || []
!opts[:notify?] ->
# Nothing to do, just return nil
nil
true ->
# Here notify? == true and return_notifications? == false, so we need to notify as a side effect
# (possibly storing in ash_notifications for later) and return nil
unsent_notifications =
(Process.delete({:bulk_notifications, ref}) || [])
|> Ash.Notifier.notify()
if Process.get(:ash_started_transaction?) do
Process.put(
:ash_notifications,
Process.get(:ash_notifications, []) ++ unsent_notifications
)
end
nil
end
end
defp stop_on_error?(opts), do: opts[:stop_on_error?] && !opts[:return_stream?]
@doc """
Looks up the changeset for a result record using ref metadata.
First tries ref metadata (new), falls back to index->ref lookup (legacy).
"""
def lookup_changeset(result, changesets_by_ref, changesets_by_index, opts) do
index_key = opts[:index_key]
ref_key = opts[:ref_key] || :bulk_action_ref
result.__metadata__
|> get_ref_from_metadata(ref_key, index_key, changesets_by_index)
|> then(&changesets_by_ref[&1])
end
defp get_ref_from_metadata(metadata, ref_key, index_key, changesets_by_index) do
with nil <- metadata[ref_key],
index when not is_nil(index) <- metadata[index_key],
ref when not is_nil(ref) <- changesets_by_index[index] do
ref
else
ref when not is_nil(ref) -> ref
_ -> nil
end
end
@doc """
Sets bulk operation metadata on a result record from a changeset.
Uses pattern matching to extract index/ref from changeset context.
"""
def put_metadata(result, %{context: %{bulk_create: %{index: index, ref: ref}}}) do
result
|> Ash.Resource.put_metadata(:bulk_create_index, index)
|> maybe_put_ref_metadata(ref, :bulk_action_ref)
end
def put_metadata(result, %{context: %{bulk_destroy: %{index: index, ref: ref}}}) do
result
|> Ash.Resource.put_metadata(:bulk_destroy_index, index)
|> maybe_put_ref_metadata(ref, :bulk_action_ref)
end
def put_metadata(result, %{context: %{bulk_update: %{index: index, ref: ref}}}) do
result
|> Ash.Resource.put_metadata(:bulk_update_index, index)
|> maybe_put_ref_metadata(ref, :bulk_action_ref)
end
def put_metadata(result, _changeset), do: result
@doc """
Sets bulk operation metadata with explicit values (for batch operations).
Used when index/ref are already extracted as separate variables.
"""
def put_metadata(result, index, ref, index_key) when is_integer(index) do
result
|> Ash.Resource.put_metadata(index_key, index)
|> maybe_put_ref_metadata(ref, :bulk_action_ref)
end
defp maybe_put_ref_metadata(result, ref, ref_key) do
if Application.get_env(:ash, :test_bulk_index_only, false) do
result
else
Ash.Resource.put_metadata(result, ref_key, ref)
end
end
@doc """
Sets the error path to include the bulk operation index from the changeset context.
Extracts the index from bulk_create, bulk_update, or bulk_destroy context
and prepends it to the error's path.
"""
def set_index_path(error, %{context: %{bulk_create: %{index: index}}}),
do: Ash.Error.set_path(error, [index])
def set_index_path(error, %{context: %{bulk_update: %{index: index}}}),
do: Ash.Error.set_path(error, [index])
def set_index_path(error, %{context: %{bulk_destroy: %{index: index}}}),
do: Ash.Error.set_path(error, [index])
def set_index_path(error, _changeset), do: error
@doc """
Validates multitenancy requirements for bulk operations.
Checks if tenant is required based on resource configuration, action settings,
and context overrides. Returns `:ok` or an error with `TenantRequired` exception.
"""
@spec validate_multitenancy(Ash.Resource.t(), Ash.Resource.Actions.action(), keyword()) ::
:ok | {:error, Exception.t()}
def validate_multitenancy(resource, action, opts) do
if Ash.Resource.Info.multitenancy_strategy(resource) &&
!Ash.Resource.Info.multitenancy_global?(resource) && !opts[:tenant] &&
Map.get(action, :multitenancy) not in [:bypass, :bypass_all, :allow_global] &&
get_in(opts, [:context, :shared, :private, :multitenancy]) not in [
:bypass,
:bypass_all,
:allow_global
] do
{:error, Ash.Error.Invalid.TenantRequired.exception(resource: resource)}
else
:ok
end
end
@doc """
Marks that at least one record succeeded in a bulk operation.
Only sets to true, never overwrites true with false.
This ensures partial_success status when some batches succeed and others fail.
"""
@spec set_success(reference(), atom()) :: :ok
def set_success(ref, status) when status != :error do
Process.put({:any_success?, ref}, true)
:ok
end
def set_success(_ref, :error), do: :ok
@doc """
Stores notification in process dictionary for bulk operations.
"""
@spec store_notification(reference(), list() | term() | nil, Keyword.t()) :: :ok
def store_notification(_ref, empty, _opts) when empty in [[], nil], do: :ok
def 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
:ok
end
@doc """
Accumulates errors for bulk operation results.
Returns `{error_count, errors}` tuple.
"""
@spec errors(Ash.BulkResult.t(), term(), Keyword.t()) :: {non_neg_integer(), list()}
def 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
def errors(result, nil, _opts) do
{result.error_count + 1, []}
end
def errors(result, {:error, error}, opts) do
if opts[:return_errors?] do
{result.error_count + 1, [error | List.wrap(result.errors)]}
else
{result.error_count + 1, []}
end
end
def 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
@doc """
Splits a list of changesets into valid and invalid ones.
Returns a tuple where the first element contains valid changesets and the second
contains error tuples for invalid changesets in the form `{:error, error, changeset}`.
If `stop_on_error?` is true and `return_stream?` is false, throws on the first
invalid changeset encountered.
"""
@spec split_valid_invalid_changesets([Ash.Changeset.t()], Keyword.t()) ::
{[Ash.Changeset.t()], [{:error, Ash.Error.t(), Ash.Changeset.t()}]}
def split_valid_invalid_changesets(changesets, _opts) do
changesets
|> Enum.reduce({[], []}, fn
%{valid?: false} = changeset, {batch_acc, results_acc} ->
# in case of stop_on_error? we need to throw from here
# and stop processing further changesets
error = Ash.Error.to_error_class(changeset.errors, changeset: changeset)
{batch_acc, [{:error, error, changeset} | results_acc]}
changeset, {batch_acc, results_acc} ->
{[changeset | batch_acc], results_acc}
end)
|> then(fn {batch, invalid_changeset_errors} ->
{Enum.reverse(batch), Enum.reverse(invalid_changeset_errors)}
end)
end
@doc """
Checks if notifications need to be tracked for this bulk operation.
"""
@spec need_notifications?(Keyword.t()) :: boolean()
def need_notifications?(opts) do
opts[:notify?] || opts[:return_notifications?]
end
@typedoc """
Map correlating changeset references to their changesets.
Built during result processing to enable matching loaded records back to
their original changesets via the `bulk_changeset_id` metadata.
"""
@type changeset_by_id :: %{reference() => Ash.Changeset.t()}
@typedoc """
Function that creates a notification from a changeset and loaded record.
"""
@type notification_fn ::
(Ash.Changeset.t(), Ash.Resource.Record.t(), Keyword.t() ->
Ash.Notifier.Notification.t())
@doc """
Stores notifications for loaded records by matching via changeset_id in metadata.
Uses a unique reference (changeset_id) to correlate loaded records back to their
original changesets, which works for all resources including those without primary keys.
Buffers notifications in the process dictionary via `store_notification/3`.
"""
@spec store_notifications_for_loaded_records(
loaded_records :: [Ash.Resource.Record.t()],
changeset_by_id(),
bulk_ref :: reference(),
notification_fn(),
opts :: Keyword.t()
) :: :ok
def store_notifications_for_loaded_records(
loaded_records,
changeset_by_id,
ref,
notification_fn,
opts
) do
if need_notifications?(opts) do
Enum.each(loaded_records, fn loaded ->
changeset_id = loaded.__metadata__[:bulk_changeset_id]
case Map.fetch(changeset_by_id, changeset_id) do
{:ok, changeset} ->
store_notification(ref, notification_fn.(changeset, loaded, opts), opts)
:error ->
Logger.warning("""
Bulk operation: Could not find changeset for loaded record.
bulk_changeset_id: #{inspect(changeset_id)}
This may indicate a bug in bulk operation result processing.
""")
end
end)
end
:ok
end
@doc """
Removes internal bulk_changeset_id metadata from records before returning.
The bulk_changeset_id is an internal implementation detail used for matching
loaded records back to their changesets. It should not be exposed to users.
"""
@spec clean_changeset_id_metadata(list()) :: list()
def clean_changeset_id_metadata(records) when is_list(records) do
Enum.map(records, &clean_changeset_id_metadata/1)
end
def clean_changeset_id_metadata(record) when is_struct(record) do
case record.__metadata__[:bulk_changeset_id] do
nil -> record
_id -> Map.update!(record, :__metadata__, &Map.delete(&1, :bulk_changeset_id))
end
end
def clean_changeset_id_metadata(other), do: other
@doc """
Runs batch validation for a validation that implements `batch_validate/3`.
Splits the batch into matching/non-matching changesets based on `where` and
`only_when_valid?` conditions, runs `batch_validate` on the matches, and
merges the results back together.
"""
def run_batch_validation(
%{validation: {module, opts}} = validation,
batch,
validation_context,
actor
) do
{matches, non_matches} =
if Enum.empty?(validation.where) && !validation.only_when_valid? do
{batch, []}
else
Enum.split_with(batch, fn changeset ->
applies_from_only_when_valid? =
if validation.only_when_valid?, do: changeset.valid?, else: true
applies_from_where? =
Enum.all?(validation.where || [], fn {where_module, where_opts} ->
where_opts =
Ash.Actions.Helpers.templated_opts(
where_opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, where_opts} = where_module.init(where_opts)
Ash.Resource.Validation.validate(
where_module,
changeset,
where_opts,
validation_context
) == :ok
end)
applies_from_where? and applies_from_only_when_valid?
end)
end
if Enum.empty?(matches) do
non_matches
else
opts =
case opts do
{:templated, opts} -> opts
opts -> opts
end
{:ok, opts} = module.init(opts)
validated =
module.batch_validate(
matches,
opts,
struct(validation_context, bulk?: true)
)
|> Enum.to_list()
Enum.concat(validated, non_matches)
end
end
end