Packages
ash
3.9.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
# SPDX-FileCopyrightText: 2019 ash contributors <https://github.com/ash-project/ash/graphs.contributors>
#
# SPDX-License-Identifier: MIT
defmodule Ash.Actions.Read do
@moduledoc false
alias Ash.Actions.Helpers
alias Ash.Error.Invalid.{LimitRequired, NonCountableAction, 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
def read_and_return_unpaged(query, action \\ nil, opts \\ []) do
run(query, action, Keyword.put(opts, :return_unpaged?, 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] || Ash.Resource.Info.domain(query.resource)
if !domain do
raise Ash.Error.Framework.AssumptionFailed, message: "got a query without a domain"
end
query =
if opts[:initial_data] do
Ash.Query.set_context(query, %{query_for: :load})
else
query
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)
try do
tracer =
if opts[:initial_data] do
nil
else
opts[:tracer]
end
Ash.Tracer.span :action,
fn ->
Ash.Domain.Info.span_name(query.domain, query.resource, action.name)
end,
tracer do
metadata = fn ->
%{
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?]
}
end
Ash.Tracer.telemetry_span [:ash, Ash.Domain.Info.short_name(query.domain), :read],
metadata,
skip?: !!opts[:initial_data] do
Ash.Tracer.set_metadata(tracer, :action, metadata)
run_around_transaction_hooks(query, fn query ->
case do_run(query, action, opts) do
{:error, error} ->
error =
Ash.Error.to_error_class(
error,
bread_crumbs: "Error returned from: #{inspect(query.resource)}.#{action.name}"
)
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
rescue
e ->
reraise Ash.Error.to_error_class(e,
query: query,
stacktrace: __STACKTRACE__,
bread_crumbs: [
"Exception raised in: #{inspect(query.resource)}.#{action.name}"
]
),
__STACKTRACE__
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 run_before_transaction_hooks(%{before_transaction: []} = query) do
{:ok, query}
end
defp run_before_transaction_hooks(query) do
Enum.reduce_while(
query.before_transaction,
{:ok, query},
fn before_transaction, {:ok, query} ->
tracer = query.context[:private][:tracer]
metadata = fn ->
%{
domain: query.domain,
resource: query.resource,
resource_short_name: Ash.Resource.Info.short_name(query.resource),
actor: query.context[:private][:actor],
tenant: query.tenant,
action: query.action && query.action.name,
authorize?: query.context[:private][:authorize?]
}
end
result =
Ash.Tracer.span :before_transaction,
"before_transaction",
tracer do
Ash.Tracer.set_metadata(tracer, :before_transaction, metadata)
Ash.Tracer.telemetry_span [:ash, :before_transaction], metadata do
before_transaction.(query)
end
end
case result do
%Ash.Query{} = query ->
{:cont, {:ok, query}}
{:error, error} ->
{:halt, {:error, error}}
other ->
raise """
Invalid return value from before_transaction hook. Expected one of:
* %Ash.Query{}
* {:error, error}
Got:
#{inspect(other)}
"""
end
end
)
end
defp run_after_transaction_hooks(result, %{after_transaction: []} = _query) do
result
end
defp run_after_transaction_hooks(result, query) do
query.after_transaction
|> Enum.reduce(
result,
fn after_transaction, result ->
tracer = query.context[:private][:tracer]
metadata = fn ->
%{
domain: query.domain,
resource: query.resource,
resource_short_name: Ash.Resource.Info.short_name(query.resource),
actor: query.context[:private][:actor],
tenant: query.tenant,
action: query.action && query.action.name,
authorize?: query.context[:private][:authorize?]
}
end
Ash.Tracer.span :after_transaction,
"after_transaction",
tracer do
Ash.Tracer.set_metadata(tracer, :after_transaction, metadata)
Ash.Tracer.telemetry_span [:ash, :after_transaction], metadata do
after_transaction.(query, result)
end
end
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 =
if action.type == :read do
for_read(
query,
action,
opts[:initial_data],
actor: opts[:actor],
authorize?: opts[:authorize?],
timeout: opts[:timeout],
tenant: opts[:tenant]
)
else
query
end
initial_query = query
query = add_field_level_auth(query, query.domain, opts)
query = %{
query
| timeout:
opts[:timeout] || query.timeout || (query.action && 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,
expand?: false,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
)
relationship? = Map.has_key?(query.context, :accessing_from)
page_opts =
if Keyword.has_key?(opts, :page) do
page_opts(action, opts[:page], relationship?)
else
page_opts(action, query.page, relationship?)
end
opts = Keyword.delete(opts, :page)
query = Ash.Query.page(query, page_opts)
query =
if opts[:initial_data] do
query
else
load_and_select_sort(query, page_opts)
end
query = add_relationship_count_aggregates(query)
pkey = Ash.Resource.Info.primary_key(query.resource)
# we should probably lazily check this
missing_pkeys? =
Enum.empty?(pkey) ||
(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)
case Ash.Actions.Read.Calculations.split_and_load_calculations(
query.domain,
query,
missing_pkeys?,
Keyword.fetch(opts, :initial_data),
reuse_values?,
opts[:authorize?]
) do
{:error, %Ash.Query{errors: errors} = query} ->
{:error, Ash.Error.to_error_class(errors, query: query)}
{:error, error} ->
{:error, Ash.Error.to_error_class(error)}
{:ok, calculations_in_query, calculations_at_runtime, query} ->
query =
add_calc_context_to_query(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
expand?: false,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
)
calculations_at_runtime =
Enum.map(
calculations_at_runtime,
&add_calc_context(
&1,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
query.resource,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
)
)
calculations_in_query =
Enum.map(
calculations_in_query,
&add_calc_context(
&1,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
query.resource,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
)
)
source_fields =
if !opts[:initial_data] do
source_fields(query)
end
query =
if opts[:initial_data] do
select =
source_fields(query, opts[:lazy?] && opts[:initial_data]) ++ (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)
end
{query, stop?} = add_async_limiter(query, calculations_at_runtime, opts)
if opts[:data_layer_query?] do
data_layer_query(
query,
calculations_at_runtime,
calculations_in_query,
source_fields,
opts
)
else
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,
source_fields,
opts
)
end
{data_result, query_ran} =
case data_result do
{:ok, _result, _count, _calculations_at_runtime, _calculations_in_query, query} =
data_result ->
{data_result, query}
{{:error, _} = data_result, query} ->
{data_result, query}
data_result ->
{data_result, query}
end
query = Ash.Query.set_context(query, %{shared: query_ran.context[:shared]})
with {:ok, data, count, calculations_at_runtime, calculations_in_query, new_query} <-
data_result,
data = add_tenant(data, new_query),
{:ok, data} <-
load_through_attributes(
data,
%{query_ran | calculations: Map.new(calculations_in_query, &{&1.name, &1})},
query.domain,
opts[:actor],
opts[:tracer],
opts[:authorize?]
),
{:ok, data} <-
load_relationships(data, query, opts),
{: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
data
|> Helpers.restrict_field_access(query)
|> add_tenant(new_query)
|> attach_fields(opts[:initial_data], initial_query, query, missing_pkeys?)
|> cleanup_field_auth(query)
|> add_page(
query.action,
count,
query.sort,
query,
new_query,
opts
)
|> add_query(query, opts)
else
{:error, %Ash.Query{errors: errors} = query} ->
{:error, Ash.Error.to_error_class(errors, query: query)}
{:error,
%Ash.Error.Forbidden.Placeholder{
authorizer: authorizer
}} ->
error =
Ash.Authorizer.exception(
authorizer,
:forbidden,
query_ran.context[:private][:authorizer_state][authorizer]
)
{:error, Ash.Error.to_error_class(error)}
{: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
end
end
defp load_relationships(data, query, opts) do
lazy? = !!opts[:lazy?]
reuse_values? = !!opts[:reuse_values?]
context =
%{
actor: opts[:actor],
tenant: query.tenant,
authorize?: opts[:authorize?],
domain: query.domain
}
if query.load in [[], nil] do
{:ok, data}
else
case query.action.manual do
{module, opts} ->
if module.has_load_relationships?() do
module.load_relationships(query, data, opts, context, lazy?)
else
Ash.Actions.Read.Relationships.load(
data,
query,
lazy?,
reuse_values?
)
end
_ ->
Ash.Actions.Read.Relationships.load(
data,
query,
lazy?,
reuse_values?
)
end
end
end
defp do_read(
%{action: action} = query,
calculations_at_runtime,
calculations_in_query,
source_fields,
opts
) do
with {:ok, %{valid?: true} = query} <- handle_multitenancy(query),
query <- add_select_if_none_exists(query),
pre_authorization_query <- query,
{:ok, query} <- authorize_query(query, opts),
{:ok, sort} <-
add_calc_context_to_sort(
query.sort,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.resource,
query.domain,
parent_stack: parent_stack_from_context(query.context),
first_combination: Enum.at(query.combination_of, 0),
source_context: query.context
),
query <- %{
query
| filter:
add_calc_context_to_filter(
query.filter,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
query.resource,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
),
sort: sort
} do
case run_before_transaction_hooks(query) do
{:ok, query} ->
# Update calculation contexts with any context changes from before_transaction hooks
query =
add_calc_context_to_query(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
expand?: false,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
)
calculations_at_runtime =
Enum.map(calculations_at_runtime, fn calc ->
add_calc_context(
calc,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
query.resource,
source_context: query.context
)
end)
calculations_in_query =
Enum.map(calculations_in_query, fn calc ->
add_calc_context(
calc,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
query.resource,
source_context: query.context
)
end)
maybe_in_transaction(query, opts, fn notify_callback ->
with query_before_pagination <- query,
{query, calculations_at_runtime, calculations_in_query} <-
Ash.Actions.Read.Calculations.deselect_known_forbidden_fields(
query,
calculations_at_runtime,
calculations_in_query,
source_fields
),
{:ok, data_layer_calculations} <-
hydrate_calculations(query, calculations_in_query),
{:ok, query} <- hydrate_aggregates(query),
{:ok, query} <-
hydrate_sort(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain
),
{:ok, query} <-
hydrate_combinations(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
query.resource,
expand?: true,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
),
{: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,
parent_stack_from_context(query.context),
query.context
),
query <-
authorize_loaded_aggregates(
query,
relationship_path_filters,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer]
),
query <-
authorize_sorts(
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,
query.resource,
expand?: true,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
),
filter <-
update_aggregate_filters(
filter,
query.resource,
opts[:authorize?],
relationship_path_filters,
opts[:actor],
query.tenant,
opts[:tracer],
query.domain,
parent_stack_from_context(query.context),
query.context
),
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,
true
) do
ensure_task_stopped(count, fn ->
with {:ok, query} <- paginate(query, action, opts[:skip_pagination?]),
:ok <- validate_combinations(query, calculations_at_runtime, query.load),
{:ok, data_layer_query} <-
Ash.Query.data_layer_query(query,
data_layer_calculations: data_layer_calculations
),
{{:ok, results}, query} <-
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, query.timeout) do
notify_callback.(query, before_notifications ++ after_notifications)
with {:ok, results} <- run_after_transaction_hooks({:ok, results}, query) do
{:ok, results, count, calculations_at_runtime, calculations_in_query, query}
end
else
other ->
other
|> handle_failed_query(notify_callback)
|> run_after_transaction_hooks(query)
end
end)
else
other ->
other
|> handle_failed_query(notify_callback)
|> run_after_transaction_hooks(query)
end
end)
{:error, error} ->
error_result = {:error, error}
run_after_transaction_hooks(error_result, query)
end
else
{:ok, query} ->
{{:error, query}, query}
{:error, error} ->
{{:error, error}, query}
end
end
defp ensure_task_stopped(%Task{} = task, fun) do
fun.()
after
Task.shutdown(task, :brutal_kill)
end
defp ensure_task_stopped(_, fun), do: fun.()
defp handle_failed_query(result, notify_callback) do
case result do
{%{valid?: false} = query, before_notifications} ->
notify_callback.(query, before_notifications)
{:error, query}
{{:error, %Ash.Query{} = query}, _} ->
{:error, query}
{{:error, error}, query} ->
{:error, Ash.Query.add_error(query, error)}
{:ok, %Ash.Query{valid?: false} = query} ->
{:error, query}
%Ash.Query{} = query ->
{:error, query}
{:error, error} ->
{:error, error}
end
end
defp data_layer_query(
%{action: action} = query,
calculations_at_runtime,
calculations_in_query,
source_fields,
opts
) do
initial_query = query
with {:ok, %{valid?: true} = query} <- handle_multitenancy(query),
query <- add_select_if_none_exists(query),
pre_authorization_query <- query,
{:ok, query} <- authorize_query(query, opts),
{:ok, sort} <-
add_calc_context_to_sort(
query.sort,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.resource,
query.domain,
parent_stack: parent_stack_from_context(query.context),
first_combination: Enum.at(query.combination_of, 0),
source_context: query.context
),
query <- %{
query
| filter:
add_calc_context_to_filter(
query.filter,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
query.resource,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
),
sort: sort
},
query_before_pagination <- query,
{query, calculations_at_runtime, calculations_in_query} <-
Ash.Actions.Read.Calculations.deselect_known_forbidden_fields(
query,
calculations_at_runtime,
calculations_in_query,
source_fields
),
{:ok, data_layer_calculations} <- hydrate_calculations(query, calculations_in_query),
{:ok, query} <- hydrate_aggregates(query),
{:ok, query} <-
hydrate_sort(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain
),
{:ok, query} <-
hydrate_combinations(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
query.resource,
expand?: true,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
),
{: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,
parent_stack_from_context(query.context),
query.context
),
query <-
authorize_loaded_aggregates(
query,
relationship_path_filters,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer]
),
query <-
authorize_sorts(
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,
query.resource,
expand?: true,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
),
filter <-
update_aggregate_filters(
filter,
query.resource,
opts[:authorize?],
relationship_path_filters,
opts[:actor],
query.tenant,
opts[:tracer],
query.domain,
parent_stack_from_context(query.context),
query.context
),
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,
false
) do
ensure_task_stopped(count, fn ->
with {:ok, query} <- paginate(query, action, opts[:skip_pagination?]),
:ok <- validate_combinations(query, calculations_at_runtime, query.load),
{:ok, data_layer_query} <-
Ash.Query.data_layer_query(query, data_layer_calculations: data_layer_calculations) do
{:ok,
%{
query: data_layer_query,
ash_query: query,
load: fn query_ran, data ->
with {:ok, data} <-
load_through_attributes(
data,
%{
query_ran
| calculations: Map.new(calculations_in_query, &{&1.name, &1})
},
query.domain,
opts[:actor],
opts[:tracer],
opts[:authorize?]
),
{:ok, data} <-
load_relationships(data, query, opts),
{: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
data
|> Helpers.restrict_field_access(query)
|> add_tenant(query)
|> attach_fields(nil, initial_query, query, false)
|> cleanup_field_auth(query)
|> add_page(
query.action,
count,
query.sort,
initial_query,
query,
opts
)
else
{:error, %Ash.Query{errors: errors} = query} ->
{:error, Ash.Error.to_error_class(errors, query: query)}
{:error,
%Ash.Error.Forbidden.Placeholder{
authorizer: authorizer
}} ->
error =
Ash.Authorizer.exception(
authorizer,
:forbidden,
query_ran.context[:private][:authorizer_state][authorizer]
)
{:error, Ash.Error.to_error_class(error)}
{:error, error} ->
{:error, Ash.Error.to_error_class(error, query: query)}
end
end,
run: fn data_layer_query ->
notify? = !Process.put(:ash_started_transaction?, true)
with {{:ok, results}, query} <-
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) do
notify_or_store(query, before_notifications ++ after_notifications, notify?)
{:ok, add_tenant(results, query)}
else
{%{valid?: false} = query, before_notifications} ->
notify_or_store(query, before_notifications, notify?)
{:error, Ash.Error.to_ash_error(query)}
{{:error, %Ash.Query{} = query}, _} ->
{:error, Ash.Error.to_ash_error(query)}
{{:error, error}, query} ->
{:error, Ash.Error.to_ash_error(Ash.Query.add_error(query, error))}
{:ok, %Ash.Query{valid?: false} = query} ->
{:error, Ash.Error.to_ash_error(query)}
%Ash.Query{} = query ->
{:error, Ash.Error.to_ash_error(query)}
{:error, error} ->
{:error, Ash.Error.to_ash_error(error)}
end
end,
count: fn -> count.() end
}}
else
{:ok, query} ->
{{:error, query}, query}
{:error, error} ->
{{:error, error}, query}
end
end)
else
{:ok, query} ->
{{:error, query}, query}
{:error, error} ->
{{:error, error}, query}
end
end
defp validate_combinations(%{combination_of: []}, _, _) do
:ok
end
defp validate_combinations(query, runtime_calculations, load) do
case query.combination_of do
[%{type: type} | _] when type != :base ->
{:error, "Invalid combinations. The first combination must have type `:base`."}
combination_of ->
default_select =
MapSet.to_list(Ash.Resource.Info.selected_by_default_attribute_names(query.resource))
fieldsets =
Enum.map(
combination_of,
&Enum.sort(Enum.uniq((&1.select || default_select) ++ Map.keys(&1.calculations)))
)
case Enum.uniq(fieldsets) do
[fieldset] ->
if requires_pkey_but_not_selecting_it(
query,
fieldset,
default_select,
runtime_calculations,
load
) do
:ok
else
{:error,
"When runtime calculations, loads, or more attributes in the parent select but not the combination fieldsets are present, all fieldsets must contain the primary key of the resource."}
end
_ ->
{:error,
"""
Invalid combinations. All fieldsets must be the same, got the following:
#{inspect(query.combination_of)}
"""}
end
end
end
defp requires_pkey_but_not_selecting_it(
query,
fieldset,
default_select,
runtime_calculations,
load
) do
Enum.all?(query.select || default_select, &(&1 in fieldset)) ||
(Enum.empty?(runtime_calculations) && Enum.empty?(load)) ||
Enum.all?(
Ash.Resource.Info.primary_key(query.resource),
&Enum.member?(fieldset, &1)
)
end
defp add_relationship_count_aggregates(query) do
Enum.reduce(query.load, query, fn {relationship_name, related_query}, query ->
relationship = Ash.Resource.Info.relationship(query.resource, relationship_name)
related_query =
case related_query do
[] -> Ash.Query.new(relationship.destination)
query -> query
end
needs_count? = related_query.page && related_query.page[:count] == true
if needs_count? do
related_query =
Ash.Query.unset(related_query, [
:sort,
:distinct,
:distinct_sort,
:lock,
:load,
:page,
:aggregates
])
aggregate_name = paginated_relationship_count_aggregate_name(relationship.name)
query
|> Ash.Query.aggregate(aggregate_name, :count, relationship.name,
query: related_query,
default: 0
)
else
query
end
end)
end
@doc false
def paginated_relationship_count_aggregate_name(relationship_name) do
"__paginated_#{relationship_name}_count__"
end
@doc false
def cleanup_field_auth(records, query, top_level? \\ true)
def cleanup_field_auth(
records,
%{context: %{private: %{loading_relationship?: true}}},
_top_level?
) do
records
end
def cleanup_field_auth(nil, _query, _top_level?), do: nil
def cleanup_field_auth([], _query, _top_level?), do: []
def cleanup_field_auth(%Ash.NotLoaded{} = not_loaded, _query, _top_level?), do: not_loaded
def cleanup_field_auth([%resource{} | _] = records, [], top_level?),
do: cleanup_field_auth(records, resource |> Ash.Query.new(), top_level?)
def cleanup_field_auth(%struct{results: results} = page, query, top_level?)
when struct in [Ash.Page.Keyset, Ash.Page.Offset] do
%{page | results: cleanup_field_auth(results, query, top_level?)}
end
def cleanup_field_auth(records, %{resource: resource} = query, top_level?)
when is_list(records) do
records =
if top_level? do
records
else
query =
Ash.Query.set_context(query, %{private: %{cleaning_up_field_auth?: true}})
Helpers.restrict_field_access(records, query)
end
records =
Enum.reduce(query.load, records, fn {name, related_query}, records ->
Enum.map(records, fn record ->
case record do
%Ash.ForbiddenField{} -> record
%Ash.NotLoaded{} -> record
nil -> record
record -> Map.update!(record, name, &cleanup_field_auth(&1, related_query, false))
end
end)
end)
records =
Enum.reduce(
query.load_through[:attribute] || %{},
records,
fn {attr_name, further_load}, records ->
attribute = Ash.Resource.Info.attribute(resource, attr_name)
Enum.map(records, fn record ->
case record do
%Ash.ForbiddenField{} ->
record
%Ash.NotLoaded{} ->
record
nil ->
record
record ->
Map.update!(record, attribute.name, fn value ->
Ash.Type.rewrite(
attribute.type,
value,
[{:cleanup_field_auth, further_load}],
attribute.constraints
)
end)
end
end)
end
)
Enum.reduce(
query.load_through[:calculation] || %{},
records,
fn {calc_name, further_load}, records ->
Enum.map(records, fn record ->
case Map.get(query.calculations, calc_name) do
%{load: load, type: type, constraints: constraints} when not is_nil(load) ->
case record do
%Ash.ForbiddenField{} ->
record
%Ash.NotLoaded{} ->
record
nil ->
record
record ->
Map.update!(record, load, fn value ->
Ash.Type.rewrite(
type,
value,
[{:cleanup_field_auth, further_load}],
constraints
)
end)
end
%{load: nil, name: name, type: type, constraints: constraints} ->
case record do
%Ash.ForbiddenField{} ->
record
%Ash.NotLoaded{} ->
record
nil ->
record
record ->
if Map.has_key?(record.calculations, name) do
Map.update!(record, :calculations, fn calculations ->
Map.update!(calculations, name, fn value ->
Ash.Type.rewrite(
type,
value,
[{:cleanup_field_auth, further_load}],
constraints
)
end)
end)
else
record
end
end
nil ->
record
end
end)
end
)
end
def cleanup_field_auth(record, query, top_level?) do
record
|> List.wrap()
|> cleanup_field_auth(query, top_level?)
|> Enum.at(0)
end
defp agg_refs(query, calculations_in_query) do
calculations_in_query
|> Enum.flat_map(fn {_, expr} ->
expr
|> Ash.Filter.hydrate_refs(%{
resource: query.resource,
public?: false,
first_combination: Enum.at(query.combination_of, 0),
parent_stack: parent_stack_from_context(query.context)
})
|> case do
{:ok, expr} ->
expr
|> Ash.Filter.used_aggregates(:*, true)
|> Enum.map(fn %Ash.Query.Ref{
attribute: aggregate,
relationship_path: relationship_path
} ->
%{
aggregate
| resource: query.resource,
relationship_path: relationship_path ++ aggregate.relationship_path,
related?: Map.get(aggregate, :related?, true)
}
end)
_ ->
[]
end
end)
|> Enum.concat(Map.values(query.aggregates))
end
defp source_fields(query, lazy_for_initial_data \\ nil) do
Enum.flat_map(query.load, fn {name, _} ->
if lazy_for_initial_data && Ash.Resource.loaded?(lazy_for_initial_data, name, lists: :any) do
[]
else
case Ash.Resource.Info.relationship(query.resource, name) do
%{no_attributes?: true} ->
[]
%{manual: {module, opts}, source_attribute: source_attribute} ->
fields =
module.select(opts)
[source_attribute | fields]
%{source_attribute: source_attribute} ->
[source_attribute]
end
end
end)
end
defp clear_async_limiter(%{context: %{private: %{async_limiter: async_limiter}}} = query)
when is_pid(async_limiter) do
put_in(query.context.private.async_limiter, nil)
end
defp clear_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? = !Process.put(:ash_started_transaction?, true)
try do
cond do
query.action.transaction? ->
Ash.DataLayer.transaction(
[query.resource | query.action.touches_resources],
fn ->
func.(¬ify_or_store(&1, &2, notify?))
end,
query.timeout,
%{
type: :read,
metadata: %{
query: query,
resource: query.resource,
action: query.action.name
},
data_layer_context: query.context[:data_layer]
},
rollback_on_error?: false
)
|> case do
{:error, :rollback} when not is_nil(query.timeout) ->
{:error,
Ash.Error.Invalid.Timeout.exception(
timeout: query.timeout,
name: fn -> "#{inspect(query.resource)}.#{query.action.name}" end
)}
{:error, {:error, error}} ->
{:error, error}
{:error, error} ->
{:error, error}
{:ok, result} ->
result
end
query.timeout ->
Ash.ProcessHelpers.task_with_timeout(
fn ->
func.(¬ify_or_store(&1, &2, notify?))
end,
query.resource,
query.timeout,
"#{inspect(query.resource)}.#{query.action.name}",
opts[:tracer]
)
true ->
func.(¬ify_or_store(&1, &2, notify?))
end
after
if notify? do
Process.delete(:ash_started_transaction?)
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
Ash.Query.select(
query,
Ash.Resource.Info.selected_by_default_attribute_names(query.resource)
)
end
end
defp load(
[first | _] = initial_data,
query,
calculations_at_runtime,
calculations_in_query,
missing_pkeys?,
opts
) do
must_be_reselected =
query.select
|> List.wrap()
|> Kernel.--(Ash.Resource.Info.primary_key(query.resource))
must_be_reselected =
if opts[:reuse_values?] && !missing_pkeys? do
must_be_reselected
|> Enum.reject(&Ash.Resource.selected?(first, &1, forbidden_means_selected?: true))
else
must_be_reselected
end
{query, calculations_at_runtime, calculations_in_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, 0, calculations_at_runtime, calculations_in_query, query}
else
reselect_and_load(
initial_data,
query,
must_be_reselected,
calculations_in_query,
calculations_at_runtime,
opts
)
end
end
defp reselect_and_load(
initial_data,
query,
must_be_reselected,
calculations_in_query,
calculations_at_runtime,
opts
) do
primary_key = Ash.Resource.Info.primary_key(query.resource)
if Enum.empty?(primary_key) do
{:error,
"""
Cannot reselect fields or perform query/runtime calculations for resources that do not have a primary key using `Ash.load`.
You must ensure that all necessary fields are present, and use the `reuse_values?` option, and compute calculations when
the query is run.
Attempted to reselect #{inspect(must_be_reselected)}.
Attempted to perform query calculations #{inspect(calculations_in_query)}.
Attempted to perform runtime calculations #{inspect(calculations_at_runtime)}.
"""}
else
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.select([])
|> 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, %{valid?: true} = query} <- handle_multitenancy(query),
{:ok, data_layer_calculations} <-
hydrate_calculations(
query,
calculations_in_query
),
{:ok, query} <- hydrate_aggregates(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,
parent_stack_from_context(query.context),
query.context
),
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}, query} <-
run_query(
query,
data_layer_query,
%{
actor: opts[:actor],
tenant: query.tenant,
authorize?: opts[:authorize?],
domain: query.domain
},
true
) do
results
|> attach_fields(initial_data, query, query, false, true)
|> cleanup_field_auth(query)
|> compute_expression_at_runtime_for_missing_records(query, data_layer_calculations)
|> case do
{:ok, result} ->
{:ok, result, 0, calculations_at_runtime, calculations_in_query, query}
{:error, error} ->
{:error, error}
end
end
end
end
defp hydrate_combinations(
query,
actor,
authorize?,
tenant,
tracer,
domain,
_resource,
opts
) do
Enum.reduce_while(query.combination_of, {:ok, []}, fn combination, {:ok, acc} ->
{:ok, sort} =
add_calc_context_to_sort(
combination.sort,
actor,
authorize?,
tenant,
tracer,
query.resource,
domain,
opts
)
Enum.reduce_while(combination.calculations, {:ok, %{}}, fn {key, calc}, {:ok, acc} ->
case hydrate_calculations(query, [calc]) do
{:ok, [{calc, expression}]} ->
{:cont,
{:ok,
Map.put(acc, key, %{
calc
| module: Ash.Resource.Calculation.Expression,
opts: [expr: expression]
})}}
{:error, error} ->
{:halt, {:error, error}}
end
end)
|> case do
{:error, error} ->
{:halt, {:error, error}}
{:ok, calculations} ->
{:cont,
{:ok,
[
%{
combination
| filter:
add_calc_context_to_filter(
combination.filter,
actor,
authorize?,
tenant,
tracer,
domain,
query.resource,
opts
),
calculations:
Map.new(calculations, fn {key, val} ->
{key,
add_calc_context(
val,
actor,
authorize?,
tenant,
tracer,
domain,
query.resource,
opts
)}
end),
sort: sort
}
| acc
]}}
end
end)
|> case do
{:ok, combinations} -> {:ok, %{query | combination_of: Enum.reverse(combinations)}}
{:error, error} -> {:error, error}
end
end
defp hydrate_sort(query, actor, authorize?, tenant, tracer, domain) do
[:sort, :distinct, :distinct_sort]
|> Enum.reject(&(Map.get(query, &1) in [nil, []]))
|> Enum.reduce({:ok, query}, fn key, {:ok, query} ->
query
|> Map.get(key)
|> Enum.map(fn {field, direction} ->
if is_atom(field) do
case Ash.Resource.Info.field(query.resource, field) do
%Ash.Resource.Calculation{} = calc -> {calc, direction}
%Ash.Resource.Aggregate{} = agg -> {agg, direction}
_field -> {field, direction}
end
else
{field, direction}
end
end)
|> Enum.reduce_while({:ok, []}, fn
{%Ash.Resource.Calculation{} = resource_calculation, direction}, {:ok, sort} ->
case Ash.Query.Calculation.from_resource_calculation(
query.resource,
resource_calculation,
source_context: query.context
) do
{:ok, calc} ->
case hydrate_calculations(query, [calc]) do
{:ok, [{calc, expression}]} ->
{:cont,
{:ok,
[
{%{
calc
| module: Ash.Resource.Calculation.Expression,
opts: [expr: expression]
}, direction}
| sort
]}}
{:error, error} ->
{:halt, {:error, error}}
end
{:error, error} ->
{:halt, {:error, error}}
end
{%Ash.Query.Calculation{} = calc, direction}, {:ok, sort} ->
case hydrate_calculations(query, [calc]) do
{:ok, [{calc, expression}]} ->
{:cont,
{:ok,
[
{%{
calc
| module: Ash.Resource.Calculation.Expression,
opts: [expr: expression]
}, direction}
| sort
]}}
{:error, error} ->
{:halt, {:error, error}}
end
{%Ash.Resource.Aggregate{} = agg, direction}, {:ok, sort} ->
case query_aggregate_from_resource_aggregate(query, agg) do
{:ok, agg} -> {:cont, {:ok, [{agg, direction} | sort]}}
{:error, error} -> {:halt, {:error, error}}
end
{other, direction}, {:ok, sort} ->
{:cont, {:ok, [{other, direction} | sort]}}
end)
|> case do
{:ok, sort} ->
sort =
Enum.map(sort, fn {field, direction} ->
case field do
%struct{} = field
when struct in [
Ash.Query.Calculation,
Ash.Aggregate.Calculation,
Ash.Resource.Calculation,
Ash.Resource.Aggregate
] ->
{add_calc_context(
field,
actor,
authorize?,
tenant,
tracer,
domain,
query.resource,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
), direction}
field ->
{field, direction}
end
end)
{:ok, %{query | key => Enum.reverse(sort)}}
{:error, error} ->
{:error, error}
end
end)
end
defp hydrate_aggregates(query) do
Enum.reduce_while(query.aggregates, {:ok, %{}}, fn {key, aggregate}, {:ok, aggregates} ->
aggregate = %{
aggregate
| query: %{
aggregate.query
| filter:
Ash.Expr.fill_template(
aggregate.query.filter,
actor: query.context[:private][:actor],
tenant: query.to_tenant,
args: %{},
context: query.context
)
}
}
case hydrate_sort(
aggregate.query,
query.context[:private][:actor],
query.context[:private][:authorize?],
query.tenant,
query.context[:private][:tracer],
query.domain
) do
{:ok, agg_query} ->
aggregate = %{aggregate | query: agg_query}
case aggregate.field do
%Ash.Query.Calculation{} = calculation ->
if calculation.module.has_expression?() and
Ash.DataLayer.data_layer_can?(query.resource, :expression_calculation) do
expression = calculation.module.expression(calculation.opts, calculation.context)
expression =
Ash.Expr.fill_template(
expression,
actor: query.context[:private][:actor],
tenant: query.to_tenant,
args: calculation.context.arguments,
context: query.context
)
case Ash.Filter.hydrate_refs(expression, %{
resource:
Ash.Resource.Info.related(query.resource, aggregate.relationship_path),
public?: false,
parent_stack: parent_stack_from_context(query.context)
}) do
{:ok, expression} ->
new_field = %{
calculation
| module: Ash.Resource.Calculation.Expression,
opts: [expr: expression]
}
{:cont, {:ok, Map.put(aggregates, key, %{aggregate | field: new_field})}}
{:error, error} ->
{:halt, {:error, error}}
end
else
{:cont, {:ok, Map.put(aggregates, key, aggregate)}}
end
_other ->
{:cont, {:ok, Map.put(aggregates, key, aggregate)}}
end
{:error, error} ->
{:error, error}
end
end)
|> case do
{:ok, aggregates} ->
{:ok, %{query | aggregates: aggregates}}
{:error, error} ->
{:error, error}
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?] && !Enum.empty?(Ash.Resource.Info.authorizers(query.resource)) do
case Ash.can(query, opts[:actor],
return_forbidden_error?: true,
maybe_is: false,
pre_flight?: false,
filter_with: opts[:authorize_with] || :filter,
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 ->
loaded_keys =
calculations
|> Enum.filter(fn
{{:__calc_dep__, _}, _} ->
false
{_key, calc} ->
Ash.Resource.loaded?(initial_data, calc)
end)
|> Enum.map(&elem(&1, 0))
Map.drop(calculations, loaded_keys)
end)
|> Map.update!(:aggregates, fn aggregates ->
loaded_keys =
aggregates
|> Enum.filter(fn {_key, calc} ->
Ash.Resource.loaded?(initial_data, calc)
end)
|> Enum.map(&elem(&1, 0))
Map.drop(aggregates, loaded_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) && (is_nil(query.select) || name in query.select)
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.reject(fn {_, v} -> is_nil(v) end)
|> 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, type} ->
Enum.reduce_while(values, {:ok, []}, fn list, {:ok, acc} ->
case Ash.Type.load(
type,
list,
load_statement,
calculation.constraints[:items] || [],
%{
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.reject(fn {_, v} -> is_nil(v) end)
|> 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, type} ->
Enum.reduce_while(values, {:ok, []}, fn list, {:ok, acc} ->
case Ash.Type.load(
type,
list,
load_statement,
attribute.constraints[:items] || [],
%{
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 \\ nil,
resource \\ nil,
opts \\ []
) do
# backward compatibility fix
# really need to stop these from being called elsewhere
{domain, resource, opts} =
if is_list(domain) and resource == nil and opts == [] do
case filter do
%Ash.Filter{resource: resource} ->
{nil, resource, opts}
_ ->
{nil, nil, []}
end
else
{domain, resource, opts}
end
Ash.Filter.map(filter, fn
%Ash.Query.Parent{} = parent ->
if List.wrap(opts[:parent_stack]) != [] do
%{
parent
| expr:
add_calc_context_to_filter(
parent.expr,
actor,
authorize?,
tenant,
tracer,
domain,
hd(opts[:parent_stack]),
Keyword.update!(opts, :parent_stack, &tl/1)
)
}
else
parent
end
%Ash.Query.Exists{} = exists ->
if List.wrap(opts[:parent_stack]) != [] do
%{
exists
| expr:
add_calc_context_to_filter(
exists.expr,
actor,
authorize?,
tenant,
tracer,
domain,
Ash.Resource.Info.related(resource, exists.path),
Keyword.update(opts, :parent_stack, [resource], &[resource, &1])
)
}
else
exists
end
%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, resource, opts)
if Keyword.get(opts, :expand?, false) && calc.module.has_expression?() do
expr =
case calc.module.expression(calc.opts, calc.context) do
%Ash.Query.Function.Type{} = expr ->
expr
%Ash.Query.Call{name: :type} = expr ->
expr
expr ->
if calc.type do
{:ok, expr} =
Ash.Query.Function.Type.new([expr, calc.type, calc.constraints])
expr
else
expr
end
end
{:ok, expr} =
Ash.Filter.hydrate_refs(
expr,
%{
resource: ref.resource,
parent_stack: opts[:parent_stack] || [],
public?: false
}
)
expr =
Ash.Expr.fill_template(
expr,
actor: actor,
tenant: tenant,
args: calc.context.arguments,
context: opts[:source_context] || %{}
)
expr =
add_calc_context_to_filter(
expr,
actor,
authorize?,
tenant,
tracer,
domain,
ref.resource,
opts
)
expanded_calc = %Ash.Query.Calculation{
name: calc.name,
module: Ash.Resource.Calculation.Expression,
opts: [expr: expr],
type: calc.type,
constraints: calc.constraints,
filterable?: calc.filterable?,
sortable?: calc.sortable?,
sensitive?: calc.sensitive?,
load: calc.load,
select: calc.select,
context: calc.context
}
# Return a Ref at the relationship path with the expanded calculation
%{ref | attribute: expanded_calc, relationship_path: relationship_path}
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, ref.resource, opts)
}
other ->
other
end)
end
defp add_calc_context_to_sort(empty, _, _, _, _, _, _, _opts) when empty in [[], nil],
do: {:ok, empty}
defp add_calc_context_to_sort(sort, actor, authorize?, tenant, tracer, resource, domain, opts) do
sort
|> Enum.reduce_while({:ok, []}, fn
{%struct{} = calc, order}, {:ok, acc}
when struct in [
Ash.Query.Calculation,
Ash.Aggregate.Calculation,
Ash.Resource.Calculation,
Ash.Resource.Aggregate
] ->
calc = add_calc_context(calc, actor, authorize?, tenant, tracer, domain, resource, opts)
if should_expand_expression?(struct, calc, opts) do
case expand_expression(calc, resource, opts[:parent_stack], opts[:first_combination]) do
{:ok, expr} ->
args =
case calc do
%Ash.Query.Calculation{context: %{arguments: arguments}} -> arguments
_ -> %{}
end
expr =
Ash.Expr.fill_template(
expr,
actor: actor,
tenant: tenant,
args: args,
context: opts[:source_context]
)
expr =
add_calc_context_to_filter(
expr,
actor,
authorize?,
tenant,
tracer,
domain,
resource,
opts
)
module = Ash.Resource.Calculation.Expression
calc = %{calc | module: module, opts: [expr: expr]}
{:cont, {:ok, [{calc, order} | acc]}}
error ->
{:halt, error}
end
else
{:cont, {:ok, [{calc, order} | acc]}}
end
{calc, order}, {:ok, acc} ->
{:cont, {:ok, [{calc, order} | acc]}}
end)
|> then(fn
{:ok, sort} -> {:ok, Enum.reverse(sort)}
result -> result
end)
end
defp should_expand_expression?(struct, calc, opts) do
struct == Ash.Query.Calculation &&
Keyword.get(opts, :expand?, true) &&
calc.module.has_expression?()
end
defp expand_expression(calc, resource, parent_stack, first_combination) do
calc.module.expression(calc.opts, calc.context)
|> case do
%Ash.Query.Function.Type{} = expr ->
expr
%Ash.Query.Call{name: :type} = expr ->
expr
expr ->
if calc.type do
{:ok, expr} =
Ash.Query.Function.Type.new([expr, calc.type, calc.constraints])
expr
else
expr
end
end
|> Ash.Filter.hydrate_refs(%{
resource: resource,
public?: false,
parent_stack: parent_stack,
first_combination: first_combination
})
end
@doc false
def handle_multitenancy(query) do
action_multitenancy =
get_shared_multitenancy(query) || get_action(query.resource, query.action).multitenancy
case action_multitenancy do
:enforce ->
query = handle_attribute_multitenancy(query)
with :ok <- validate_multitenancy(query) do
{:ok, query}
end
:allow_global ->
{:ok, handle_attribute_multitenancy(query)}
:bypass ->
{:ok, query}
:bypass_all ->
query = Ash.Query.set_context(query, %{shared: %{multitenancy: :bypass_all}})
{:ok, query}
end
|> case do
{:ok, query} -> handle_aggregate_multitenancy(query)
other -> other
end
end
defp handle_attribute_multitenancy(query) do
if query.tenant && Ash.Resource.Info.multitenancy_strategy(query.resource) == :attribute do
multitenancy_attribute = Ash.Resource.Info.multitenancy_attribute(query.resource)
if multitenancy_attribute do
{m, f, a} = Ash.Resource.Info.multitenancy_parse_attribute(query.resource)
attribute_value = apply(m, f, [query.to_tenant | a])
Ash.Query.filter(query, ^ref(multitenancy_attribute) == ^attribute_value)
else
query
end
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
@doc false
def for_read(query, action, initial_data, args \\ %{}, opts) do
if query.__validated_for_action__ == action.name do
query
else
load = query.load
calculations = query.calculations
aggregates = query.aggregates
load_through = query.load_through
query = Ash.Query.for_read(query, action.name, args, opts)
if strip_load?(initial_data) do
%{
query
| load: load,
aggregates: aggregates,
calculations: calculations,
load_through: load_through
}
else
if prefer_existing_loads?(query) do
%{
query
| load: Keyword.merge(load, query.load),
aggregates: Map.merge(aggregates, query.aggregates),
calculations: Map.merge(calculations, query.calculations),
load_through: %{
calculation:
Map.merge(
load_through[:calculation] || %{},
query.load_through[:calculation] || %{}
),
attribute:
Map.merge(
load_through[:attribute] || %{},
query.load_through[:attribute] || %{}
)
}
}
else
%{
query
| load: Keyword.merge(query.load, load),
aggregates: Map.merge(query.aggregates, aggregates),
calculations: Map.merge(query.calculations, calculations),
load_through: %{
calculation:
Map.merge(
query.load_through[:calculation] || %{},
load_through[:calculation] || %{}
),
attribute:
Map.merge(
query.load_through[:attribute] || %{},
load_through[:attribute] || %{}
)
}
}
end
end
end
end
# The function `keep_read_action_loads_when_loading?` always returns a constant value
# because its a compile attr
# So dialyzer always complains that `!false` can never be true
@dialyzer {:nowarn_function, strip_load?: 1}
defp strip_load?(initial_data) do
initial_data && !Ash.Actions.Helpers.keep_read_action_loads_when_loading?()
end
@dialyzer {:nowarn_function, prefer_existing_loads?: 1}
defp prefer_existing_loads?(query) do
query.context[:loading_relationships?] &&
!Ash.Actions.Helpers.keep_read_action_loads_when_loading?()
end
defp 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 handle_aggregate_multitenancy(query) do
Enum.reduce_while(query.aggregates, {:ok, %{}}, fn {key, aggregate}, {:ok, acc} ->
case handle_multitenancy(
Ash.Query.set_tenant(aggregate.query, aggregate.query.tenant || query.tenant)
) do
{:ok, %{valid?: true} = query} ->
{:cont, {:ok, Map.put(acc, key, %{aggregate | query: query})}}
{:ok, query} ->
{:halt, {:error, Ash.Error.set_path(query.errors, aggregate.name)}}
{:error, error} ->
{:halt, {:error, Ash.Error.set_path(error, aggregate.name)}}
end
end)
|> case do
{:ok, aggregates} -> {:ok, %{query | aggregates: aggregates}}
{:error, error} -> {:error, error}
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_new(item.__metadata__, :tenant, query.tenant)}
end)
else
data
end
end
defp add_query(result, query, opts) do
if opts[:return_query?] do
query = clear_async_limiter(query)
{:ok, result, query}
else
{:ok, result}
end
end
@doc false
def add_page(data, action, count, sort, original_query, new_query, opts) do
cond do
opts[:skip_pagination?] ->
data
action.pagination == false ->
data
original_query.page == false ->
data
opts[:return_unpaged?] && original_query.page[:limit] ->
Ash.Page.Unpaged.new(data, opts)
original_query.page[:limit] ->
to_page(data, action, count, sort, original_query, new_query, opts)
true ->
data
end
end
@doc false
def to_page(data, action, count, sort, original_query, new_query, opts) do
# prevent leakage of stale pid as we stop it at the end of reading
original_query =
clear_async_limiter(original_query)
count = count || new_query.context[:full_count]
page_opts = original_query.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? ->
case Application.get_env(:ash, :default_page_type, :offset) do
:keyset ->
Ash.Page.Keyset.new(data, count, sort, original_query, more?, opts)
:offset ->
Ash.Page.Offset.new(data, count, original_query, more?, opts)
end
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,
_query,
_missing_pkeys?
) do
data
end
defp attach_fields(
data_with_selected,
data,
original_query,
query,
missing_pkeys?,
no_relationships? \\ false
) 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 || []) ++
Enum.map(aggregates_in_data, & &1.load) ++ Enum.map(calculations_in_data, & &1.load)
fields_from_data =
if no_relationships? do
fields_from_data
else
Keyword.keys(original_query.load || []) ++ fields_from_data
end
fields_from_aggregates =
Enum.map(aggregates_in_aggregates, & &1.name)
fields_from_calculations =
Enum.map(calculations_in_calculations, & &1.name)
fields_from_calculations =
query.calculations
|> Map.keys()
|> Enum.filter(&match?({:__ash_fields_are_visible__, _}, &1))
|> Enum.concat(fields_from_calculations)
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 Ash.Resource.Info.embedded?(original_query.resource) || 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
if Ash.Resource.Info.primary_key_simple_equality?(original_query.resource) do
key_fun =
case Ash.Resource.Info.primary_key(original_query.resource) do
[field] -> &{Map.get(&1, field), &1}
fields -> &{Map.take(&1, fields), &1}
end
data_with_selected = Map.new(data_with_selected, key_fun)
Enum.map(data, fn record ->
case Map.fetch(data_with_selected, elem(key_fun.(record), 0)) do
{:ok, 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)
)
)
:error ->
Ash.Resource.put_metadata(record, :private, %{missing_from_data_layer: true})
end
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
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
@doc false
def add_calc_context_to_query(query, actor, authorize?, tenant, tracer, domain, opts) do
{:ok, sort} =
add_calc_context_to_sort(
query.sort,
actor,
authorize?,
tenant,
tracer,
query.resource,
domain,
Keyword.put(opts, :first_combination, Enum.at(query.combination_of, 0))
)
%{
query
| sort: sort,
aggregates:
Map.new(query.aggregates, fn {key, agg} ->
{key,
add_calc_context(
agg,
actor,
agg.authorize? && authorize?,
tenant,
tracer,
domain,
query.resource,
opts
)}
end),
calculations:
Map.new(query.calculations, fn {key, calc} ->
{key,
add_calc_context(
calc,
actor,
authorize?,
tenant,
tracer,
domain,
query.resource,
opts
)}
end),
filter:
add_calc_context_to_filter(
query.filter,
actor,
authorize?,
tenant,
tracer,
domain,
query.resource,
opts
)
}
|> add_calc_context_to_loads(actor, authorize?, tenant, tracer, domain)
end
defp add_calc_context_to_loads(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_loads(
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_loads(actor, authorize?, tenant, tracer, domain)
{key, load}
end
end)
}
end
def add_calc_context(
item,
actor,
authorize?,
tenant,
tracer,
domain,
resource \\ nil,
opts \\ []
)
@doc false
def add_calc_context(
%Ash.Query.Aggregate{} = agg,
actor,
authorize?,
tenant,
tracer,
domain,
resource,
opts
) do
query =
if is_nil(agg.query.__validated_for_action__) && agg.relationship_path != [] do
read_action =
agg.read_action || Ash.Resource.Info.primary_action!(agg.query.resource, :read).name
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,
actor: actor,
tenant: tenant,
authorize?: agg.authorize? && authorize?
)
else
Ash.Query.set_context(agg.query, %{
private: %{
authorize?: agg.authorize? && authorize?,
actor: actor
}
})
end
authorize? =
case agg.name do
{:__calc_dep__, _} ->
true
_ ->
authorize?
end
opts =
if resource do
Keyword.update(opts, :parent_stack, [resource], &[resource, &1])
else
opts
end
field =
if is_atom(agg.field) do
Ash.Resource.Info.field(query.resource, agg.field)
else
agg.field
end
field =
case field do
%struct{} = nested
when struct in [
Ash.Query.Aggregate,
Ash.Query.Calculation,
Ash.Resource.Aggregate,
Ash.Resource.Calculation
] ->
add_calc_context(
nested,
actor,
authorize?,
tenant,
tracer,
domain,
query.resource,
opts
)
_ ->
agg.field
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, opts),
field: field,
join_filters:
Map.new(agg.join_filters, fn {key, filter} ->
{key,
add_calc_context_to_filter(
filter,
actor,
authorize?,
tenant,
tracer,
domain,
query.resource,
opts
)}
end)
}
end
def add_calc_context(
%Ash.Resource.Aggregate{} = agg,
actor,
authorize?,
tenant,
tracer,
domain,
resource,
opts
) do
agg_opts = [
actor: actor,
authorize?: authorize?,
tenant: tenant,
tracer: tracer,
domain: domain
]
{:ok, aggregate} = resource_aggregate_to_query_aggregate(resource, agg, agg_opts)
add_calc_context(aggregate, actor, authorize?, tenant, tracer, domain, resource, opts)
end
def add_calc_context(
%Ash.Resource.Calculation{} = calc,
actor,
authorize?,
tenant,
tracer,
domain,
resource,
opts
) do
{:ok, calc} =
Ash.Query.Calculation.from_resource_calculation(resource, calc.name,
source_context: opts[:source_context]
)
add_calc_context(calc, actor, authorize?, tenant, tracer, domain, resource, opts)
end
def add_calc_context(calc, actor, authorize?, tenant, tracer, _domain, _resource, opts) do
authorize? =
case calc.name do
{:__calc_dep__, _} ->
true
_ ->
authorize?
end
%{
calc
| context: %{
calc.context
| actor: actor,
authorize?: authorize?,
tenant: tenant,
tracer: tracer,
source_context: opts[:source_context] || %{}
}
}
end
@doc false
def update_aggregate_filters(
filter,
resource,
authorize?,
relationship_path_filters,
actor,
tenant,
tracer,
domain,
parent_stack,
source_context
) do
if authorize? do
Filter.update_aggregates(
filter,
resource,
fn aggregate, ref, parent_stack ->
if aggregate.authorize? do
authorize_aggregate(
aggregate,
relationship_path_filters,
actor,
authorize?,
tenant,
tracer,
domain,
resource,
ref.relationship_path,
parent_stack,
source_context
)
else
aggregate
end
end,
[],
Enum.map(parent_stack, &{&1, []})
)
else
filter
end
end
defp add_join_filters(
current_join_filters,
aggregate_relationship_path,
resource,
path_filters,
prefix \\ []
) do
if aggregate_relationship_path == [] do
current_join_filters
else
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
end
defp maybe_await(%Task{} = task, timeout) do
case Task.await(task, timeout) do
{:__exception__, e, stacktrace} ->
reraise e, stacktrace
other ->
other
end
end
defp maybe_await(other, _timeout), do: other
defp fetch_count(
%{context: %{full_count: full_count}},
_,
_,
_,
return?
)
when not is_nil(full_count) do
if return? do
{:ok, {:ok, full_count}}
else
{:ok, fn -> {:ok, full_count} end}
end
end
defp fetch_count(
%{action: %{manual: {_mod, _opts}}},
_,
_,
_,
return?
) do
if return? do
{:ok, {:ok, nil}}
else
{:ok, fn -> {:ok, nil} end}
end
end
defp fetch_count(
%{action: action, resource: resource, page: page} = query,
query_before_pagination,
relationship_path_filters,
opts,
return?
) do
needs_count? =
action.pagination && page &&
(page[:count] == true ||
(page[:count] != false and action.pagination.countable == :by_default))
cond do
query.filter && query.filter.expression == false ->
result = if needs_count?, do: 0, else: nil
if return? do
{:ok, {:ok, result}}
else
{:ok, fn -> {:ok, result} end}
end
Map.has_key?(query.context, :accessing_from) and needs_count? ->
if return? do
# Relationship count is fetched by the parent using aggregates, just return nil here
{:ok, {:ok, nil}}
else
{:ok, fn -> {:ok, nil} end}
end
needs_count? ->
with {:ok, filter} <-
filter_with_related(
query,
opts[:authorize?],
relationship_path_filters
),
query <-
query
|> Ash.Query.unset([:sort, :distinct_sort, :lock, :load, :limit, :offset, :page])
|> 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 return? do
if Ash.DataLayer.in_transaction?(resource) ||
!Ash.DataLayer.can?(:async_engine, resource) ||
Application.get_env(:ash, :disable_async?) 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
else
{:ok, fn -> do_fetch_count(query, data_layer_query) end}
end
end
true ->
if return? do
{:ok, {:ok, nil}}
else
{:ok, fn -> {:ok, nil} end}
end
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,
subject: query,
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,
query.domain
)
state =
cond do
is_struct(state) ->
if Map.has_key?(state, :subject) && !state.subject do
Map.put(state, :subject, query)
else
state
end
is_map(state) && !Map.has_key?(state, :subject) ->
Map.put(state, :subject, query)
true ->
state
end
context = Map.take(data, 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, tenant: query.tenant) do
{:ok, aggregate} ->
Ash.DataLayer.run_aggregate_query(data_layer_query, [aggregate], query.resource)
{:error, error} ->
{:error, error}
end
end
def page_opts(action, page_opts, relationship?) do
cond do
action.type != :read ->
nil
action.pagination == false ->
nil
Keyword.keyword?(page_opts) && !Keyword.has_key?(page_opts, :limit) &&
action.pagination.default_limit ->
Keyword.put(page_opts, :limit, action.pagination.default_limit)
is_nil(page_opts) and action.pagination.required? and not relationship? ->
if action.pagination.default_limit do
[limit: action.pagination.default_limit]
else
page_opts
end
true ->
page_opts
end
end
@doc false
def paginate(starting_query, _action, true) do
{:ok, starting_query}
end
def paginate(starting_query, action, _skip?) do
page_opts = starting_query.page
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[:count] == true && !action.pagination.countable ->
{:error,
NonCountableAction.exception(resource: starting_query.resource, action: action.name)}
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,
pagination.stable_sort || Ash.Resource.Info.primary_key(query.resource)
)
end
paginated =
case pagination_type(page_opts, query.action.pagination) do
:keyset -> keyset_pagination(query, pagination, page_opts)
:offset -> limit_offset_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 pagination_type(_page_opts, pagination) when pagination in [nil, false], do: nil
defp pagination_type(page_opts, pagination) do
cond do
!page_opts ->
nil
page_opts[:before] || page_opts[:after] ->
:keyset
page_opts[:offset] ->
:offset
pagination.keyset? ->
:keyset
pagination.offset? ->
:offset
true ->
nil
end
end
defp keyset_pagination(query, pagination, opts) do
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, page_opts) do
if Application.get_env(:ash, :show_keysets_for_all_actions?, true) do
# in 4.0 remove this branch
show_keyset? =
query.resource
|> Ash.Resource.Info.actions()
|> Enum.any?(&match?(%{pagination: %{keyset?: true}}, &1))
if show_keyset? do
query.sort
|> Enum.map(&elem(&1, 0))
|> then(fn load ->
Ash.Query.load(query, load)
end)
else
query
end
else
case pagination_type(page_opts, query.action.pagination) do
:keyset ->
query.sort
|> Enum.map(&elem(&1, 0))
|> then(fn load ->
Ash.Query.load(query, load)
end)
_ ->
query
end
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
cond do
query.limit == 0 ->
{:ok, []}
query.filter && query.filter.expression == false ->
Logger.debug("""
#{inspect(query.resource)}.#{query.action.name}: skipped query run due to filter being false"
""")
{:ok, []}
true ->
result
|> Helpers.select(query)
|> Helpers.load_runtime_types(query, load_attributes?)
|> case do
{:ok, result} ->
Ash.load(result, query, domain: query.domain, reuse_values?: true)
other ->
other
end
end
|> then(&{&1, query})
end
defp run_query(
%{
resource: destination_resource,
context: %{
data_layer: %{
lateral_join_source: {root_data, path}
}
}
} = query,
data_layer_query,
_context,
load_attributes?
) do
cond do
query.limit == 0 ->
{:ok, []}
query.filter && query.filter.expression == false ->
Logger.debug("""
#{inspect(query.resource)}.#{query.action.name}: skipped query run due to filter being false"
""")
{:ok, []}
true ->
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
|> then(&{&1, query})
end
defp run_query(
%{action: %{manual: {mod, opts}}} = query,
data_layer_query,
context,
load_attributes?
) do
{result, query} =
query
|> mod.read(data_layer_query, opts, context)
|> case do
{:ok, result, extra_info} ->
query =
if extra_info[:full_count] do
Ash.Query.set_context(query, %{full_count: extra_info[:full_count]})
else
query
end
{{:ok, result}, query}
other ->
{other, query}
end
result
|> validate_manual_action_return_result!(query.resource, query.action)
|> Helpers.select(query)
|> Helpers.load_runtime_types(query, load_attributes?)
|> then(&{&1, query})
end
defp run_query(
%{resource: resource} = query,
data_layer_query,
_context,
load_attributes?
) do
cond do
query.limit == 0 ->
{:ok, []}
query.filter && query.filter.expression == false ->
Logger.debug("""
#{inspect(query.resource)}.#{query.action.name}: skipped query run due to filter being false"
""")
{:ok, []}
true ->
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
|> then(&{&1, query})
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.InvalidReturnType,
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,
actor: calculation.context.actor,
tenant: query.to_tenant,
args: calculation.context.arguments,
context: calculation.context.source_context
)
case Ash.Filter.hydrate_refs(expression, %{
resource: query.resource,
public?: false,
first_combination: Enum.at(query.combination_of, 0),
parent_stack: parent_stack_from_context(query.context)
}) 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
@doc false
def parent_stack_from_context(
%{
data_layer: %{lateral_join_source: {_, [{%{resource: resource}, _, _, _} | _]}}
} = context
) do
[resource] ++ List.wrap(context[:parent_stack])
end
def parent_stack_from_context(context) do
List.wrap(context[:parent_stack])
end
defp authorize_calculation_expressions(
hydrated_calculations,
resource,
authorize?,
relationship_path_filters,
actor,
tenant,
tracer,
domain,
parent_stack,
source_context
) do
Enum.map(hydrated_calculations, fn {calculation, expression} ->
{calculation,
update_aggregate_filters(
expression,
resource,
authorize?,
relationship_path_filters,
actor,
tenant,
tracer,
domain,
parent_stack,
source_context
)}
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,
query.resource,
[],
parent_stack_from_context(query.context),
query.context
)
else
add_calc_context(
aggregate,
actor,
authorize?,
tenant,
tracer,
query.domain,
query.resource,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
)
end
%{query | aggregates: Map.put(query.aggregates, name, aggregate)}
end)
end
defp authorize_sorts(query, path_filters, actor, authorize?, tenant, tracer) do
Enum.reduce_while(query.sort, {:ok, []}, fn
{%Ash.Query.Aggregate{} = aggregate, direction}, {:ok, sort} ->
new_agg =
if authorize? && aggregate.authorize? do
authorize_aggregate(
aggregate,
path_filters,
actor,
authorize?,
tenant,
tracer,
query.domain,
query.resource,
[],
parent_stack_from_context(query.context),
query.context
)
else
add_calc_context(
aggregate,
actor,
authorize?,
tenant,
tracer,
query.domain,
query.resource,
parent_stack: parent_stack_from_context(query.context),
source_context: query.context
)
end
{:cont, {:ok, [{new_agg, direction} | sort]}}
{%Ash.Query.Calculation{
module: Ash.Resource.Calculation.Expression,
opts: [expr: expression]
} = calc, direction},
{:ok, sort} ->
new_expr =
update_aggregate_filters(
expression,
query.resource,
authorize?,
path_filters,
actor,
tenant,
tracer,
query.domain,
parent_stack_from_context(query.context),
query.context
)
new_calc = %{calc | opts: [expr: new_expr]}
{:cont, {:ok, [{new_calc, direction} | sort]}}
{other, direction}, {:ok, sort} ->
{:cont, {:ok, [{other, direction} | sort]}}
end)
|> case do
{:ok, reversed_sort} ->
%{query | sort: Enum.reverse(reversed_sort)}
{:error, error} ->
Ash.Query.add_error(query, error)
end
end
defp query_aggregate_from_resource_aggregate(query, resource_aggregate) do
resource = query.resource
{related_resource, aggregate_opts} =
if Map.get(resource_aggregate, :related?, true) do
related_resource =
Ash.Resource.Info.related(resource, resource_aggregate.relationship_path)
opts = [path: resource_aggregate.relationship_path]
{related_resource, opts}
else
related_resource = resource_aggregate.resource
opts = [resource: related_resource]
{related_resource, opts}
end
read_action =
resource_aggregate.read_action ||
Ash.Resource.Info.primary_action!(related_resource, :read).name
opts = query.context[:private] |> Map.take([:actor, :authorize?, :tenant]) |> Map.to_list()
with %{valid?: true} = aggregate_query <-
Ash.Query.for_read(related_resource, read_action, %{}, opts),
%{valid?: true} = aggregate_query <-
Ash.Query.Aggregate.build_query(
aggregate_query,
resource,
filter: resource_aggregate.filter,
sort: resource_aggregate.sort
),
{:ok, query_aggregate} <-
(
full_opts =
[
agg_name: resource_aggregate.name,
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,
include_nil?: resource_aggregate.include_nil?,
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})
] ++ aggregate_opts
Ash.Query.Aggregate.new(
resource,
resource_aggregate.name,
resource_aggregate.kind,
full_opts
)
) do
{:ok, Map.put(query_aggregate, :load, resource_aggregate.name)}
else
{:error, error} ->
{:error, error}
%{errors: errors} ->
{:error, errors}
end
end
defp authorize_aggregate(
aggregate,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain,
resource,
ref_path,
parent_stack,
source_context
) do
aggregate =
add_calc_context(aggregate, actor, authorize?, tenant, tracer, domain, resource,
parent_stack: parent_stack,
source_context: source_context
)
additional_filter =
if Map.get(aggregate, :related?, true) do
last_relationship = last_relationship(aggregate.resource, aggregate.relationship_path)
case Map.fetch(
path_filters,
{last_relationship.source, last_relationship.name, aggregate.query.action.name}
) do
:error ->
true
{:ok, filter} ->
filter
end
else
case Map.fetch(
path_filters,
{aggregate.query.resource, aggregate.query.action.name}
) do
:error ->
true
{:ok, filter} ->
filter
end
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,
[resource | parent_stack],
source_context
),
{:ok, field} <-
aggregate_field_with_related_filters(
aggregate,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain,
ref_path,
parent_stack,
source_context
) 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,
_ref_path,
_parent_stack,
_source_context
),
do: {:ok, nil}
defp aggregate_field_with_related_filters(
%{field: %Ash.Query.Calculation{} = field} = agg,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain,
ref_path,
parent_stack,
source_context
) do
calc =
add_calc_context(field, actor, authorize?, tenant, tracer, domain, agg.resource,
parent_stack: parent_stack,
source_context: source_context
)
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
%Ash.Query.Call{name: :type} = expr ->
expr
expr ->
if calc.type do
{:ok, expr} =
Ash.Query.Function.Type.new([expr, calc.type, calc.constraints])
expr
else
expr
end
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,
agg.query.resource,
source_context: agg.query.context
)
case do_filter_with_related(related_resource, expr, path_filters, ref_path) 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,
ref_path,
parent_stack,
source_context
) do
field =
add_calc_context(field, actor, authorize?, tenant, tracer, domain, agg.resource,
parent_stack: parent_stack,
source_context: source_context
)
if authorize? && field.authorize? do
{:ok,
authorize_aggregate(
field,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain,
agg.resource,
ref_path,
[agg.resource | parent_stack],
source_context
)}
else
{:ok, agg}
end
end
defp aggregate_field_with_related_filters(
aggregate,
path_filters,
actor,
authorize?,
tenant,
tracer,
domain,
ref_path,
parent_stack,
source_context
)
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 ->
case Ash.Query.Calculation.from_resource_calculation(
aggregate.resource,
resource_calculation,
source_context: aggregate.context[:source_context] || %{}
) do
{:ok, calculation} ->
aggregate_field_with_related_filters(
%{aggregate | field: calculation},
path_filters,
actor,
authorize?,
tenant,
tracer,
domain,
ref_path,
parent_stack,
source_context
)
{:error, error} ->
{:error, error}
end
%Ash.Resource.Aggregate{} = resource_aggregate ->
opts = [
actor: actor,
authorize?: authorize?,
tenant: tenant,
tracer: tracer,
domain: domain
]
case resource_aggregate_to_query_aggregate(related_resource, resource_aggregate, opts) do
{:ok, query_aggregate} ->
aggregate_field_with_related_filters(
%{aggregate | field: query_aggregate},
path_filters,
actor,
authorize?,
tenant,
tracer,
domain,
ref_path,
parent_stack,
source_context
)
{:error, error} ->
{:error, error}
end
_ ->
{:ok, aggregate.field}
end
end
defp aggregate_field_with_related_filters(
aggregate,
_path_filters,
_actor,
_authorize?,
_tenant,
_tracer,
_domain,
_ref_path,
_parent_stack,
_source_context
)
when is_atom(aggregate.field) do
{:ok, aggregate.field}
end
defp resource_aggregate_to_query_aggregate(resource, resource_aggregate, opts) do
{agg_related_resource, aggregate_opts} =
if Map.get(resource_aggregate, :related?, true) do
agg_related_resource =
Ash.Resource.Info.related(resource, resource_aggregate.relationship_path)
opts_for_aggregate = [path: resource_aggregate.relationship_path]
{agg_related_resource, opts_for_aggregate}
else
agg_related_resource = resource_aggregate.resource
opts_for_aggregate = [resource: agg_related_resource]
{agg_related_resource, opts_for_aggregate}
end
read_action =
resource_aggregate.read_action ||
Ash.Resource.Info.primary_action!(agg_related_resource, :read).name
with %{valid?: true} = aggregate_query <-
Ash.Query.for_read(agg_related_resource, read_action, %{}, opts),
%{valid?: true} = aggregate_query <-
Ash.Query.Aggregate.build_query(
aggregate_query,
resource,
filter: resource_aggregate.filter,
sort: resource_aggregate.sort
) do
Ash.Query.Aggregate.new(
resource,
resource_aggregate.name,
resource_aggregate.kind,
[
agg_name: resource_aggregate.name,
query: aggregate_query,
field: resource_aggregate.field,
default: resource_aggregate.default,
filterable?: resource_aggregate.filterable?,
type: resource_aggregate.type,
constraints: resource_aggregate.constraints,
include_nil?: resource_aggregate.include_nil?,
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})
] ++ aggregate_opts
)
end
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,
related?: related?,
resource: unrelated_resource
} = exists ->
if related? do
{:ok, new_expr} =
do_filter_with_related(
resource,
exists_expr,
path_filters,
prefix ++ at_path ++ exists_path
)
{:halt, %{exists | expr: new_expr}}
else
primary_read_action = Ash.Resource.Info.primary_action!(unrelated_resource, :read)
filter_key = {:unrelated_exists, unrelated_resource, primary_read_action.name}
case Map.get(path_filters, filter_key) do
nil ->
{:ok, new_expr} =
do_filter_with_related(
unrelated_resource,
exists_expr,
path_filters,
[]
)
{:halt, %{exists | expr: new_expr}}
%Ash.Filter{expression: false} ->
{:halt, false}
auth_filter ->
{:ok, new_expr} =
do_filter_with_related(
unrelated_resource,
exists_expr,
path_filters,
[]
)
combined_expr =
Ash.Query.BooleanExpression.optimized_new(
:and,
new_expr,
auth_filter.expression
)
{:halt, %{exists | expr: combined_expr}}
end
end
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}
defp get_shared_multitenancy(%{context: %{multitenancy: multitenancy}}) do
multitenancy
end
defp get_shared_multitenancy(_) do
nil
end
end