Packages
ash
3.0.0-rc.0
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/read/read.ex
defmodule Ash.Actions.Read do
@moduledoc false
alias Ash.Actions.Helpers
alias Ash.Error.Invalid.{LimitRequired, PaginationRequired}
alias Ash.Filter
require Logger
require Ash.Flags
require Ash.Query
import Ash.Expr
require Ash.Tracer
def unpaginated_read(query, action \\ nil, opts \\ []) do
run(query, action, Keyword.put(opts, :skip_pagination?, true))
end
@spec run(Ash.Query.t(), Ash.Resource.Actions.action(), Keyword.t()) ::
{:ok, Ash.Page.page() | list(Ash.Resource.record())}
| {:ok, Ash.Page.page() | list(Ash.Resource.record()), Ash.Query.t()}
| {:error, term}
def run(query, action, opts \\ [])
def run(%{valid?: false, errors: errors}, _action, _opts) do
{:error, Ash.Error.to_error_class(errors)}
end
def run(query, action, opts) do
query = Ash.Query.new(query)
domain = query.domain || opts[:domain]
if !domain do
raise Ash.Error.Framework.AssumptionFailed, message: "got a query without a domain"
end
{query, opts} = Ash.Actions.Helpers.set_context_and_get_opts(domain, query, opts)
query = Helpers.apply_opts_load(query, opts)
query = %{query | domain: domain}
action = get_action(query.resource, action || query.action)
Ash.Tracer.span :action,
Ash.Domain.Info.span_name(query.domain, query.resource, action.name),
opts[:tracer] do
metadata = %{
domain: query.domain,
resource: query.resource,
resource_short_name: Ash.Resource.Info.short_name(query.resource),
actor: opts[:actor],
tenant: opts[:tenant],
action: action.name,
authorize?: opts[:authorize?]
}
Ash.Tracer.telemetry_span [:ash, Ash.Domain.Info.short_name(query.domain), :read],
metadata do
Ash.Tracer.set_metadata(opts[:tracer], :action, metadata)
run_around_transaction_hooks(query, fn query ->
case do_run(query, action, opts) do
{:error, error} ->
if opts[:tracer] do
stacktrace =
case error do
%{stacktrace: %{stacktrace: stacktrace}} ->
stacktrace || []
_ ->
{:current_stacktrace, stacktrace} =
Process.info(self(), :current_stacktrace)
stacktrace
end
Ash.Tracer.set_handled_error(opts[:tracer], Ash.Error.to_error_class(error),
stacktrace: stacktrace
)
end
{:error, error}
other ->
other
end
end)
end
end
end
defp run_around_transaction_hooks(%{around_transaction: []} = query, func),
do: func.(query)
defp run_around_transaction_hooks(%{around_transaction: [around | rest]} = query, func) do
query
|> set_phase(:around_transaction)
|> around.(fn query ->
run_around_transaction_hooks(%{query | around_transaction: rest}, func)
end)
end
defp do_run(query, action, opts) do
{query, opts} =
if opts[:unsafe_no_authorize?] do
{Ash.Query.set_context(query, %{private: %{authorize?: false}}),
Keyword.put(opts, :authorize?, false)}
else
{query, opts}
end
opts = sanitize_opts(opts, query)
action = get_action(query.resource, query.action || action)
query =
for_read(
query,
action,
actor: opts[:actor],
authorize?: opts[:authorize?],
timeout: opts[:timeout],
tenant: opts[:tenant]
)
initial_query = query
query = add_field_level_auth(query, query.domain, opts)
query = %{
query
| timeout:
opts[:timeout] || query.timeout || query.action.timeout ||
Ash.Domain.Info.timeout(query.domain)
}
query =
add_calc_context_to_query(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain
)
page_opts = page_opts(action, opts)
query =
if opts[:page] do
query
|> Ash.Query.set_context(%{
initial_limit: query.limit,
initial_offset: query.offset,
page_opts:
unless opts[:inital_data] do
page_opts
end,
initial_query: query,
query_opts: opts
})
else
query
end
opts = Keyword.put(opts, :page, page_opts)
query =
if opts[:page] && opts[:page][:limit] &&
(opts[:page][:before] || opts[:page][:after] ||
(action.pagination.keyset? && !opts[:page][:offset])) do
load_and_select_sort(query)
else
query
end
pkey = Ash.Resource.Info.primary_key(query.resource)
missing_pkeys? =
opts[:initial_data] &&
Enum.any?(opts[:initial_data], fn record ->
Enum.any?(Map.take(record, pkey), fn {_, v} -> is_nil(v) end)
end)
reuse_values? = Keyword.get(opts, :reuse_values?, false)
{calculations_in_query, calculations_at_runtime, query} =
Ash.Actions.Read.Calculations.split_and_load_calculations(
query.domain,
query,
missing_pkeys?,
Keyword.fetch(opts, :initial_data),
reuse_values?
)
query =
add_calc_context_to_query(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain
)
query =
if opts[:initial_data] do
select = source_fields(query) ++ (query.select || [])
select =
if reuse_values? do
remove_already_selected(select, opts[:initial_data])
else
select
end
query = %{query | select: select}
if opts[:lazy?] do
unload_loaded_calculations_and_aggregates(query, opts[:initial_data])
else
query
end
else
Ash.Query.ensure_selected(query, source_fields(query))
end
{query, stop?} = add_async_limiter(query, calculations_at_runtime, opts)
try do
data_result =
if opts[:initial_data] do
load(
opts[:initial_data],
query,
calculations_at_runtime,
calculations_in_query,
missing_pkeys?,
opts
)
else
do_read(query, calculations_at_runtime, calculations_in_query, opts)
end
data_result =
case data_result do
{:ok, result} -> {:ok, result, nil}
{:ok, result, count} -> {:ok, result, count}
other -> other
end
with {:ok, data, count} <- data_result,
{:ok, data} <-
load_through_attributes(
data,
%{query | calculations: Map.new(calculations_in_query, &{&1.name, &1})},
query.domain,
opts[:actor],
opts[:tracer],
opts[:authorize?]
),
{:ok, data} <-
Ash.Actions.Read.Relationships.load(
data,
query,
opts[:lazy?]
),
{:ok, data} <-
Ash.Actions.Read.Calculations.run(
data,
query,
calculations_at_runtime,
calculations_in_query
),
{:ok, data} <-
load_through_attributes(
data,
%{
query
| calculations: Map.new(calculations_at_runtime, &{&1.name, &1}),
load_through: Map.delete(query.load_through || %{}, :attribute)
},
query.domain,
opts[:actor],
opts[:tracer],
opts[:authorize?],
false
) do
query =
query
# prevent leakage of stale pid as we stop it at the end of reading
|> clearup_async_limiter()
data
|> Helpers.restrict_field_access(query)
|> add_tenant(query)
|> attach_fields(opts[:initial_data], initial_query, missing_pkeys?)
|> add_page(
query.action,
count,
query.sort,
query,
opts
)
|> add_query(query, opts)
else
{:error, %Ash.Query{errors: errors} = query} ->
{:error, Ash.Error.to_error_class(errors, query: query)}
{:error, error} ->
{:error, Ash.Error.to_error_class(error, query: query)}
end
after
if stop? do
Agent.stop(query.context[:private][:async_limiter])
end
end
end
defp do_read(%{action: action} = query, calculations_at_runtime, calculations_in_query, opts) do
maybe_in_transaction(query, opts, fn ->
with %{valid?: true} = query <- handle_attribute_multitenancy(query),
:ok <- validate_multitenancy(query),
{:ok, sort} <-
Ash.Actions.Sort.process(
query.resource,
query.sort,
query.aggregates,
query.context
),
query <- Map.put(query, :sort, sort),
query <- add_select_if_none_exists(query),
query <- %{
query
| filter:
add_calc_context_to_filter(
query.filter,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain
)
},
pre_authorization_query <- query,
{:ok, query} <- authorize_query(query, opts),
query_before_pagination <- query,
query <-
Ash.Actions.Read.Calculations.deselect_known_forbidden_fields(
query,
calculations_at_runtime ++ calculations_in_query
),
{:ok, data_layer_calculations} <- hydrate_calculations(query, calculations_in_query),
{:ok, relationship_path_filters} <-
Ash.Filter.relationship_filters(
query.domain,
pre_authorization_query,
opts[:actor],
query.tenant,
agg_refs(query, data_layer_calculations ++ [{nil, query.filter}]),
opts[:authorize?]
),
data_layer_calculations <-
authorize_calculation_expressions(
data_layer_calculations,
query.resource,
opts[:authorize?],
relationship_path_filters,
opts[:actor],
query.tenant,
opts[:tracer],
query.domain
),
query <-
authorize_loaded_aggregates(
query,
relationship_path_filters,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer]
),
{:ok, filter} <-
filter_with_related(
query,
opts[:authorize?],
relationship_path_filters
),
{:ok, filter} <-
Filter.run_other_data_layer_filters(
query.domain,
query.resource,
filter,
query.tenant
),
filter <-
add_calc_context_to_filter(
filter,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain
),
filter <-
update_aggregate_filters(
filter,
query.resource,
opts[:authorize?],
relationship_path_filters,
opts[:actor],
query.tenant,
opts[:tracer],
query.domain
),
query <- Map.put(query, :filter, filter),
query <- Ash.Query.unset(query, :calculations),
{%{valid?: true} = query, before_notifications} <- run_before_action(query),
{:ok, count} <-
fetch_count(
query,
query_before_pagination,
relationship_path_filters,
opts
),
{:ok, query} <- paginate(query, action, opts[:page], opts[:skip_pagination?]),
{:ok, data_layer_query} <-
Ash.Query.data_layer_query(query, data_layer_calculations: data_layer_calculations),
{:ok, results} <-
run_query(
set_phase(query, :executing),
data_layer_query,
%{
actor: opts[:actor],
tenant: query.tenant,
authorize?: opts[:authorize?],
domain: query.domain
},
!Keyword.has_key?(opts, :initial_data)
)
|> Helpers.rollback_if_in_transaction(
query.resource,
query
),
:ok <- validate_get(results, query.action, query),
results <- add_keysets(query, results, query.sort),
{:ok, results} <- run_authorize_results(query, results),
{:ok, results, after_notifications} <- run_after_action(query, results),
{:ok, count} <- maybe_await(count) do
{:ok, results, count, before_notifications ++ after_notifications}
else
{%{valid?: false} = query, before_notifications} ->
{:error, query, before_notifications}
{:error, %Ash.Query{} = query} ->
{:error, query}
%Ash.Query{} = query ->
{:error, query}
{:error, error} ->
{:error, error}
end
end)
end
defp agg_refs(query, calculations_in_query) do
calculations_in_query
|> Enum.flat_map(fn {_, expr} ->
Ash.Filter.used_aggregates(expr, :*)
end)
|> Enum.concat(Map.values(query.aggregates))
end
defp source_fields(query) do
query
|> Ash.Query.accessing([:relationships])
|> Enum.flat_map(fn name ->
case Ash.Resource.Info.relationship(query.resource, name) do
%{no_attributes?: true} ->
[]
%{manual: {module, opts}, source_attribute: source_attribute} ->
fields =
module.select(opts)
if Ash.Resource.Info.attribute(query.resource, source_attribute) do
[source_attribute | fields]
else
fields
end
%{source_attribute: source_attribute} ->
[source_attribute]
end
end)
end
defp clearup_async_limiter(
%{context: %{private: %{async_limiter: async_limiter}} = ctx} = query
)
when is_pid(async_limiter) do
Ash.Query.set_context(query, put_in(ctx[:private][:async_limiter], nil))
end
defp clearup_async_limiter(query), do: query
def add_async_limiter(
%{context: %{private: %{async_limiter: async_limiter}}} = query,
_context,
_opts
)
when is_pid(async_limiter) do
{query, false}
end
def add_async_limiter(query, calculations_at_runtime, opts) do
if Enum.count_until(calculations_at_runtime, 2) + Enum.count_until(query.load, 2) >= 2 &&
!Application.get_env(:ash, :disable_async?) do
{:ok, limiter} =
Ash.Actions.Read.AsyncLimiter.start_link(
opts[:max_concurrency] || System.schedulers_online() * 2
)
{Ash.Query.set_context(query, %{private: %{async_limiter: limiter}}), true}
else
{query, false}
end
end
defp maybe_in_transaction(query, opts, func) do
notify? =
if Process.get(:ash_started_transaction?) do
false
else
Process.put(:ash_started_transaction?, true)
true
end
try do
cond do
query.action.transaction? ->
Ash.DataLayer.transaction(
[query.resource | query.action.touches_resources],
func,
query.timeout,
%{
type: :read,
metadata: %{
query: query,
resource: query.resource,
action: query.action.name
},
data_layer_context: query.context[:data_layer]
}
)
query.timeout ->
{:ok,
Ash.ProcessHelpers.task_with_timeout(
func,
query.resource,
query.timeout,
"#{inspect(query.resource)}.#{query.action.name}",
opts[:tracer]
)}
true ->
{:ok, func.()}
end
|> case do
{:ok, {:ok, result, count, notifications}} ->
notify_or_store(query, notifications, notify?)
{:ok, result, count}
{:ok, {:error, error, notifications}} ->
notify_or_store(query, notifications, notify?)
{:error, error}
{:ok, value} ->
value
other ->
other
end
after
if notify? do
Process.put(:ash_started_transaction?, false)
end
end
end
defp notify_or_store(_query, notifications, notify?) do
if notify? do
notifications =
Ash.Notifier.notify(notifications ++ (Process.delete(:ash_notifications) || []))
Process.put(:ash_notifications, notifications)
else
current_notifications = Process.get(:ash_notifications, [])
Process.put(
:ash_notifications,
notifications ++ current_notifications
)
end
end
defp add_select_if_none_exists(query) do
if query.select do
query
else
to_select =
query.resource
|> Ash.Resource.Info.attributes()
|> Enum.map(& &1.name)
Ash.Query.select(query, to_select)
end
end
defp load(
initial_data,
query,
calculations_at_runtime,
calculations_in_query,
missing_pkeys?,
opts
) do
must_be_reselected = List.wrap(query.select) -- Ash.Resource.Info.primary_key(query.resource)
query =
Ash.Actions.Read.Calculations.deselect_known_forbidden_fields(
query,
calculations_at_runtime ++ calculations_in_query
)
if missing_pkeys? ||
(Enum.empty?(must_be_reselected) && Enum.empty?(query.aggregates) &&
Enum.empty?(calculations_in_query)) do
{:ok, initial_data}
else
reselect_and_load(
initial_data,
query,
must_be_reselected,
calculations_in_query,
opts
)
end
end
defp reselect_and_load(
initial_data,
query,
must_be_reselected,
calculations_in_query,
opts
) do
primary_key = Ash.Resource.Info.primary_key(query.resource)
filter =
initial_data
|> List.wrap()
|> Enum.map(&Map.take(&1, primary_key))
|> case do
[] ->
false
[single] ->
[single]
multiple ->
[or: multiple]
end
with %{valid?: true} = query <-
query
|> Ash.Query.unset([
:filter,
:sort,
:limit,
:offset,
:distinct,
:select,
:calculations
])
|> Ash.Query.load(calculations_in_query)
|> Ash.Query.select(must_be_reselected)
|> Ash.DataLayer.Simple.set_data(initial_data)
|> Ash.Query.do_filter(filter),
{:ok, data_layer_calculations} <-
hydrate_calculations(
query,
calculations_in_query
),
{:ok, relationship_path_filters} <-
Ash.Filter.relationship_filters(
query.domain,
%{query | filter: nil},
opts[:actor],
query.tenant,
agg_refs(query, data_layer_calculations),
opts[:authorize?]
),
data_layer_calculations <-
authorize_calculation_expressions(
data_layer_calculations,
query.resource,
opts[:authorize?],
relationship_path_filters,
opts[:actor],
query.tenant,
opts[:tracer],
query.domain
),
query <-
authorize_loaded_aggregates(
query,
relationship_path_filters,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer]
),
{:ok, data_layer_query} <- Ash.Query.data_layer_query(query),
{:ok, data_layer_query} <-
Ash.DataLayer.add_aggregates(
data_layer_query,
Map.values(query.aggregates),
query.resource
),
{:ok, data_layer_query} <-
Ash.DataLayer.add_calculations(
data_layer_query,
data_layer_calculations,
query.resource
),
{:ok, results} <-
run_query(
query,
data_layer_query,
%{
actor: opts[:actor],
tenant: query.tenant,
authorize?: false,
domain: query.domain
},
true
) do
results
|> attach_fields(initial_data, query, false)
|> compute_expression_at_runtime_for_missing_records(query, data_layer_calculations)
|> case do
{:ok, result} ->
{:ok, result, 0}
{:error, error} ->
{:error, error}
end
end
end
defp compute_expression_at_runtime_for_missing_records(data, query, data_layer_calculations) do
if Enum.any?(data, & &1.__metadata__[:private][:missing_from_data_layer]) do
{require_calculating, rest} =
data
|> Stream.with_index()
|> Stream.map(fn {record, index} ->
Ash.Resource.put_metadata(record, :private, %{result_index: index})
end)
|> Enum.split_with(& &1.__metadata__[:private][:missing_from_data_layer])
require_calculating
|> recalculate(query, Enum.map(data_layer_calculations, &elem(&1, 0)))
|> case do
{:ok, result} ->
{:ok,
result
|> Enum.concat(rest)
|> Enum.sort_by(& &1.__metadata__[:private][:result_index])}
{:error, error} ->
{:error, error}
end
else
{:ok, data}
end
end
defp recalculate(require_calculating, query, data_layer_calculations) do
Ash.Actions.Read.Calculations.run(require_calculating, query, data_layer_calculations, [])
end
defp authorize_query(query, opts) do
if opts[:authorize?] do
case Ash.can(query, opts[:actor],
return_forbidden_error?: true,
maybe_is: false,
run_queries?: false,
alter_source?: true
) do
{:ok, true} ->
{:ok, query}
{:ok, true, query} ->
{:ok, query}
{:ok, false, error} ->
{:error, error}
{:error, error} ->
{:error, error}
end
else
{:ok, query}
end
end
defp unload_loaded_calculations_and_aggregates(query, initial_data) do
query =
query
|> Map.update!(:calculations, fn calculations ->
keys =
calculations
|> Enum.reject(fn {_key, calc} ->
Ash.Resource.loaded?(initial_data, calc)
end)
|> Enum.map(&elem(&1, 0))
Map.drop(calculations, keys)
end)
|> Map.update!(:aggregates, fn aggregates ->
keys =
aggregates
|> Enum.reject(fn {_key, calc} ->
Ash.Resource.loaded?(initial_data, calc)
end)
|> Enum.map(&elem(&1, 0))
Map.drop(aggregates, keys)
end)
query
|> Map.update!(:load_through, fn load_through ->
Map.update(load_through, :calculation, %{}, fn calculations ->
Map.take(calculations, Map.keys(query.calculations))
end)
end)
end
defp load_through_attributes(
results,
query,
domain,
actor,
tracer,
authorize?,
attrs? \\ true
) do
query.resource
|> Ash.Resource.Info.attributes()
|> Enum.filter(fn %{name: name, type: type, constraints: constraints} ->
Ash.Type.can_load?(type, constraints) && Ash.Query.selecting?(query, name)
end)
|> Enum.map(& &1.name)
|> Enum.reduce(query.load_through, fn name, load_through ->
Map.update(load_through, :attribute, %{name => []}, &Map.put_new(&1, name, []))
end)
|> Enum.reduce_while({:ok, results}, fn
{:calculation, load_through}, {:ok, results} ->
load_through
|> Map.take(Map.keys(query.calculations))
|> Enum.reduce_while({:ok, results}, fn {name, load_statement}, {:ok, results} ->
calculation = Map.get(query.calculations, name)
values =
case calculation.load do
nil ->
Enum.map(results, &Map.get(&1.calculations, calculation.name))
load ->
Enum.map(results, &Map.get(&1, load))
end
case calculation.type do
{:array, _} ->
Enum.reduce_while(values, {:ok, []}, fn list, {:ok, acc} ->
case Ash.Type.load(
calculation.type,
list,
load_statement,
calculation.constraints,
%{
domain: domain,
actor: actor,
tenant: query.tenant,
tracer: tracer,
authorize?: authorize?
}
) do
{:ok, new_values} ->
{:cont, {:ok, [new_values | acc]}}
{:error, error} ->
{:halt, {:error, error}}
end
end)
|> case do
{:ok, list} ->
case calculation.load do
nil ->
{:cont,
{:ok,
Enum.zip_with(results, Enum.reverse(list), fn record, value ->
Map.update!(record, :calculations, fn calculations ->
Map.put(calculations, calculation.name, value)
end)
end)}}
load ->
{:cont,
{:ok,
Enum.zip_with(results, Enum.reverse(list), fn record, value ->
Map.put(record, load, value)
end)}}
end
{:error, error} ->
{:halt, {:error, error}}
end
_ ->
case Ash.Type.load(
calculation.type,
values,
load_statement,
calculation.constraints,
%{
domain: domain,
actor: actor,
tenant: query.tenant,
tracer: tracer,
authorize?: authorize?
}
) do
{:ok, new_values} ->
case calculation.load do
nil ->
{:cont,
{:ok,
Enum.zip_with(results, new_values, fn record, value ->
Map.update!(record, :calculations, fn calculations ->
Map.put(calculations, calculation.name, value)
end)
end)}}
load ->
{:cont,
{:ok,
Enum.zip_with(results, new_values, fn record, value ->
Map.put(record, load, value)
end)}}
end
{:error, error} ->
{:halt, {:error, error}}
end
end
end)
|> case do
{:ok, results} ->
{:cont, {:ok, results}}
{:error, error} ->
{:halt, {:error, error}}
end
{:attribute, load_through}, {:ok, results} when attrs? ->
load_through
|> Enum.reduce_while({:ok, results}, fn {name, load_statement}, {:ok, results} ->
load_statement =
if is_map(load_statement) and not is_struct(load_statement) do
Map.to_list(load_statement)
else
load_statement
end
attribute = Ash.Resource.Info.attribute(query.resource, name)
values = Enum.map(results, &Map.get(&1, attribute.name))
case attribute.type do
{:array, _} ->
Enum.reduce_while(values, {:ok, []}, fn list, {:ok, acc} ->
case Ash.Type.load(
attribute.type,
list,
load_statement,
attribute.constraints,
%{
domain: domain,
actor: actor,
tenant: query.tenant,
tracer: tracer,
authorize?: authorize?
}
) do
{:ok, new_values} ->
{:cont, {:ok, [new_values | acc]}}
{:error, error} ->
{:halt, {:error, error}}
end
end)
|> case do
{:ok, list} ->
{:cont,
{:ok,
Enum.zip_with(results, Enum.reverse(list), fn record, value ->
Map.put(record, attribute.name, value)
end)}}
{:error, error} ->
{:halt, {:error, error}}
end
_ ->
case Ash.Type.load(
attribute.type,
values,
load_statement,
attribute.constraints,
%{
domain: domain,
actor: actor,
tenant: query.tenant,
tracer: tracer,
authorize?: authorize?
}
) do
{:ok, new_values} ->
{:cont,
{:ok,
Enum.zip_with(results, new_values, fn record, value ->
Map.put(record, attribute.name, value)
end)}}
{:error, error} ->
{:halt, {:error, error}}
end
end
end)
|> case do
{:ok, results} ->
{:cont, {:ok, results}}
{:error, error} ->
{:halt, {:error, error}}
end
_, {:ok, results} ->
{:cont, {:ok, results}}
end)
end
@doc false
def add_calc_context_to_filter(filter, actor, authorize?, tenant, tracer, domain) do
Ash.Filter.map(filter, fn
%Ash.Query.Parent{} = parent ->
%{
parent
| expr:
add_calc_context_to_filter(parent.expr, actor, authorize?, tenant, tracer, domain)
}
%Ash.Query.Exists{} = exists ->
%{
exists
| expr:
add_calc_context_to_filter(exists.expr, actor, authorize?, tenant, tracer, domain)
}
%Ash.Query.Ref{attribute: %Ash.Resource.Calculation{}} = ref ->
raise Ash.Error.Framework.AssumptionFailed,
message: "unhandled calculation in filter statement #{inspect(ref)}"
%Ash.Query.Ref{attribute: %Ash.Resource.Aggregate{}} = ref ->
raise Ash.Error.Framework.AssumptionFailed,
message: "unhandled calculation in filter statement #{inspect(ref)}"
%Ash.Query.Ref{
attribute: %Ash.Query.Calculation{} = calc,
relationship_path: relationship_path
} = ref ->
calc = add_calc_context(calc, actor, authorize?, tenant, tracer, domain)
if calc.module.has_expression?() do
expr =
case calc.module.expression(calc.opts, calc.context) do
%Ash.Query.Function.Type{} = expr ->
expr
expr ->
{:ok, expr} = Ash.Query.Function.Type.new([expr, calc.type, calc.constraints])
expr
end
{:ok, expr} =
Ash.Filter.hydrate_refs(
expr,
%{
resource: ref.resource,
public?: false
}
)
expr =
Ash.Filter.move_to_relationship_path(expr, relationship_path)
add_calc_context_to_filter(
expr,
actor,
authorize?,
tenant,
tracer,
domain
)
else
%{ref | attribute: calc}
end
%Ash.Query.Ref{attribute: %Ash.Query.Aggregate{} = agg} = ref ->
%{
ref
| attribute: add_calc_context(agg, actor, authorize?, tenant, tracer, domain)
}
other ->
other
end)
end
@doc false
def handle_attribute_multitenancy(query) do
multitenancy_attribute = Ash.Resource.Info.multitenancy_attribute(query.resource)
if multitenancy_attribute && query.tenant do
{m, f, a} = Ash.Resource.Info.multitenancy_parse_attribute(query.resource)
attribute_value = apply(m, f, [query.tenant | a])
Ash.Query.filter(query, ^ref(multitenancy_attribute) == ^attribute_value)
else
query
end
end
defp get_action(resource, action) do
cond do
action && is_atom(action) ->
Ash.Resource.Info.action(resource, action)
action ->
action
true ->
Ash.Resource.Info.primary_action!(resource, :read)
end
end
defp sanitize_opts(opts, query) do
Keyword.merge(opts, Map.get(query.context, :override_domain_params) || [])
end
defp for_read(query, action, opts) do
if query.__validated_for_action__ == action.name do
query
else
Ash.Query.for_read(query, action.name, %{}, opts)
end
end
@doc false
def validate_multitenancy(query) do
if is_nil(Ash.Resource.Info.multitenancy_strategy(query.resource)) ||
Ash.Resource.Info.multitenancy_global?(query.resource) || query.tenant do
:ok
else
{:error, Ash.Error.Invalid.TenantRequired.exception(resource: query.resource)}
end
end
defp add_tenant(data, query) do
if Ash.Resource.Info.multitenancy_strategy(query.resource) do
Enum.map(data, fn item ->
%{item | __metadata__: Map.put(item.__metadata__, :tenant, query.tenant)}
end)
else
data
end
end
defp add_query(result, query, opts) do
if opts[:return_query?] do
{:ok, result, query}
else
{:ok, result}
end
end
@doc false
def add_page(data, action, count, sort, original_query, opts) do
cond do
opts[:skip_pagination?] ->
data
action.pagination == false ->
data
opts[:page] == false ->
data
opts[:page][:limit] ->
to_page(data, action, count, sort, original_query, opts)
true ->
data
end
end
@doc false
def to_page(data, action, count, sort, original_query, opts) do
page_opts = opts[:page]
{data, rest} =
if page_opts[:limit] do
Enum.split(data, page_opts[:limit])
else
{data, []}
end
data =
if page_opts[:before] do
Enum.reverse(data)
else
data
end
more? = not Enum.empty?(rest)
if page_opts[:offset] do
Ash.Page.Offset.new(data, count, original_query, more?, opts)
else
cond do
action.pagination.keyset? && (page_opts[:after] || page_opts[:before]) ->
Ash.Page.Keyset.new(data, count, sort, original_query, more?, opts)
action.pagination.offset? && action.pagination.keyset? ->
Ash.Page.Offset.new(data, count, original_query, more?, opts)
action.pagination.offset? ->
Ash.Page.Offset.new(data, count, original_query, more?, opts)
true ->
Ash.Page.Keyset.new(data, count, sort, original_query, more?, opts)
end
end
end
defp add_keysets(original_query, data, sort) do
if Enum.any?(
Ash.Resource.Info.actions(original_query.resource),
&(&1.type == :read && &1.pagination && &1.pagination.keyset?)
) do
Ash.Page.Keyset.data_with_keyset(data, original_query.resource, sort)
else
data
end
end
defp remove_already_selected(fields, %struct{results: results})
when struct in [Ash.Page.Keyset, Ash.Page.Offset],
do: remove_already_selected(fields, results)
defp remove_already_selected(fields, record) when not is_list(record),
do: remove_already_selected(fields, List.wrap(record))
defp remove_already_selected(fields, initial_data) do
Enum.reject(fields, fn field ->
Enum.any?(initial_data, &Ash.Resource.selected?(&1, field))
end)
end
defp attach_fields(
data,
nil,
_original_query,
_missing_pkeys?
) do
data
end
defp attach_fields(
data_with_selected,
data,
original_query,
missing_pkeys?
) do
{aggregates_in_data, aggregates_in_aggregates} =
original_query.aggregates
|> Map.values()
|> Enum.split_with(& &1.load)
{calculations_in_data, calculations_in_calculations} =
original_query.calculations
|> Map.values()
|> Enum.split_with(& &1.load)
fields_from_data =
(original_query.select || []) ++
Keyword.keys(original_query.load || []) ++
Enum.map(aggregates_in_data, & &1.load) ++ Enum.map(calculations_in_data, & &1.load)
fields_from_aggregates =
Enum.map(aggregates_in_aggregates, & &1.name)
fields_from_calculations =
Enum.map(calculations_in_calculations, & &1.name)
if Enum.empty?(fields_from_calculations) and Enum.empty?(fields_from_aggregates) and
Enum.empty?(fields_from_data) do
data
else
pkey = Ash.Resource.Info.primary_key(original_query.resource)
# we have to assume they are all there and in the same order. Not my
# favorite thing, but no way around it in the short term.
if Enum.empty?(pkey) || missing_pkeys? do
Enum.zip_with([data, data_with_selected], fn [record, match] ->
record
|> Map.merge(Map.take(match, fields_from_data))
|> Map.update!(
:aggregates,
&Map.merge(
&1,
Map.take(match.aggregates, fields_from_aggregates)
)
)
|> Map.update!(
:calculations,
&Map.merge(
&1,
Map.take(match.calculations, fields_from_calculations)
)
)
end)
else
Enum.map(data, fn record ->
case Enum.find(data_with_selected, fn selected_record ->
record.__struct__.primary_key_matches?(record, selected_record)
end) do
nil ->
Ash.Resource.put_metadata(record, :private, %{missing_from_data_layer: true})
match ->
record
|> Map.merge(Map.take(match, fields_from_data))
|> Map.update!(
:aggregates,
&Map.merge(
&1,
Map.take(match.aggregates, fields_from_aggregates)
)
)
|> Map.update!(
:calculations,
&Map.merge(
&1,
Map.take(match.calculations, fields_from_calculations)
)
)
end
end)
end
end
end
defp validate_get([_, _ | _] = results, %{get?: true}, query) do
{:error,
Ash.Error.Invalid.MultipleResults.exception(
count: Enum.count(results),
at_least?: true,
query: query
)}
end
defp validate_get(_, _, _), do: :ok
defp add_calc_context_to_query(query, actor, authorize?, tenant, tracer, domain) do
%{
query
| load:
Keyword.new(query.load, fn {key, related_query} ->
case related_query do
%Ash.Query{} = related_query ->
{key,
add_calc_context_to_query(
related_query,
actor,
authorize?,
tenant,
tracer,
domain
)}
other ->
load =
query.resource
|> Ash.Resource.Info.related(key)
|> Ash.Query.new(domain: query.domain)
|> Ash.Query.load(other)
|> add_calc_context_to_query(actor, authorize?, tenant, tracer, domain)
{key, load}
end
end),
aggregates:
Map.new(query.aggregates, fn {key, agg} ->
{key,
add_calc_context(agg, actor, agg.authorize? && authorize?, tenant, tracer, domain)}
end),
calculations:
Map.new(query.calculations, fn {key, calc} ->
{key, add_calc_context(calc, actor, authorize?, tenant, tracer, domain)}
end),
filter:
add_calc_context_to_filter(query.filter, actor, authorize?, tenant, tracer, domain)
}
end
@doc false
def add_calc_context(%Ash.Query.Aggregate{} = agg, actor, authorize?, tenant, tracer, domain) do
read_action =
agg.read_action || Ash.Resource.Info.primary_action!(agg.query.resource, :read).name
query =
if agg.query.__validated_for_action__ != read_action do
domain =
if agg.query.domain do
agg.query.domain
else
Ash.Domain.Info.related_domain(agg.resource, agg.relationship_path, domain)
end
agg.query
|> Ash.Query.set_context(%{private: %{require_actor?: false}})
|> Ash.Query.for_read(read_action, %{}, domain: domain)
else
agg.query
end
%{
agg
| context:
Map.merge(
%{
actor: actor,
authorize?: authorize?,
tenant: tenant,
tracer: tracer
},
agg.context
),
query: add_calc_context_to_query(query, actor, authorize?, tenant, tracer, domain),
join_filters:
Map.new(agg.join_filters, fn {key, filter} ->
{key, add_calc_context_to_filter(filter, actor, authorize?, tenant, tracer, domain)}
end)
}
end
def add_calc_context(calc, actor, authorize?, tenant, tracer, _domain) do
%{
calc
| context: %{
calc.context
| actor: actor,
authorize?: authorize?,
tenant: tenant,
tracer: tracer
}
}
end
@doc false
def add_calc_context(calc, context, domain) do
add_calc_context(
calc,
context.actor,
context.authorize?,
context.tenant,
context.tracer,
domain
)
end
@doc false
def update_aggregate_filters(
filter,
_resource,
authorize?,
relationship_path_filters,
actor,
tenant,
tracer,
domain
) do
if authorize? do
Filter.update_aggregates(filter, fn aggregate, _ref ->
if aggregate.authorize? do
authorize_aggregate(
aggregate,
relationship_path_filters,
actor,
authorize?,
tenant,
tracer,
domain
)
else
aggregate
end
end)
else
filter
end
end
defp add_join_filters(
current_join_filters,
aggregate_relationship_path,
resource,
path_filters,
prefix \\ []
) do
aggregate_relationship_path
|> :lists.droplast()
|> Ash.Query.Aggregate.subpaths()
|> Enum.reduce(current_join_filters, fn path, current_join_filters ->
action =
resource
|> Ash.Resource.Info.related(path)
|> Ash.Resource.Info.primary_action!(:read)
|> Map.get(:name)
last_relationship = last_relationship(resource, prefix ++ path)
case Map.fetch(path_filters, {last_relationship.source, last_relationship.name, action}) do
{:ok, filter} ->
Map.update(current_join_filters, path, filter, fn current_filter ->
Ash.Query.BooleanExpression.new(:and, current_filter, filter)
end)
:error ->
current_join_filters
end
end)
end
defp maybe_await(%Task{} = task) do
case Task.await(task) do
{:__exception__, e, stacktrace} ->
reraise e, stacktrace
other ->
other
end
end
defp maybe_await(other), do: other
defp fetch_count(
%{action: action, resource: resource} = query,
query_before_pagination,
relationship_path_filters,
opts
) do
if action.pagination &&
opts[:page] &&
(opts[:page][:count] == true ||
(opts[:page][:count] != false and action.pagination.countable == :by_default)) do
with {:ok, filter} <-
filter_with_related(
query,
opts[:authorize?],
relationship_path_filters
),
query <-
query
|> Ash.Query.unset([:sort, :distinct_sort, :lock, :load, :limit, :offset])
|> Ash.Query.limit(query_before_pagination.limit)
|> Ash.Query.offset(query_before_pagination.offset)
|> Map.put(:filter, filter),
{:ok, data_layer_query} <- Ash.Query.data_layer_query(query) do
if Ash.DataLayer.in_transaction?(resource) || !Ash.DataLayer.can?(:async_engine, resource) do
case do_fetch_count(query, data_layer_query) do
{:ok, count} -> {:ok, {:ok, count}}
{:error, error} -> {:error, error}
end
else
{:ok,
Ash.ProcessHelpers.async(
fn ->
do_fetch_count(query, data_layer_query)
end,
opts
)}
end
end
else
{:ok, {:ok, nil}}
end
end
defp run_before_action(query) do
query =
query
|> Ash.Query.put_context(:private, %{in_before_action?: true})
|> set_phase(:before_action)
query.before_action
|> Enum.reduce_while({query, []}, fn before_action, {query, notifications} ->
case before_action.(query) do
{%{valid?: false} = query, new_notifications} ->
{:halt, {query, notifications ++ new_notifications}}
%{valid?: false} = query ->
{:halt, {query, notifications}}
{query, new_notifications} ->
{:cont, {query, notifications ++ new_notifications}}
query ->
{:cont, {query, notifications}}
end
end)
|> then(fn {query, notifications} -> {set_phase(query), notifications} end)
end
@doc false
def run_authorize_results(query, results) do
query = set_phase(query, :after_action)
query.authorize_results
|> Enum.reduce_while({query, {:ok, results}}, fn authorize_results, {query, {:ok, results}} ->
case authorize_results.(query, results) do
{:ok, results} ->
{:cont, {query, {:ok, results}}}
{:error, error} ->
{:halt, {query, {:error, error}}}
end
end)
|> elem(1)
end
defp run_after_action(query, results) do
query = set_phase(query, :after_action)
query.after_action
|> Enum.reduce_while({query, {:ok, results, []}}, fn after_action,
{query, {:ok, results, notifications}} ->
case after_action.(query, results) do
{:ok, results} ->
{:cont, {query, {:ok, results, notifications}}}
{:ok, results, new_notifications} ->
{:cont, {query, {:ok, results, notifications ++ new_notifications}}}
{:error, error} ->
{:halt, {query, {:error, error}}}
end
end)
|> elem(1)
end
defp add_field_level_auth(query, domain, opts) do
if opts[:authorize?] do
do_add_field_level_auth(query, domain, opts)
else
query
end
end
defp do_add_field_level_auth(query, domain, opts) do
data = %{
query: query,
changeset: nil,
domain: domain,
resource: query.resource,
action_input: nil
}
query.resource
|> Ash.Resource.Info.authorizers()
|> Enum.reduce(query, fn authorizer, query ->
state =
Ash.Authorizer.initial_state(
authorizer,
opts[:actor],
query.resource,
query.action
)
context = Ash.Authorizer.strict_check_context(authorizer, data)
case Ash.Authorizer.add_calculations(authorizer, query, state, context) do
{:ok, query, _} ->
query
{:error, error} ->
Ash.Query.add_error(query, error)
end
end)
end
defp do_fetch_count(query, data_layer_query) do
with {:ok, %{count: count}} <- run_count_query(query, data_layer_query) do
{:ok, count}
end
end
defp run_count_query(
%{
resource: destination_resource,
context: %{
data_layer: %{lateral_join_source: {root_data, path}}
},
action: %{
name: read_action
}
} = query,
data_layer_query
) do
case Ash.Query.Aggregate.new(destination_resource, :count, :count,
read_action: read_action,
query: Ash.Query.unset(query, :sort)
) do
{:ok, aggregate} ->
Ash.DataLayer.run_aggregate_query_with_lateral_join(
data_layer_query,
[aggregate],
root_data,
destination_resource,
path
)
{:error, error} ->
{:error, error}
end
end
defp run_count_query(query, data_layer_query) do
case Ash.Query.Aggregate.new(query.resource, :count, :count) do
{:ok, aggregate} ->
Ash.DataLayer.run_aggregate_query(data_layer_query, [aggregate], query.resource)
{:error, error} ->
{:error, error}
end
end
def page_opts(action, opts) do
cond do
action.pagination == false ->
nil
Keyword.keyword?(opts[:page]) && !Keyword.has_key?(opts[:page], :limit) &&
action.pagination.default_limit ->
Keyword.put(opts[:page], :limit, action.pagination.default_limit)
is_nil(opts[:page]) and action.pagination.required? ->
if action.pagination.default_limit do
[limit: action.pagination.default_limit]
else
opts[:page]
end
true ->
opts[:page]
end
end
@doc false
def paginate(starting_query, _action, _page_opts, true) do
{:ok, starting_query}
end
def paginate(starting_query, action, page_opts, _skip?) do
cond do
action.pagination == false && page_opts ->
{:error, "Pagination is not supported"}
action.pagination == false ->
{:ok, starting_query}
page_opts == false ->
if action.pagination.required? do
{:error, PaginationRequired.exception([])}
else
{:ok, starting_query}
end
page_opts[:limit] ->
page_opts =
Keyword.put(
page_opts || [],
:limit,
page_opts[:limit] || action.pagination.default_limit
)
case do_paginate(starting_query, action.pagination, page_opts) do
{:ok, query} ->
{:ok, query}
{:error, error} ->
{:error, error}
end
action.pagination.required? ->
{:error, LimitRequired.exception([])}
true ->
{:ok, starting_query}
end
end
defp do_paginate(query, pagination, page_opts) do
# We want to make 100% sure that there is a stable sort at the end
# of the sort for pagination
query =
if Ash.Actions.Sort.sorting_on_identity?(query) do
query
else
Ash.Query.sort(query, Ash.Resource.Info.primary_key(query.resource))
end
paginated =
cond do
page_opts[:before] || page_opts[:after] ->
keyset_pagination(query, pagination, page_opts)
page_opts[:offset] ->
limit_offset_pagination(query, pagination, page_opts)
pagination.offset? && pagination.keyset? ->
keyset_pagination(query, pagination, page_opts)
pagination.offset? ->
limit_offset_pagination(query, pagination, page_opts)
true ->
keyset_pagination(query, pagination, page_opts)
end
case paginated do
{:ok, query} ->
if page_opts[:filter] do
{:ok, Ash.Query.filter(query, ^page_opts[:filter])}
else
{:ok, query}
end
{:error, error} ->
{:error, error}
end
end
defp keyset_pagination(query, pagination, opts) do
query =
load_and_select_sort(query)
limited = Ash.Query.limit(query, limit(query, opts[:limit], query.limit, pagination) + 1)
if opts[:before] || opts[:after] do
reversed =
if opts[:before] do
reversed_sort = Ash.Sort.reverse(limited.sort)
max_index = Enum.count(reversed_sort) - 1
inverted_sort_input_indices = Enum.map(query.sort_input_indices, &(max_index - &1))
limited
|> Ash.Query.unset(:sort)
|> Map.put(:sort, reversed_sort)
|> Map.put(:sort_input_indices, inverted_sort_input_indices)
else
limited
end
after_or_before =
if opts[:before] do
:before
else
:after
end
case Ash.Page.Keyset.filter(
query,
opts[:before] || opts[:after],
query.sort,
after_or_before
) do
{:ok, filter} ->
{:ok, Ash.Query.do_filter(reversed, filter)}
{:error, error} ->
{:error, error}
end
else
{:ok, limited}
end
end
defp load_and_select_sort(query) do
query.sort
|> Enum.map(fn
{%Ash.Query.Calculation{} = calc, _} ->
{:calc, calc}
{field, _} ->
cond do
Ash.Resource.Info.aggregate(query.resource, field) ->
{:agg, field}
Ash.Resource.Info.attribute(query.resource, field) ->
{:attr, field}
end
end)
|> Enum.reduce(query, fn
{:calc, %{load: nil} = calc}, query ->
Ash.Query.calculate(
query,
calc.name,
{calc.module, calc.opts},
calc.type,
calc.context.arguments,
calc.constraints,
calc.context
)
{:calc, calc}, query ->
Ash.Query.load(query, calc)
{:agg, field}, query ->
Ash.Query.load(query, field)
{:attr, field}, query ->
Ash.Query.ensure_selected(query, field)
end)
end
defp limit(query, page_size, query_limit, pagination) do
max_page_size = pagination && pagination.max_page_size
if query.context[:private][:bypass_max_page_size?] do
[page_size, query_limit]
|> Enum.filter(&is_integer/1)
|> Enum.min(fn -> max_page_size end)
else
[page_size, query_limit, max_page_size]
|> Enum.filter(&is_integer/1)
|> Enum.min()
end
end
defp limit_offset_pagination(query, pagination, opts) do
limited = Ash.Query.limit(query, limit(query, opts[:limit], query.limit, pagination) + 1)
if opts[:offset] do
{:ok, Ash.Query.offset(limited, opts[:offset])}
else
{:ok, limited}
end
end
defp run_query(
%{context: %{private: %{action_result: result}}} = query,
_data_layer_query,
_context,
load_attributes?
) do
result
|> Helpers.select(query)
|> Helpers.load_runtime_types(query, load_attributes?)
|> case do
{:ok, result} ->
Ash.load(result, query, domain: query.domain)
other ->
other
end
end
defp run_query(
%{
resource: destination_resource,
context: %{
data_layer: %{
lateral_join_source: {root_data, path}
}
}
} = query,
data_layer_query,
_context,
load_attributes?
) do
if query.limit == 0 do
{:ok, []}
else
data_layer_query
|> Ash.DataLayer.run_query_with_lateral_join(
root_data,
destination_resource,
path
)
|> Helpers.select(query)
|> Helpers.load_runtime_types(query, load_attributes?)
end
end
defp run_query(
%{action: %{manual: {mod, opts}}} = query,
data_layer_query,
context,
load_attributes?
) do
query
|> mod.read(data_layer_query, opts, context)
|> validate_manual_action_return_result!(query.resource, query.action)
|> Helpers.select(query)
|> Helpers.load_runtime_types(query, load_attributes?)
end
defp run_query(
%{resource: resource} = query,
data_layer_query,
_context,
load_attributes?
) do
if query.limit == 0 do
{:ok, []}
else
data_layer_query
|> Ash.DataLayer.run_query(resource)
|> Helpers.rollback_if_in_transaction(query.resource, query)
|> Helpers.select(query)
|> Helpers.load_runtime_types(query, load_attributes?)
end
end
defp validate_manual_action_return_result!({:ok, list} = result, _resource, _)
when is_list(list) do
result
end
defp validate_manual_action_return_result!({:error, _error} = result, _resource, _) do
result
end
defp validate_manual_action_return_result!(other, resource, action) do
raise Ash.Error.Framework.AssumptionFailed,
message: """
Manual action #{inspect(action.name)} on #{inspect(resource)} returned an invalid result.
Expected one of the following:
* {:ok, [list, of, results]}
* {:error, error}
Got:
#{inspect(other)}
"""
end
defp hydrate_calculations(
query,
calculations_to_add
) do
Enum.reduce_while(calculations_to_add, {:ok, []}, fn calculation, {:ok, calculations} ->
if Ash.DataLayer.data_layer_can?(query.resource, :expression_calculation) do
expression = calculation.module.expression(calculation.opts, calculation.context)
expression =
Ash.Expr.fill_template(
expression,
calculation.context.actor,
calculation.context.arguments,
calculation.context.source_context
)
case Ash.Filter.hydrate_refs(expression, %{
resource: query.resource,
public?: false
}) do
{:ok, expression} ->
{:cont, {:ok, [{calculation, expression} | calculations]}}
{:error, error} ->
{:halt, {:error, error}}
end
else
{:halt,
{:error,
"Expression calculations are not supported by #{Ash.Resource.Info.data_layer(query.resource)}"}}
end
end)
end
defp authorize_calculation_expressions(
hydrated_calculations,
resource,
authorize?,
relationship_path_filters,
actor,
tenant,
tracer,
domain
) do
Enum.map(hydrated_calculations, fn {calculation, expression} ->
{calculation,
update_aggregate_filters(
expression,
resource,
authorize?,
relationship_path_filters,
actor,
tenant,
tracer,
domain
)}
end)
end
defp authorize_loaded_aggregates(query, path_filters, actor, authorize?, tenant, tracer) do
Enum.reduce(query.aggregates, query, fn {name, aggregate}, query ->
aggregate =
if authorize? && aggregate.authorize? do
authorize_aggregate(
aggregate,
path_filters,
actor,
authorize?,
tenant,
tracer,
query.domain
)
else
add_calc_context(aggregate, actor, authorize?, tenant, tracer, query.domain)
end
%{query | aggregates: Map.put(query.aggregates, name, aggregate)}
end)
end
defp authorize_aggregate(aggregate, path_filters, actor, authorize?, tenant, tracer, domain) do
aggregate = add_calc_context(aggregate, actor, authorize?, tenant, tracer, domain)
last_relationship = last_relationship(aggregate.resource, aggregate.relationship_path)
additional_filter =
case Map.fetch(
path_filters,
{last_relationship.source, last_relationship.name, aggregate.query.action.name}
) do
:error ->
true
{:ok, filter} ->
filter
end
with {:ok, filter} <-
filter_with_related(aggregate.query, authorize?, path_filters),
filter =
update_aggregate_filters(
filter,
aggregate.query.resource,
authorize?,
path_filters,
actor,
tenant,
tracer,
domain
),
{:ok, field} <-
aggregate_field_with_related_filters(
aggregate,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain
) do
%{
aggregate
| query: Ash.Query.filter(%{aggregate.query | filter: filter}, ^additional_filter),
field: field,
join_filters:
add_join_filters(
aggregate.join_filters,
aggregate.relationship_path,
aggregate.resource,
path_filters
)
}
else
{:error, error} ->
raise "Error processing aggregate authorization filter for #{inspect(aggregate)}: #{inspect(error)}"
end
end
defp aggregate_field_with_related_filters(
%{field: nil},
_path_filters,
_actor,
_authorize?,
_tenant,
_tracer,
_domain
),
do: {:ok, nil}
defp aggregate_field_with_related_filters(
%{field: %Ash.Query.Calculation{} = field} = agg,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain
) do
calc = add_calc_context(field, actor, authorize?, tenant, tracer, domain)
related_resource = Ash.Resource.Info.related(agg.resource, agg.relationship_path)
if calc.module.has_expression?() do
expr =
case calc.module.expression(calc.opts, calc.context) do
%Ash.Query.Function.Type{} = expr ->
expr
expr ->
{:ok, expr} = Ash.Query.Function.Type.new([expr, calc.type, calc.constraints])
expr
end
{:ok, expr} =
Ash.Filter.hydrate_refs(
expr,
%{
resource: related_resource,
public?: false
}
)
expr =
add_calc_context_to_filter(
expr,
actor,
authorize?,
tenant,
tracer,
domain
)
case do_filter_with_related(related_resource, expr, path_filters, []) do
{:ok, expr} ->
{:ok, %{field | module: Ash.Resource.Calculation.Expression, opts: [expr: expr]}}
{:error, error} ->
{:error, error}
end
else
{:ok, calc}
end
end
defp aggregate_field_with_related_filters(
%{field: %Ash.Query.Aggregate{} = field} = agg,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain
) do
field = add_calc_context(field, actor, authorize?, tenant, tracer, domain)
if authorize? && field.authorize? do
authorize_aggregate(field, path_filters, actor, authorize?, tenant, tracer, domain)
else
{:ok, agg}
end
end
defp aggregate_field_with_related_filters(
aggregate,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain
)
when is_atom(aggregate.field) do
related_resource = Ash.Resource.Info.related(aggregate.resource, aggregate.relationship_path)
case Ash.Resource.Info.field(related_resource, aggregate.field) do
%Ash.Resource.Calculation{} = resource_calculation ->
{module, opts} = resource_calculation.calculation
case Ash.Query.Calculation.new(
resource_calculation.name,
module,
opts,
resource_calculation.type,
resource_calculation.constraints,
filterable?: resource_calculation.filterable?,
sortable?: resource_calculation.sortable?,
sensitive?: resource_calculation.sensitive?,
load: resource_calculation.load
) do
{:ok, calculation} ->
aggregate_field_with_related_filters(
%{aggregate | field: calculation},
path_filters,
actor,
authorize?,
tenant,
tracer,
domain
)
{:error, error} ->
{:error, error}
end
%Ash.Resource.Aggregate{} = resource_aggregate ->
read_action =
resource_aggregate.read_action ||
Ash.Resource.Info.primary_action!(related_resource, :read).name
with %{valid?: true} = aggregate_query <-
Ash.Query.for_read(related_resource, read_action),
%{valid?: true} = aggregate_query <-
Ash.Query.Aggregate.build_query(aggregate_query,
filter: aggregate.filter,
sort: aggregate.sort
),
{:ok, query_aggregate} <-
Ash.Query.Aggregate.new(
related_resource,
resource_aggregate.name,
resource_aggregate.kind,
path: resource_aggregate.relationship_path,
query: aggregate_query,
field: resource_aggregate.field,
default: resource_aggregate.default,
filterable?: resource_aggregate.filterable?,
type: resource_aggregate.type,
constraints: resource_aggregate.constraints,
implementation: resource_aggregate.implementation,
uniq?: resource_aggregate.uniq?,
read_action: read_action,
authorize?: resource_aggregate.authorize?,
join_filters:
Map.new(resource_aggregate.join_filters, &{&1.relationship_path, &1.filter})
) do
aggregate_field_with_related_filters(
%{aggregate | field: query_aggregate},
path_filters,
actor,
authorize?,
tenant,
tracer,
domain
)
else
{:error, error} ->
{:error, error}
%{errors: errors} ->
{:error, errors}
end
_ ->
{:ok, aggregate.field}
end
end
defp aggregate_field_with_related_filters(
aggregate,
_path_filters,
_actor,
_authorize?,
_tenant,
_tracer,
_domain
)
when is_atom(aggregate.field) do
{:ok, aggregate.field}
end
defp filter_with_related(
query,
authorize?,
path_filters
) do
if authorize? do
do_filter_with_related(query.resource, query.filter, path_filters, [])
else
{:ok, query.filter}
end
end
defp do_filter_with_related(
resource,
%Ash.Filter{expression: expression} = filter,
path_filters,
prefix
) do
case do_filter_with_related(
resource,
expression,
path_filters,
prefix
) do
{:ok, new_expr} ->
{:ok, %{filter | expression: new_expr}}
other ->
other
end
end
defp do_filter_with_related(
resource,
%Ash.Query.BooleanExpression{op: :or, left: left, right: right},
path_filters,
prefix
) do
with {:ok, left} <- do_filter_with_related(resource, left, path_filters, prefix),
{:ok, right} <- do_filter_with_related(resource, right, path_filters, prefix) do
{:ok, Ash.Query.BooleanExpression.optimized_new(:or, left, right)}
end
end
defp do_filter_with_related(resource, filter_expr, path_filters, prefix) do
paths_to_global_filter_on =
filter_expr
|> Ash.Filter.list_refs()
|> Enum.filter(&(&1.input? && &1.relationship_path != []))
|> Enum.map(& &1.relationship_path)
|> Enum.uniq()
paths_to_global_filter_on
|> Enum.reduce_while(
{:ok, filter_expr},
fn path, {:ok, filter} ->
last_relationship =
Enum.reduce(path, nil, fn
relationship, nil ->
Ash.Resource.Info.relationship(resource, relationship)
relationship, acc ->
Ash.Resource.Info.relationship(acc.destination, relationship)
end)
read_action =
last_relationship.read_action ||
Ash.Resource.Info.primary_action!(last_relationship.destination, :read).name
case Map.get(
path_filters,
{last_relationship.source, last_relationship.name, read_action}
) do
nil ->
{:cont, {:ok, filter}}
%Ash.Filter{expression: authorization_filter} ->
{:cont,
{:ok,
Ash.Query.BooleanExpression.optimized_new(
:and,
filter_expr,
Ash.Filter.move_to_relationship_path(authorization_filter, path)
)}}
end
end
)
|> case do
{:ok, filter} ->
{:ok,
filter
|> Ash.Filter.map(fn
%Ash.Query.Exists{at_path: at_path, path: exists_path, expr: exists_expr} = exists ->
{:ok, new_expr} =
do_filter_with_related(
resource,
exists_expr,
path_filters,
prefix ++ at_path ++ exists_path
)
{:halt, %{exists | expr: new_expr}}
other ->
other
end)}
other ->
other
end
end
defp last_relationship(resource, list) do
path = :lists.droplast(list)
last = List.last(list)
Ash.Resource.Info.relationship(Ash.Resource.Info.related(resource, path), last)
end
defp set_phase(query, phase \\ :preparing)
when phase in ~w[preparing before_action after_action executing around_transaction]a,
do: %{query | phase: phase}
end