Packages
ash
2.4.21
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.ex
defmodule Ash.Actions.Read do
@moduledoc false
alias Ash.Actions.{Helpers, Load}
alias Ash.Engine.Request
alias Ash.Error.Invalid.{LimitRequired, PaginationRequired}
alias Ash.Error.Query.NoReadAction
alias Ash.Filter
alias Ash.Query.Aggregate
require Logger
require Ash.Query
require Ash.Tracer
def unpaginated_read(query, action \\ nil, opts \\ []) do
action =
cond do
action && is_atom(action) ->
Ash.Resource.Info.action(query.resource, action)
action ->
action
true ->
Ash.Resource.Info.primary_action!(query.resource, :read)
end
cond do
!action ->
{:error, NoReadAction.exception(resource: query.resource, when: "reading")}
action.pagination ->
opts = Keyword.put(opts, :page, false)
run(query, %{action | pagination: false}, opts)
true ->
run(query, action, opts)
end
end
def unpaginated_read_request(query, action \\ nil, opts \\ []) do
action =
cond do
action && is_atom(action) ->
Ash.Resource.Info.action(query.resource, action)
action ->
action
true ->
Ash.Resource.Info.primary_action!(query.resource, :read)
end
cond do
!action ->
{:error, NoReadAction.exception(resource: query.resource, when: "reading")}
action.pagination ->
opts = Keyword.put(opts, :page, false)
run(query, %{action | pagination: false}, opts)
true ->
run(query, action, opts)
end
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 \\ []) do
{query, opts} = Ash.Actions.Helpers.add_process_context(query.api, query, opts)
Ash.Tracer.span :action,
Ash.Api.Info.span_name(query.api, query.resource, action.name),
opts[:tracer] do
metadata = %{
api: query.api,
resource: query.resource,
resource_short_name: Ash.Resource.Info.short_name(query.resource),
actor: opts[:actor],
tenant: opts[:tenant],
action: action.name,
authorize?: opts[:authorize?]
}
Ash.Tracer.telemetry_span [:ash, Ash.Api.Info.short_name(query.api), :read], metadata do
Ash.Tracer.set_metadata(opts[:tracer], :action, metadata)
case do_run(query, action, opts) do
{:error, error} ->
if opts[:tracer] do
opts[:tracer].set_error(Ash.Error.to_error_class(error))
end
{:error, error}
other ->
other
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
authorize? = authorize?(opts)
opts = sanitize_opts(opts, authorize?, query)
query = set_tenant_opt(query, opts)
action = get_action(query.resource, action)
engine_opts =
opts
|> Keyword.put(:authorize?, authorize?)
|> engine_opts(action, query.api, query.resource, opts[:tracer], query)
query =
if opts[:load] do
Ash.Query.load(query, opts[:load])
else
query
end
query =
for_read(
query,
action,
actor: engine_opts[:actor],
authorize?: engine_opts[:authorize?],
timeout: opts[:timeout],
tenant: opts[:tenant]
)
request_opts =
Keyword.merge(engine_opts,
query: query,
page: opts[:page],
return_query?: opts[:return_query?],
timeout: opts[:timeout]
)
request_opts =
if Keyword.has_key?(opts, :initial_data) do
Keyword.put(request_opts, :initial_data, opts[:initial_data])
else
request_opts
end
request_opts =
request_opts
|> Keyword.put(:lazy?, opts[:lazy?] || false)
|> Keyword.put(:tracer, opts[:tracer])
requests =
as_requests(
[:data],
query.resource,
query.api,
action,
request_opts
)
case Ash.Engine.run(requests, engine_opts) do
{:ok, %{data: %{data: %{data: data} = all_data}}} ->
add_query(data, all_data[:fetch][:ultimate_query], request_opts)
{:error, %Ash.Engine{errors: errors, requests: requests}} ->
case Enum.find_value(requests, fn request ->
if request.path == [:fetch] && match?(%Ash.Query{}, request.query) do
request.changeset
end
end) do
nil ->
{:error, Ash.Error.to_error_class(errors, query: query)}
query ->
{:error, Ash.Error.to_error_class(errors, query: query)}
end
{:error, error} ->
{:error, Ash.Error.to_error_class(error, query: query)}
end
end
def as_requests(path, resource, api, action, request_opts) do
query_dependencies = request_opts[:query_dependencies] || []
query_input = request_opts[:query_input] || fn _ -> %{} end
modify_query = request_opts[:modify_query] || fn query, _ -> query end
initial_data = request_opts[:initial_data]
query = request_opts[:query]
get? = !!request_opts[:get?]
tenant = request_opts[:tenant]
timeout = request_opts[:timeout]
error_path = request_opts[:error_path]
lazy? = request_opts[:lazy?]
tracer = request_opts[:tracer]
not_found_error? = request_opts[:not_found_error?]
query =
if initial_data && query && lazy? do
new_load =
query.load
|> List.wrap()
|> Enum.reject(fn load ->
case load do
{key, _value} ->
calculation_or_aggregate?(resource, key) &&
Ash.Resource.loaded?(resource, key)
key ->
calculation_or_aggregate?(resource, key) &&
Ash.Resource.loaded?(resource, key)
end
end)
%{query | load: new_load}
else
query
end
fetch =
Request.new(
resource: resource,
api: api,
action: action,
error_path: error_path,
query:
Request.resolve(
query_dependencies,
fn %{authorize?: authorize?, actor: actor} = context ->
query_opts = query_opts(request_opts, context)
authorize? =
if Keyword.has_key?(request_opts, :initial_data) do
request_opts[:authorize?]
else
authorize?
end
input = query_input.(context) || %{}
tenant =
case tenant do
nil ->
nil
tenant when is_function(tenant) ->
tenant.(context)
tenant ->
tenant
end
query =
case query do
nil ->
Ash.Query.for_read(resource, action.name, input,
tenant: tenant,
actor: actor,
tracer: tracer,
authorize?: authorize?,
timeout: timeout
)
query ->
for_read(
query,
action,
actor: actor,
tenant: tenant,
authorize?: authorize?,
timeout: timeout,
tracer: tracer
)
end
query = %{
query
| api: api,
timeout: timeout || query.timeout || Ash.Api.Info.timeout(api)
}
query =
if tenant do
Ash.Query.set_tenant(query, tenant)
else
query
end
with %{valid?: true} = query <- modify_query.(query, context),
%{limit: initial_limit, offset: initial_offset} <- query,
%{valid?: true} = query <-
handle_attribute_multitenancy(query),
:ok <- validate_multitenancy(query, initial_data),
%{valid?: true} = query <-
query_with_initial_data(query, request_opts),
{:ok, initial_query, query, page_opts} <-
paginate(query, action, page: request_opts[:page]),
page_opts <- page_opts && Keyword.delete(page_opts, :filter),
{%{valid?: true} = query, before_notifications} <-
run_before_action(query),
{:ok, filter_requests} <-
filter_requests(query, path, request_opts, actor, tenant),
load_requests <-
Load.requests(
loaded_query(query),
lazy?,
[
actor: actor,
authorize?: authorize?,
tracer: request_opts[:tracer]
],
path ++ [:fetch]
),
{:ok, sort} <-
Ash.Actions.Sort.process(
query.resource,
query.sort,
query.aggregates,
query.context
) do
load_paths = Enum.map(load_requests, & &1.path)
{:ok,
query
|> Map.put(:sort, sort)
|> Ash.Query.set_context(%{
load_paths: load_paths,
initial_limit: initial_limit,
initial_offset: initial_offset,
page_opts: page_opts,
initial_query: initial_query,
filter_requests: filter_requests,
query_opts: query_opts
}),
%{
requests: load_requests,
notifications: before_notifications
}}
else
%{valid?: false} = query ->
{:error, query.errors}
other ->
other
end
end
),
authorize?: !Keyword.has_key?(request_opts, :initial_data),
data: data_field(request_opts, path),
path: path ++ [:fetch],
async?: !Keyword.has_key?(request_opts, :initial_data),
name: "fetch #{inspect(resource)}.#{action.name}"
)
process =
Request.new(
resource: resource,
api: api,
path: path ++ [:data],
action: action,
authorize?: false,
async?: false,
name: "process #{inspect(resource)}.#{action.name}",
error_path: error_path,
data:
Request.resolve([path ++ [:fetch, :data], path ++ [:fetch, :query]], fn context ->
query_opts = query_opts(request_opts, context)
query = get_in(context, path ++ [:fetch, :query])
fetched_data = get_in(context, path ++ [:fetch, :data])
data = fetched_data[:results]
load_paths = query.context.load_paths
initial_query = query.context.initial_query
aggregate_value_request_paths = fetched_data[:aggregate_value_request_paths] || []
if !Enum.empty?(aggregate_value_request_paths) &&
!get_in(context, path ++ [:aggregate_values]) do
{:new_deps, aggregate_value_request_paths}
else
case Enum.filter(load_paths, fn path ->
!get_in(context, path ++ [:data])
end) do
[] ->
data
|> Load.attach_loads(get_in(context, path ++ [:fetch, :load]) || %{})
|> add_aggregate_values(
query.aggregates,
query.resource,
get_in(context, path ++ [:aggregate_values]) || %{},
Map.get(fetched_data, :aggregates_in_query) || []
)
|> add_calculation_values(
query.resource,
api,
action,
error_path,
path,
Map.get(fetched_data, :ultimate_query) || query,
Map.get(fetched_data, :calculations_at_runtime) || [],
get_in(context, path ++ [:calculation_results]) || :error,
lazy?,
query.tenant
)
|> case do
{:ok, values} ->
values
|> add_tenant(query)
|> add_page(
action,
Map.get(fetched_data, :count),
query.sort,
initial_query,
Keyword.put(query_opts, :page, query.context[:page_opts])
)
|> add_query(Map.get(fetched_data, :ultimate_query), request_opts)
|> unwrap_for_get(get?, not_found_error?, query.resource)
{:requests, requests} ->
{:requests, Enum.map(requests, &{&1, :data})}
end
deps ->
{:new_deps, Enum.map(deps, &(&1 ++ [:data]))}
end
end
end)
)
[fetch, process]
end
defp unwrap_for_get({:ok, [value | _]}, true, _, _resource), do: {:ok, value}
defp unwrap_for_get({:ok, []}, true, true, resource),
do: {:error, Ash.Error.Query.NotFound.exception(resource: resource)}
defp unwrap_for_get({:ok, []}, true, _, _resource),
do: {:ok, nil}
defp unwrap_for_get(other, false, _, _resource), do: other
defp query_opts(request_opts, %{authorize?: authorize?, actor: actor, verbose?: verbose?}) do
request_opts
|> Keyword.take([
:page,
:return_query?,
:action,
:tenant,
:stacktraces?
])
|> Keyword.put(:authorize?, authorize?)
|> Keyword.put(:actor, actor)
|> Keyword.put(:verbose?, verbose?)
end
defp calculation_or_aggregate?(resource, field) do
!!(Ash.Resource.Info.aggregate(resource, field) ||
Ash.Resource.Info.calculation(resource, field))
end
defp handle_attribute_multitenancy(query) do
multitenancy_attribute = Ash.Resource.Info.multitenancy_attribute(query.resource)
if multitenancy_attribute && query.tenant do
{m, f, a} = Ash.Resource.Info.multitenancy_parse_attribute(query.resource)
attribute_value = apply(m, f, [query.tenant | a])
Ash.Query.filter(query, [{multitenancy_attribute, attribute_value}])
else
query
end
end
defp authorize?(opts) do
if opts[:authorize?] == false do
false
else
opts[:authorize?] || Keyword.has_key?(opts, :actor)
end
end
defp set_tenant_opt(query, opts) do
if opts[:tenant] do
Ash.Query.set_tenant(query, opts[:tenant])
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, authorize?, query) do
opts
|> Keyword.put(:authorize?, authorize?)
|> Keyword.merge(Map.get(query.context, :override_api_params) || [])
end
defp engine_opts(opts, action, api, resource, tracer, query) do
opts
|> Keyword.take([:verbose?, :actor, :authorize?, :timeout])
|> Keyword.put(:transaction?, action.transaction? || opts[:transaction?])
|> Keyword.put(:default_timeout, Ash.Api.Info.timeout(api))
|> Keyword.put(:resource, resource)
|> Keyword.put(:name, "#{inspect(resource)}.#{action.name}")
|> Keyword.put(:tracer, tracer)
|> Keyword.put(
:transaction_reason,
%{
type: :read,
metadata: %{
query: query,
resource: resource,
action: action.name
}
}
)
end
defp for_read(query, action, opts) do
if query.__validated_for_action__ == action.name do
query
else
Ash.Query.for_read(query, action.name, %{}, opts)
end
end
defp validate_multitenancy(query, initial_data) do
if is_nil(Ash.Resource.Info.multitenancy_strategy(query.resource)) ||
Ash.Resource.Info.multitenancy_global?(query.resource) || query.tenant ||
initial_data do
:ok
else
{:error,
"Queries against the #{inspect(query.resource)} resource require a tenant to be specified"}
end
end
defp add_tenant(data, query) do
if Ash.Resource.Info.multitenancy_strategy(query.resource) do
Enum.map(data, fn item ->
%{item | __metadata__: Map.put(item.__metadata__, :tenant, query.tenant)}
end)
else
data
end
end
defp add_query(result, query, opts) do
if opts[:return_query?] do
{:ok, result, query}
else
{:ok, result}
end
end
@doc false
def add_page(data, action, count, sort, original_query, opts) do
data =
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
if opts[:page] && action.pagination && !opts[:initial_data] do
to_page(data, action, count, sort, original_query, opts)
else
data
end
end
@doc false
def to_page(data, action, count, sort, original_query, opts) do
page_opts = opts[:page]
{data, rest} =
if opts[:page][:limit] do
Enum.split(data, opts[:page][: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
if action.pagination.keyset? do
Ash.Page.Offset.new(data, count, original_query, more?, opts)
else
Ash.Page.Offset.new(data, count, original_query, more?, opts)
end
else
cond do
action.pagination.keyset? && (page_opts[:after] || page_opts[:before]) ->
Ash.Page.Keyset.new(data, count, sort, original_query, more?, opts)
action.pagination.offset? && action.pagination.keyset? ->
Ash.Page.Offset.new(data, count, original_query, more?, opts)
action.pagination.offset? ->
Ash.Page.Offset.new(data, count, original_query, more?, opts)
true ->
Ash.Page.Keyset.new(data, count, sort, original_query, more?, opts)
end
end
end
defp filter_requests(query, request_path, opts, actor, tenant) do
authorizing? =
if opts[:authorize?] == false do
false
else
Keyword.has_key?(opts, :actor) || opts[:authorize?]
end
if not Keyword.has_key?(opts, :initial_data) &&
authorizing? do
Filter.read_requests(query.api, query.filter, request_path, actor, tenant)
else
{:ok, []}
end
end
defp query_with_initial_data(query, opts) do
case Keyword.fetch(opts, :initial_data) do
:error ->
query
{:ok, nil} ->
Ash.Query.filter(query, false)
{:ok, []} ->
Ash.Query.filter(query, false)
{:ok, [record]} ->
pkey_value =
record |> Map.take(Ash.Resource.Info.primary_key(query.resource)) |> Map.to_list()
Ash.Query.filter(query, ^pkey_value)
{:ok, %{} = record} ->
pkey_value =
record |> Map.take(Ash.Resource.Info.primary_key(query.resource)) |> Map.to_list()
Ash.Query.filter(query, ^pkey_value)
{:ok, records} when is_list(records) ->
pkey = Ash.Resource.Info.primary_key(query.resource)
pkey_value = Enum.map(records, fn record -> record |> Map.take(pkey) |> Map.to_list() end)
filter = [or: pkey_value]
Ash.Query.filter(query, ^filter)
end
end
defp data_field(request_opts, path) do
Request.resolve(
[path ++ [:fetch, :query]],
fn data ->
ash_query = get_in(data, path ++ [:fetch, :query])
actor = data[:actor]
authorize? = data[:authorize?]
initial_query = ash_query.context[:initial_query]
filter_requests = ash_query.context[:filter_requests] || []
initial_limit = initial_query.context[:initial_limit]
initial_offset = initial_query.context[:initial_offset]
ash_query = loaded_query(ash_query)
used_calculations =
ash_query.filter
|> Ash.Filter.used_calculations(
ash_query.resource,
[],
ash_query.calculations,
ash_query.aggregates
)
can_be_in_query? =
not Keyword.has_key?(request_opts, :initial_data) && !ash_query.action.manual
authorizing? =
if request_opts[:authorize?] == false do
false
else
request_opts[:authorize?] || Keyword.has_key?(request_opts, :actor)
end
{calculations_in_query, calculations_at_runtime} =
if Ash.DataLayer.data_layer_can?(ash_query.resource, :expression_calculation) &&
!request_opts[:initial_data] do
ash_query.calculations
|> Map.values()
|> Enum.split_with(fn calculation ->
Enum.find(used_calculations, &(&1.name == calculation.name)) ||
calculation.name in Enum.map(ash_query.sort || [], &elem(&1, 0)) ||
(:erlang.function_exported(calculation.module, :expression, 2) &&
!calculation.allow_async?)
end)
else
{[], Map.values(ash_query.calculations)}
end
must_be_reselected =
if request_opts[:initial_data] do
calc_selects =
Enum.flat_map(calculations_at_runtime, fn %{select: select} ->
List.wrap(select)
end)
ash_query.load
|> Enum.map(fn
{key, _} ->
key
key ->
key
end)
|> Enum.flat_map(fn key ->
if relationship = Ash.Resource.Info.relationship(ash_query.resource, key) do
[relationship.source_attribute]
else
[]
end
end)
|> Enum.concat(calc_selects)
|> remove_already_selected(request_opts[:initial_data])
else
[]
end
{aggregate_auth_requests, aggregate_value_requests, aggregates_in_query} =
Aggregate.requests(
ash_query,
can_be_in_query?,
authorizing?,
calculations_in_query,
path
)
cond do
# if aggregate auth requests is not empty but we have not received the data from
# those requests, we should ask the engine to run the aggregate value requests
!Enum.empty?(aggregate_auth_requests) && !get_in(data, path ++ [:aggregate]) ->
{:requests, Enum.map(aggregate_auth_requests, &{&1, :authorization_filter})}
!Enum.empty?(filter_requests) && !get_in(data, path ++ [:filter]) ->
{:requests, Enum.map(filter_requests, &{&1, :authorization_filter})}
request_opts[:initial_data] && !Enum.empty?(must_be_reselected) ->
primary_key = Ash.Resource.Info.primary_key(ash_query.resource)
filter =
request_opts[:initial_data]
|> List.wrap()
|> Enum.map(&Map.take(&1, primary_key))
|> case do
[] ->
false
[single] ->
[single]
multiple ->
[or: multiple]
end
initial_query
|> Ash.Query.unset([
:filter,
:aggregates,
:sort,
:limit,
:offset,
:distinct,
:select
])
|> Ash.Query.select(must_be_reselected)
|> Ash.Query.set_tenant(request_opts[:tenant] || ash_query.tenant)
|> Ash.Query.do_filter(filter)
|> Ash.Query.data_layer_query()
|> case do
{:ok, data_layer_query} ->
case Ash.DataLayer.run_query(data_layer_query, ash_query.resource) do
{:ok, results} ->
{:ok,
%{
results:
attach_newly_selected_fields(
request_opts[:initial_data],
results,
primary_key,
must_be_reselected
),
aggregates_in_query: aggregates_in_query,
calculations_at_runtime: calculations_at_runtime,
aggregate_value_request_paths:
Enum.map(aggregate_value_requests, &(&1.path ++ [:data]))
},
%{
requests: aggregate_value_requests
}}
{:error, error} ->
{:error, error}
end
{:error, error} ->
{:error, error}
end
request_opts[:initial_data] ->
{:ok,
%{
results: request_opts[:initial_data],
aggregates_in_query: aggregates_in_query,
calculations_at_runtime: calculations_at_runtime,
aggregate_value_request_paths:
Enum.map(aggregate_value_requests, &(&1.path ++ [:data]))
},
%{
requests: aggregate_value_requests
}}
true ->
query =
initial_query
|> Ash.Query.unset([
:filter,
:aggregates,
:sort,
:limit,
:offset,
:distinct,
:select
])
|> Ash.Query.set_context(%{action: ash_query.action})
|> Ash.Query.data_layer_query(only_validate_filter?: true)
ash_query =
if ash_query.select || calculations_at_runtime == [] do
ash_query
else
to_select =
ash_query.resource
|> Ash.Resource.Info.attributes()
|> Enum.map(& &1.name)
Ash.Query.select(ash_query, to_select)
end
ash_query =
Enum.reduce(calculations_at_runtime, ash_query, fn calculation, ash_query ->
if calculation.select do
Ash.Query.select(ash_query, calculation.select || [])
else
ash_query
end
end)
with %{valid?: true} <- ash_query,
{:ok, query} <- query,
{:ok, filter} <-
filter_with_related(
Enum.map(filter_requests, & &1.path),
ash_query.filter,
data
),
{:ok, filter} <-
Filter.run_other_data_layer_filters(
ash_query.api,
ash_query.resource,
filter,
{path, ash_query.tenant, data}
),
filter <- update_aggregate_filters(filter, data, path),
{:ok, query} <-
Ash.DataLayer.select(
query,
Helpers.attributes_to_select(ash_query),
ash_query.resource
),
{:ok, query} <-
add_aggregates(
query,
ash_query,
aggregates_in_query,
Map.get(data, :aggregate, %{})
),
{:ok, query} <-
add_calculations(
query,
ash_query,
calculations_in_query
),
{:ok, query} <-
Ash.DataLayer.filter(
query,
filter,
ash_query.resource
),
{:ok, query} <-
Ash.DataLayer.sort(query, ash_query.sort, ash_query.resource),
{:ok, query} <-
Ash.DataLayer.distinct(query, ash_query.distinct, ash_query.resource),
{:ok, count} <-
fetch_count(
ash_query,
query,
ash_query.resource,
ash_query.action,
initial_limit,
initial_offset,
request_opts
),
{:ok, query} <- apply_keyset_filter(query, ash_query),
{:ok, query} <-
Ash.DataLayer.limit(query, ash_query.limit, ash_query.resource),
{:ok, query} <-
Ash.DataLayer.offset(query, ash_query.offset, ash_query.resource),
{:ok, query} <- set_tenant(query, ash_query),
{:ok, results} <-
run_query(ash_query, query, %{
actor: actor,
tenant: ash_query.tenant,
authorize?: authorize?,
api: ash_query.api
}),
:ok <- validate_get(results, ash_query.action, ash_query),
{:ok, results, after_notifications} <-
run_after_action(initial_query, results),
{:ok, count} <- maybe_await(count) do
if request_opts[:return_query?] do
{:ok,
%{
results: results,
ultimate_query: %{ash_query | filter: filter},
count: count,
calculations_at_runtime: calculations_at_runtime,
aggregates_in_query: aggregates_in_query,
aggregate_value_request_paths:
Enum.map(aggregate_value_requests, &(&1.path ++ [:data]))
},
%{
notifications: after_notifications,
requests: aggregate_value_requests
}}
else
{:ok,
%{
results: results,
count: count,
calculations_at_runtime: calculations_at_runtime,
aggregates_in_query: aggregates_in_query,
aggregate_value_request_paths:
Enum.map(aggregate_value_requests, &(&1.path ++ [:data]))
},
%{
notifications: after_notifications,
requests: aggregate_value_requests
}}
end
else
{:filter_requests, other_data_layer_filter_requests} ->
{:requests, other_data_layer_filter_requests}
%{valid?: false} = query ->
{:error, query.errors}
other ->
other
end
end
end
)
end
defp loaded_query(query) do
query = load_calc_requirements(query)
query =
Enum.reduce(query.sort || [], query, fn
{%Ash.Query.Calculation{name: name, module: module, opts: opts} = calculation, _},
query ->
{resource_load, resource_select} =
if resource_calculation = Ash.Resource.Info.calculation(query.resource, name) do
{resource_calculation.load, resource_calculation.select}
else
{[], []}
end
fields_to_select =
resource_select
|> Kernel.||([])
|> Enum.concat(module.select(query, opts, calculation.context) || [])
calculation = %{calculation | load: name, select: fields_to_select}
query =
Ash.Query.load(
query,
module.load(
query,
opts,
Map.put(calculation.context, :context, query.context)
)
|> Ash.Actions.Helpers.validate_calculation_load!(module)
)
Ash.Query.load(query, resource_load)
{key, _value}, query ->
if Ash.Resource.Info.aggregate(query.resource, key) do
Ash.Query.load(query, key)
else
query
end
end)
query.load
|> Enum.reduce(query, fn
{key, _}, query when is_atom(key) ->
if relationship = Ash.Resource.Info.relationship(query.resource, key) do
Ash.Query.ensure_selected(query, relationship.source_attribute)
else
query
end
key, query when is_atom(key) ->
if relationship = Ash.Resource.Info.relationship(query.resource, key) do
Ash.Query.ensure_selected(query, relationship.source_attribute)
else
query
end
_, _ ->
query
end)
end
defp load_calc_requirements(query, checked_calcs \\ []) do
unchecked_calcs = Map.drop(query.calculations, checked_calcs)
if unchecked_calcs == %{} do
query
else
query.calculations
|> Enum.reduce(query, fn {_name, calc}, query ->
Ash.Query.load(
query,
calc.required_loads
)
|> Ash.Query.ensure_selected(calc.select)
end)
|> load_calc_requirements(Map.keys(query.calculations))
end
end
defp remove_already_selected(fields, %{results: results}),
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.all?(initial_data, &Ash.Resource.selected?(&1, field))
end)
end
defp attach_newly_selected_fields(data, data_with_selected, primary_key, reselected_fields) do
Enum.map(data, fn record ->
case Enum.find(data_with_selected, fn selected_record ->
Map.take(selected_record, primary_key) == Map.take(record, primary_key)
end) do
nil ->
Ash.Resource.put_metadata(record, :private, %{missing_from_data_layer: true})
match ->
Map.merge(record, Map.take(match, reselected_fields))
end
end)
end
defp apply_keyset_filter(data_layer_query, query) do
case query.context[:private][:keyset_filter] do
nil ->
{:ok, data_layer_query}
filter ->
case Ash.Filter.parse(
query.resource,
filter,
query.aggregates,
query.calculations
) do
{:ok, filter} ->
Ash.DataLayer.filter(data_layer_query, filter, query.resource)
{:error, error} ->
{:error, error}
end
end
end
defp validate_get([_, _ | _] = results, %{get?: true}, query) do
{:error,
Ash.Error.Invalid.MultipleResults.exception(
count: Enum.count(results),
at_least?: true,
query: query
)}
end
defp validate_get(_, _, _), do: :ok
defp update_aggregate_filters(filter, data, path) do
Filter.update_aggregates(filter, fn aggregate, ref ->
case get_in(
data,
path ++
[:aggregate, ref.relationship_path] ++
aggregate.relationship_path ++ [:authorization_filter]
) do
nil ->
aggregate
authorization_filter ->
%{aggregate | authorization_filter: authorization_filter}
end
end)
end
defp maybe_await(%Task{} = task) do
case Task.await(task) do
{:__exception__, e, stacktrace} ->
reraise e, stacktrace
other ->
other
end
end
defp maybe_await(other), do: other
defp fetch_count(ash_query, query, resource, action, initial_limit, initial_offset, opts) do
if action.pagination &&
opts[:page] &&
(opts[:page][:count] == true ||
(opts[:page][:count] != false and action.pagination.countable == :by_default)) do
if Ash.DataLayer.in_transaction?(resource) || !Ash.DataLayer.can?(:async_engine, resource) do
case do_fetch_count(ash_query, query, initial_limit, initial_offset) do
{:ok, count} -> {:ok, {:ok, count}}
{:error, error} -> {:error, error}
end
else
{:ok,
Ash.Engine.async(
fn ->
do_fetch_count(ash_query, query, initial_limit, initial_offset)
end,
opts
)}
end
else
{:ok, {:ok, nil}}
end
end
defp do_fetch_count(ash_query, query, initial_limit, initial_offset) do
with {:ok, query} <- Ash.DataLayer.limit(query, initial_limit, ash_query.resource),
{:ok, query} <- Ash.DataLayer.offset(query, initial_offset, ash_query.resource),
{:ok, %{count: count}} <- run_count_query(ash_query, query) do
{:ok, count}
end
end
defp run_before_action(query) do
query = Ash.Query.put_context(query, :private, %{in_before_action?: true})
query.before_action
|> Enum.reduce({query, []}, fn before_action, {query, notifications} ->
case before_action.(query) do
{query, new_notifications} ->
{query, notifications ++ new_notifications}
query ->
{query, notifications}
end
end)
end
defp run_after_action(query, results) do
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 set_tenant(query, ash_query) do
if Ash.Resource.Info.multitenancy_strategy(ash_query.resource) == :context && ash_query.tenant do
Ash.DataLayer.set_tenant(ash_query.resource, query, ash_query.tenant)
else
{:ok, query}
end
end
def page_opts(action, opts) do
cond do
action.pagination == false ->
nil
Keyword.keyword?(opts[:page]) && !Keyword.has_key?(opts[:page], :limit) ->
Keyword.put(opts[:page], :limit, action.pagination.default_limit)
is_nil(opts[:page]) and action.pagination.required? ->
if action.pagination.default_limit do
[limit: action.pagination.default_limit]
else
opts[:page]
end
true ->
opts[:page]
end
end
@doc false
def paginate(starting_query, action, opts) do
page_opts = page_opts(action, opts)
cond do
action.pagination == false && page_opts ->
{:error, "Pagination is not supported"}
action.pagination == false ->
{:ok, starting_query, starting_query, opts[:page]}
page_opts == false ->
if action.pagination.required? do
{:error, PaginationRequired.exception([])}
else
{:ok, starting_query, starting_query, false}
end
page_opts[:limit] || is_nil(page_opts) || page_opts == [] ->
case do_paginate(starting_query, action.pagination, opts) do
{:ok, initial_query, query} ->
{:ok, initial_query, query, page_opts}
{:error, error} ->
{:error, error}
end
action.pagination.required? ->
{:error, LimitRequired.exception([])}
true ->
{:error, LimitRequired.exception([])}
end
end
defp do_paginate(query, pagination, opts) do
paginated =
cond do
opts[:page][:before] || opts[:page][:after] ->
keyset_pagination(query, pagination, opts[:page])
opts[:page][:offset] ->
limit_offset_pagination(query, pagination, opts[:page])
pagination.offset? && pagination.keyset? ->
keyset_pagination(query, pagination, opts[:page])
pagination.offset? ->
limit_offset_pagination(query, pagination, opts[:page])
true ->
keyset_pagination(query, pagination, opts[:page])
end
case paginated do
{:ok, initial_query, query} ->
if opts[:page][:filter] do
{:ok, Ash.Query.filter(initial_query, ^opts[:page][:filter]),
Ash.Query.filter(query, ^opts[:page][:filter])}
else
{:ok, initial_query, query}
end
{:error, error} ->
{:error, error}
end
end
defp keyset_pagination(query, pagination, opts) do
query =
query.sort
|> Enum.flat_map(fn
{%Ash.Query.Calculation{} = calc, _} ->
[calc]
_ ->
[]
end)
|> Enum.reduce(query, fn
%{load: nil} = calc, query ->
Ash.Query.calculate(query, calc.name, {calc.module, calc.opts}, calc.type, calc.context)
%{load: load}, query ->
Ash.Query.load(query, load)
end)
sorted =
if Ash.Actions.Sort.sorting_on_identity?(query) do
query
else
Ash.Query.sort(query, Ash.Resource.Info.primary_key(query.resource))
end
limited = Ash.Query.limit(sorted, limit(opts[:limit], query.limit, pagination) + 1)
if opts[:before] || opts[:after] do
reversed =
if opts[:before] do
limited
|> Ash.Query.unset(:sort)
|> Ash.Query.sort(Ash.Sort.reverse(limited.sort))
else
limited
end
after_or_before =
if opts[:before] do
:before
else
:after
end
case Ash.Page.Keyset.filter(
query.resource,
opts[:before] || opts[:after],
sorted.sort,
after_or_before
) do
{:ok, filter} ->
{:ok, limited,
reversed
|> Ash.Query.set_context(%{
private: %{
keyset_filter: filter
}
})}
{:error, error} ->
{:error, error}
end
else
{:ok, limited, limited}
end
end
defp limit(page_size, query_limit, pagination) do
max_page_size = pagination && pagination.max_page_size
[page_size, query_limit, max_page_size]
|> Enum.filter(&is_integer/1)
|> Enum.min()
end
defp limit_offset_pagination(query, pagination, opts) do
limited = Ash.Query.limit(query, limit(opts[:limit], query.limit, pagination) + 1)
with_offset =
if opts[:offset] do
Ash.Query.offset(limited, opts[:offset])
else
limited
end
{:ok, with_offset, with_offset}
end
defp run_query(
%{
resource: destination_resource,
context: %{
data_layer: %{
lateral_join_source: {root_data, path}
}
}
} = ash_query,
query,
_context
) do
if ash_query.limit == 0 do
{:ok, []}
else
query
|> Ash.DataLayer.run_query_with_lateral_join(
root_data,
destination_resource,
path
)
|> Helpers.select(ash_query)
end
end
defp run_query(%{resource: resource, action: %{manual: nil}} = ash_query, query, _context) do
if ash_query.limit == 0 do
{:ok, []}
else
query
|> Ash.DataLayer.run_query(resource)
|> Helpers.select(ash_query)
end
end
defp run_query(
%{action: %{manual: {mod, opts}}} = ash_query,
query,
context
) do
ash_query
|> mod.read(query, opts, context)
|> Helpers.select(ash_query)
end
@doc false
def run_count_query(
%{
resource: destination_resource,
context: %{
data_layer: %{lateral_join_source: {root_data, path}}
}
},
query
) do
case Ash.Query.Aggregate.new(destination_resource, :count, :count, [], nil, nil) do
{:ok, aggregate} ->
Ash.DataLayer.run_aggregate_query_with_lateral_join(
query,
[aggregate],
root_data,
destination_resource,
path
)
{:error, error} ->
{:error, error}
end
end
def run_count_query(ash_query, query) do
case Ash.Query.Aggregate.new(ash_query.resource, :count, :count, [], nil, nil) do
{:ok, aggregate} ->
Ash.DataLayer.run_aggregate_query(query, [aggregate], ash_query.resource)
{:error, error} ->
{:error, error}
end
end
defp calculation_request(
all_calcs,
resource,
api,
action,
error_path,
path,
results,
calculation,
query,
lazy?,
tenant
) do
all_calcs = Enum.map(all_calcs, & &1.name)
dependencies =
calculation.required_loads
|> List.wrap()
|> reject_loaded(results, lazy?)
|> Enum.map(fn
{key, _} ->
key
key ->
key
end)
|> Enum.uniq()
|> Enum.filter(fn key ->
key in all_calcs
end)
|> Enum.map(fn key ->
path ++ [:calculation_results, key, :data]
end)
if function_exported?(calculation.module, :calculate, 3) do
Request.new(
resource: resource,
api: api,
action: action,
error_path: error_path,
query: query,
authorize?: false,
async?: true,
data:
Request.resolve(dependencies, fn data ->
primary_key = Ash.Resource.Info.primary_key(query.resource)
temp_results =
Enum.reduce(get_in(data, path ++ [:calculation_results]) || %{}, results, fn {key,
config},
results ->
add_calc_to_results(results, key, config[:data], primary_key)
end)
|> Enum.reject(& &1.__metadata__[:context][:private][:missing_from_data_layer])
primary_keys = Enum.map(temp_results, &Map.take(&1, primary_key))
case calculation.module.calculate(temp_results, calculation.opts, calculation.context) do
:unknown ->
case run_calculation_query(temp_results, [calculation], query, tenant) do
{:ok, results_with_calc} ->
{:ok,
%{
values:
Enum.map(results_with_calc, fn record ->
if calculation.load do
{Map.take(record, primary_key), Map.get(record, calculation.name)}
else
{Map.take(record, primary_key),
Map.get(record.calculations, calculation.name)}
end
end),
load?: not is_nil(calculation.load)
}}
other ->
other
end
{:ok, values} ->
{:ok,
%{
values: Enum.zip(primary_keys, values),
load?: not is_nil(calculation.load)
}}
{:error, error} ->
{:error, error}
values ->
{:ok,
%{
values: Enum.zip(primary_keys, values),
load?: not is_nil(calculation.load)
}}
end
end),
path: path ++ [:calculation_results, calculation.name],
name: "calculate #{calculation.name}"
)
else
Request.new(
resource: resource,
api: api,
action: action,
error_path: error_path,
query: query,
authorize?: false,
async?: true,
data:
Request.resolve([], fn _ ->
case run_calculation_query(results, [calculation], query, tenant) do
{:ok, results_with_calc} ->
{:ok,
%{
values:
Enum.map(results_with_calc, fn record ->
if calculation.load do
Map.get(record, calculation.name)
else
Map.get(record.calculations, calculation.name)
end
end),
load?: not is_nil(calculation.load)
}}
other ->
other
end
end),
path: path ++ [:calculation_results, calculation.name],
name: "calculate #{calculation.name}"
)
end
end
defp reject_loaded(loads, results, true) do
loads
|> List.wrap()
|> Enum.reject(fn load ->
Ash.Resource.loaded?(results, load)
end)
end
defp reject_loaded(loads, _, _) do
loads
end
defp add_calculation_values(
results,
resource,
api,
action,
error_path,
path,
query,
calculations,
:error,
lazy?,
tenant
)
when calculations != [] do
{:requests,
Enum.map(
calculations,
&calculation_request(
calculations,
resource,
api,
action,
error_path,
path,
results,
&1,
query,
lazy?,
tenant
)
)}
end
defp add_calculation_values(
results,
resource,
_api,
_action,
_error_path,
_path,
_query,
_calculations,
calculation_values,
_lazy?,
_tenant
) do
primary_key = Ash.Resource.Info.primary_key(resource)
if calculation_values == :error do
{:ok, results}
else
{:ok,
Enum.reduce(calculation_values, results, fn {name, config}, results ->
add_calc_to_results(results, name, config[:data], primary_key)
end)}
end
end
defp add_calc_to_results(results, name, config, primary_key) do
if config[:load?] do
Enum.map(results, fn record ->
case Enum.find_value(config[:values], fn {pkey, value} ->
if pkey == Map.take(record, primary_key) do
{:ok, value}
end
end) do
{:ok, value} ->
Map.put(record, name, value)
_ ->
Map.put(record, name, nil)
end
end)
else
Enum.map(results, fn record ->
case Enum.find_value(config[:values], fn {pkey, value} ->
if pkey == Map.take(record, primary_key) do
{:ok, value}
end
end) do
{:ok, value} ->
%{record | calculations: Map.put(record.calculations, name, value)}
_ ->
%{record | calculations: Map.put(record.calculations, name, nil)}
end
end)
end
end
defp run_calculation_query(results, calculations, query, tenant) do
pkey = Ash.Resource.Info.primary_key(query.resource)
results
|> List.wrap()
|> Enum.map(fn result ->
result
|> Map.take(pkey)
|> Map.to_list()
end)
|> case do
[] ->
{:ok, []}
pkey_filter ->
with query <-
Ash.Query.unset(query, [
:filter,
:aggregates,
:sort,
:limit,
:offset,
:load,
:distinct
]),
query <- Ash.Query.set_tenant(query, tenant),
query <- Ash.Query.filter(query, ^[or: pkey_filter]),
{:ok, data_layer_query} <- Ash.Query.data_layer_query(query),
{:ok, data_layer_query} <-
add_calculations(data_layer_query, query, calculations) do
Ash.DataLayer.run_query(
data_layer_query,
query.resource
)
end
end
end
defp add_aggregate_values(results, aggregates, resource, aggregate_values, aggregates_in_query) do
keys_to_aggregates =
Enum.reduce(aggregate_values, %{}, fn
{_name, %{data: keys_to_values}}, acc ->
Enum.reduce(keys_to_values, acc, fn {pkey, values}, acc ->
Map.update(acc, pkey, values, &Map.merge(&1, values))
end)
_, acc ->
acc
end)
pkey = Ash.Resource.Info.primary_key(resource)
loaded =
aggregates
|> Enum.map(fn {_, aggregate} -> aggregate.load end)
|> Enum.reject(&is_nil/1)
Enum.map(results, fn result ->
aggregate_values = Map.get(keys_to_aggregates, Map.take(result, pkey), %{})
aggregate_values =
aggregates
|> Enum.reject(fn {name, _aggregate} ->
Enum.find(aggregates_in_query, &(&1.name == name))
end)
|> Enum.reduce(aggregate_values, fn {_, aggregate}, aggregate_values ->
Map.put_new(aggregate_values, aggregate.name, aggregate.default_value)
end)
{top_level, nested} = Map.split(aggregate_values || %{}, loaded)
Map.merge(%{result | aggregates: Map.merge(result.aggregates, nested)}, top_level)
end)
end
defp add_calculations(data_layer_query, query, calculations_to_add) do
calculations =
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)
case Ash.Filter.hydrate_refs(expression, %{
resource: query.resource,
aggregates: query.aggregates,
calculations: query.calculations,
public?: false
}) do
{:ok, expression} ->
{:cont, {:ok, [{calculation, expression} | calculations]}}
{:error, error} ->
{:halt, {:error, error}}
end
else
{:halt, {:error, "Expression calculations are not supported"}}
end
end)
case calculations do
{:ok, calculations} ->
Ash.DataLayer.add_calculations(
data_layer_query,
calculations,
query.resource
)
{:error, error} ->
{:error, error}
end
end
defp add_aggregates(data_layer_query, query, aggregates_to_add, aggregate_filters) do
aggregates_to_add =
Enum.into(aggregates_to_add, %{}, fn aggregate ->
{aggregate.name, aggregate}
end)
aggregates =
aggregates_to_add
|> Enum.reduce({aggregates_to_add, aggregate_filters}, fn {name, aggregate},
{aggregates, aggregate_filters} ->
case Map.fetch(aggregate_filters, aggregate.relationship_path) do
{:ok, %{authorization_filter: filter}} ->
{Map.put(aggregates, name, %{aggregate | authorization_filter: filter}),
Map.delete(aggregates, aggregate.relationship_path)}
:error ->
{aggregates, aggregate_filters}
end
end)
|> elem(0)
Ash.DataLayer.add_aggregates(data_layer_query, Map.values(aggregates), query.resource)
end
defp filter_with_related(
relationship_filter_paths,
filter_expr,
data,
prefix \\ []
)
defp filter_with_related(
relationship_filter_paths,
%Ash.Filter{expression: expression} = filter,
data,
prefix
) do
case filter_with_related(
relationship_filter_paths,
expression,
data,
prefix
) do
{:ok, new_expr} ->
{:ok, %{filter | expression: new_expr}}
other ->
other
end
end
defp filter_with_related(
relationship_filter_paths,
%Ash.Query.BooleanExpression{op: :and, left: left, right: right},
data,
prefix
) do
with {:ok, left} <-
filter_with_related(relationship_filter_paths, left, data, prefix),
{:ok, right} <-
filter_with_related(relationship_filter_paths, right, data, prefix) do
{:ok, Ash.Query.BooleanExpression.optimized_new(:and, left, right)}
end
end
defp filter_with_related(relationship_filter_paths, filter_expr, data, prefix) do
paths_to_global_filter_on =
filter_expr
|> Ash.Filter.relationship_paths(true)
|> Enum.filter(&([:data, :filter, prefix ++ &1] in relationship_filter_paths))
paths_to_global_filter_on
|> Enum.reduce_while(
{:ok, filter_expr},
fn path, {:ok, filter} ->
case get_in(data, [:data, :filter, prefix ++ path, :authorization_filter]) do
nil ->
{:cont, {:ok, filter}}
%Ash.Filter{expression: authorization_filter} ->
{:cont,
{:ok,
Ash.Query.BooleanExpression.optimized_new(
:and,
filter_expr,
Ash.Filter.move_to_relationship_path(authorization_filter, path)
)}}
end
end
)
|> case do
{:ok, filter} ->
{:ok,
filter
|> Ash.Filter.map(fn
%Ash.Query.Exists{at_path: at_path, path: exists_path, expr: exists_expr} = exists ->
paths =
Enum.filter(
relationship_filter_paths,
&List.starts_with?(&1, prefix ++ at_path ++ exists_path)
)
{:ok, new_expr} =
filter_with_related(paths, exists_expr, data, prefix ++ at_path ++ exists_path)
{:halt, %{exists | expr: new_expr}}
other ->
other
end)}
other ->
other
end
end
end