Packages
ash
1.52.0-rc.11
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
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
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)
query =
if opts[:load] do
Ash.Query.load(query, opts[:load])
else
query
end
query =
for_read(
query,
action,
actor: engine_opts[:actor],
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 = Keyword.put(request_opts, :lazy?, opts[:lazy?] || false)
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?]
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)
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,
timeout: timeout
)
query ->
for_read(
query,
action,
actor: actor,
tenant: tenant,
timeout: timeout
)
end
query = %{
query
| api: api,
timeout: timeout || query.timeout || Ash.Api.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} <-
Filter.run_other_data_layer_filters(
query.api,
query.resource,
query.filter,
{path, query.tenant, context}
),
{query, load_requests} <-
Load.requests(
query,
lazy?,
[actor: actor, authorize?: authorize?],
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(:filter, filter)
|> 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,
query_opts: query_opts
}),
%{
requests: load_requests,
notifications: before_notifications
}}
else
{:filter_requests, requests} ->
{:requests, requests}
%{valid?: false} = query ->
{:error, query.errors}
other ->
other
end
end
),
authorize?: true,
data: data_field(request_opts, path),
path: path ++ [:fetch],
async?: !Keyword.has_key?(request_opts, :initial_data),
name: "#{inspect(path)} #{action.type} - fetch `#{action.name}`"
)
process =
Request.new(
resource: resource,
api: api,
path: path ++ [:data],
action: action,
authorize?: false,
async?: false,
name: "#{inspect(path)} #{action.type} - process `#{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
)
|> 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?, 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, resource),
do: {:error, Ash.Error.Query.NotFound.exception(resource: resource)}
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 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) do
opts
|> Keyword.take([:verbose?, :actor, :authorize?, :timeout])
|> Keyword.put(:transaction?, action.transaction? || opts[:transaction?])
|> Keyword.put(:default_timeout, Ash.Api.timeout(api))
|> Keyword.put(:resource, resource)
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
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
more? = not Enum.empty?(rest)
if page_opts[:offset] do
if action.pagination.keyset? do
data
|> Ash.Page.Keyset.data_with_keyset(original_query.resource, sort)
|> Ash.Page.Offset.new(count, original_query, more?, opts)
else
Ash.Page.Offset.new(data, count, original_query, more?, opts)
end
else
cond do
action.pagination.offset? && action.pagination.keyset? ->
data
|> Ash.Page.Keyset.data_with_keyset(original_query.resource, sort)
|> Ash.Page.Offset.new(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) 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)
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]
initial_limit = initial_query.context[:initial_limit]
initial_offset = initial_query.context[:initial_offset]
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
filter_requests = filter_requests(ash_query, path, request_opts)
{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
{aggregate_auth_requests, aggregate_value_requests, aggregates_in_query} =
Aggregate.requests(
ash_query,
can_be_in_query?,
authorizing?,
calculations_in_query,
path
)
cond do
match?({:error, _error}, filter_requests) ->
filter_requests
# 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})}
!match?({:ok, []}, filter_requests) && !get_in(data, path ++ [:filter]) ->
{:ok, filter_requests} = filter_requests
{:requests, Enum.map(filter_requests, &{&1, :authorization_filter})}
true ->
if request_opts[:initial_data] do
{: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
}}
else
query =
initial_query
|> Ash.Query.unset([:filter, :aggregates, :sort, :limit, :offset, :distinct])
|> 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)
{:ok, filter_requests} = filter_requests
with %{valid?: true} <- ash_query,
{:ok, query} <- query,
{:ok, filter} <-
filter_with_related(Enum.map(filter_requests, & &1.path), ash_query, data),
filter <- update_aggregate_filters(filter, data, path),
{:ok, query} <-
Ash.DataLayer.set_context(
ash_query.resource,
query,
Map.put(ash_query.context, :action, ash_query.action)
),
{: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
ultimate_query =
ash_query
|> Ash.Query.unset(:filter)
|> Ash.Query.filter(filter)
{:ok,
%{
results: results,
ultimate_query: ultimate_query,
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
%{valid?: false} = query ->
{:error, query.errors}
other ->
other
end
end
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
Task.await(task)
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,
Task.async(fn ->
do_fetch_count(ash_query, query, initial_limit, initial_offset)
end)}
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.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 && action.pagination.default_limit) ->
opts[:page]
Keyword.keyword?(opts[:page]) && !Keyword.has_key?(opts[:page], :limit) ->
Keyword.put(opts[:page], :limit, action.pagination.default_limit)
is_nil(opts[:page]) ->
[limit: action.pagination.default_limit]
true ->
opts[:page]
end
end
defp 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
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
query
|> Ash.DataLayer.run_query_with_lateral_join(
root_data,
destination_resource,
path
)
|> Helpers.select(ash_query)
end
defp run_query(%{resource: resource, action: %{manual: nil}} = ash_query, query, _context) do
query
|> Ash.DataLayer.run_query(resource)
|> Helpers.select(ash_query)
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
) do
all_calcs = Enum.map(all_calcs, & &1.name)
resource_load =
if resource_calculation = Ash.Resource.Info.calculation(resource, calculation.name) do
List.wrap(resource_calculation.load)
else
[]
end
dependencies =
query
|> calculation.module.load(calculation.opts, calculation.context)
|> List.wrap()
|> Enum.concat(resource_load)
|> Ash.Actions.Helpers.validate_calculation_load!(calculation.module)
|> 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 ->
temp_results =
Enum.reduce(get_in(data, path ++ [:calculation_results]) || %{}, results, fn {key,
config},
results ->
add_calc_to_results(results, key, config[:data])
end)
case calculation.module.calculate(temp_results, calculation.opts, calculation.context) do
:unknown ->
case run_calculation_query(results, [calculation], query) 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
{:ok, values} ->
{:ok,
%{
values: values,
load?: not is_nil(calculation.load)
}}
{:error, error} ->
{:error, error}
values ->
{:ok,
%{
values: values,
load?: not is_nil(calculation.load)
}}
end
end),
path: path ++ [:calculation_results, calculation.name],
name: "#{inspect(path)} calculation #{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) 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: "#{inspect(path)} calculation #{calculation.name}"
)
end
end
defp add_calculation_values(
results,
resource,
api,
action,
error_path,
path,
query,
calculations,
:error
)
when calculations != [] do
{:requests,
Enum.map(
calculations,
&calculation_request(
calculations,
resource,
api,
action,
error_path,
path,
results,
&1,
query
)
)}
end
defp add_calculation_values(
results,
_resource,
_api,
_action,
_error_path,
_path,
_query,
_calculations,
calculation_values
) do
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])
end)}
end
end
defp add_calc_to_results(results, name, config) do
if config[:load?] do
:lists.zipwith(
fn record, value -> Map.put(record, name, value) end,
results,
config[:values]
)
else
:lists.zipwith(
fn record, value ->
%{record | calculations: Map.put(record.calculations, name, value)}
end,
results,
config[:values]
)
end
end
defp run_calculation_query(results, calculations, query) do
pkey = Ash.Resource.Info.primary_key(query.resource)
pkey_filter =
results
|> List.wrap()
|> Enum.map(fn result ->
result
|> Map.take(pkey)
|> Map.to_list()
end)
with query <-
Ash.Query.unset(query, [:filter, :aggregates, :sort, :limit, :offset, :load, :distinct]),
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
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, ash_query, data) do
Enum.reduce_while(
relationship_filter_paths,
{:ok, ash_query.filter},
fn path, {:ok, filter} ->
case get_in(data, path ++ [:authorization_filter]) do
nil ->
{:cont, {:ok, filter}}
authorization_filter ->
path = List.last(path)
authorization_filter = Ash.Filter.put_at_path(authorization_filter, path)
add_authorization_filter(filter, authorization_filter)
end
end
)
end
defp add_authorization_filter(filter, authorization_filter) do
case Ash.Filter.add_to_filter(filter, authorization_filter) do
{:ok, new_filter} ->
{:cont, {:ok, new_filter}}
{:error, error} ->
{:halt, {:error, error}}
end
end
end