mirror of
https://github.com/ash-project/ash.git
synced 2024-09-20 13:33:20 +12:00
fix: don't use Aggregate.new
for builtin aggregates
This commit is contained in:
parent
4d39ab8a80
commit
993355f938
3 changed files with 140 additions and 135 deletions
|
@ -8,84 +8,90 @@ defmodule Ash.Actions.Aggregate do
|
|||
action = query.action || Ash.Resource.Info.primary_action!(query.resource, :read)
|
||||
opts = Keyword.put_new(opts, :read_action, action.name)
|
||||
|
||||
with {:ok, aggregates} <- validate_aggregates(query, aggregates, opts),
|
||||
%{valid?: true} = query <- Ash.Actions.Read.handle_attribute_multitenancy(query) do
|
||||
with %{valid?: true} = query <- Ash.Actions.Read.handle_attribute_multitenancy(query) do
|
||||
aggregates
|
||||
|> Enum.group_by(fn aggregate ->
|
||||
agg_authorize? = aggregate.authorize? && opts[:authorize?]
|
||||
|> Enum.group_by(fn
|
||||
%Ash.Query.Aggregate{} = aggregate ->
|
||||
agg_authorize? = aggregate.authorize? && opts[:authorize?]
|
||||
|
||||
read_action =
|
||||
aggregate.read_action || query.action ||
|
||||
Ash.Resource.Info.primary_action!(query.resource, :read).name
|
||||
read_action =
|
||||
aggregate.read_action || query.action ||
|
||||
Ash.Resource.Info.primary_action!(query.resource, :read).name
|
||||
|
||||
{agg_authorize?, read_action}
|
||||
{agg_authorize?, read_action}
|
||||
|
||||
{_name, _kind} ->
|
||||
{!!opts[:authorize?], opts[:read_action]}
|
||||
|
||||
{_name, _kind, agg_opts} ->
|
||||
authorize? =
|
||||
Keyword.get(agg_opts, :authorize?, true) && opts[:authorize?]
|
||||
|
||||
{authorize?, agg_opts[:read_action] || opts[:read_action] || action.name}
|
||||
end)
|
||||
|> Enum.reduce_while({:ok, %{}}, fn {{agg_authorize?, read_action}, aggregates},
|
||||
{:ok, acc} ->
|
||||
query =
|
||||
if query.__validated_for_action__ == read_action do
|
||||
query
|
||||
else
|
||||
Ash.Query.for_read(query, read_action, %{},
|
||||
tenant: opts[:tenant],
|
||||
actor: opts[:actor],
|
||||
authorize?: opts[:authorize?]
|
||||
)
|
||||
end
|
||||
|
||||
query = %{query | api: api}
|
||||
|
||||
Ash.Tracer.span :action,
|
||||
Ash.Api.Info.span_name(query.api, query.resource, :aggregate),
|
||||
opts[:tracer] do
|
||||
metadata = %{
|
||||
api: query.api,
|
||||
resource: query.resource,
|
||||
resource_short_name: Ash.Resource.Info.short_name(query.resource),
|
||||
aggregates: List.wrap(aggregates),
|
||||
actor: opts[:actor],
|
||||
tenant: opts[:tenant],
|
||||
action: action.name,
|
||||
authorize?: opts[:authorize?]
|
||||
}
|
||||
|
||||
Ash.Tracer.telemetry_span [:ash, Ash.Api.Info.short_name(query.api), :aggregate],
|
||||
metadata do
|
||||
Ash.Tracer.set_metadata(opts[:tracer], :action, metadata)
|
||||
query = Map.put(query, :aggregates, Map.new(aggregates, &{&1.name, &1}))
|
||||
|
||||
with {:ok, query} <- authorize_query(query, opts, agg_authorize?),
|
||||
{:ok, data_layer_query} <-
|
||||
Ash.Query.data_layer_query(Ash.Query.new(query.resource)),
|
||||
aggregates <- merge_query_into_aggregates(query, aggregates),
|
||||
{:ok, result} <-
|
||||
Ash.DataLayer.run_aggregate_query(data_layer_query, aggregates, query.resource) do
|
||||
{:cont, {:ok, Map.merge(acc, result)}}
|
||||
|> Enum.reduce_while({:ok, %{}}, fn
|
||||
{{agg_authorize?, read_action}, aggregates}, {:ok, acc} ->
|
||||
query =
|
||||
if query.__validated_for_action__ == read_action do
|
||||
query
|
||||
else
|
||||
{:error, error} ->
|
||||
{:halt, {:error, error}}
|
||||
Ash.Query.for_read(query, read_action, %{},
|
||||
tenant: opts[:tenant],
|
||||
actor: opts[:actor],
|
||||
authorize?: opts[:authorize?]
|
||||
)
|
||||
end
|
||||
|
||||
query = %{query | api: api}
|
||||
|
||||
Ash.Tracer.span :action,
|
||||
Ash.Api.Info.span_name(query.api, query.resource, :aggregate),
|
||||
opts[:tracer] do
|
||||
metadata = %{
|
||||
api: query.api,
|
||||
resource: query.resource,
|
||||
resource_short_name: Ash.Resource.Info.short_name(query.resource),
|
||||
aggregates: List.wrap(aggregates),
|
||||
actor: opts[:actor],
|
||||
tenant: opts[:tenant],
|
||||
action: action.name,
|
||||
authorize?: opts[:authorize?]
|
||||
}
|
||||
|
||||
Ash.Tracer.telemetry_span [:ash, Ash.Api.Info.short_name(query.api), :aggregate],
|
||||
metadata do
|
||||
Ash.Tracer.set_metadata(opts[:tracer], :action, metadata)
|
||||
|
||||
with {:ok, query} <- authorize_query(query, opts, agg_authorize?),
|
||||
{:ok, aggregates} <- validate_aggregates(query, aggregates, opts),
|
||||
{:ok, data_layer_query} <-
|
||||
Ash.Query.data_layer_query(Ash.Query.new(query.resource)),
|
||||
{:ok, result} <-
|
||||
Ash.DataLayer.run_aggregate_query(
|
||||
data_layer_query,
|
||||
aggregates,
|
||||
query.resource
|
||||
) do
|
||||
{:cont, {:ok, Map.merge(acc, result)}}
|
||||
else
|
||||
{:error, error} ->
|
||||
{:halt, {:error, error}}
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
defp merge_query_into_aggregates(query, aggregates) do
|
||||
Enum.map(aggregates, fn aggregate ->
|
||||
%{
|
||||
aggregate
|
||||
| query:
|
||||
aggregate.query
|
||||
|> Ash.Query.do_filter(query.filter)
|
||||
|> Ash.Query.sort(query.sort, prepend?: true)
|
||||
|> Ash.Query.distinct_sort(query.distinct_sort, prepend?: true)
|
||||
|> Ash.Query.limit(query.limit)
|
||||
|> Ash.Query.set_tenant(query.tenant)
|
||||
|> merge_offset(query.offset)
|
||||
|> Ash.Query.set_context(query.context)
|
||||
}
|
||||
end)
|
||||
defp merge_query(left, right) do
|
||||
left
|
||||
|> Ash.Query.do_filter(right.filter)
|
||||
|> Ash.Query.sort(right.sort, prepend?: true)
|
||||
|> Ash.Query.distinct_sort(right.distinct_sort, prepend?: true)
|
||||
|> Ash.Query.limit(right.limit)
|
||||
|> Ash.Query.set_tenant(right.tenant)
|
||||
|> merge_offset(right.offset)
|
||||
|> Ash.Query.set_context(right.context)
|
||||
end
|
||||
|
||||
defp merge_offset(query, offset) do
|
||||
|
@ -128,7 +134,7 @@ defmodule Ash.Actions.Aggregate do
|
|||
query.resource,
|
||||
name,
|
||||
kind,
|
||||
set_opts([], opts)
|
||||
set_opts(query, [], opts)
|
||||
) do
|
||||
{:ok, aggregate} ->
|
||||
{:cont, {:ok, [aggregate | aggregates]}}
|
||||
|
@ -138,7 +144,7 @@ defmodule Ash.Actions.Aggregate do
|
|||
end
|
||||
|
||||
{name, kind, agg_opts}, {:ok, aggregates} ->
|
||||
case Ash.Query.Aggregate.new(query.resource, name, kind, set_opts(agg_opts, opts)) do
|
||||
case Ash.Query.Aggregate.new(query.resource, name, kind, set_opts(query, agg_opts, opts)) do
|
||||
{:ok, aggregate} ->
|
||||
{:cont, {:ok, [aggregate | aggregates]}}
|
||||
|
||||
|
@ -148,8 +154,23 @@ defmodule Ash.Actions.Aggregate do
|
|||
end)
|
||||
end
|
||||
|
||||
defp set_opts(specified, others) do
|
||||
defp set_opts(query, specified, others) do
|
||||
{agg_opts, _} = Ash.Query.Aggregate.split_aggregate_opts(others)
|
||||
Keyword.merge(agg_opts, specified)
|
||||
|
||||
query =
|
||||
case agg_opts[:query] do
|
||||
%Ash.Query{} = agg_query ->
|
||||
merge_query(agg_query, query)
|
||||
|
||||
nil ->
|
||||
query
|
||||
|
||||
opts ->
|
||||
Ash.Query.build(query, opts)
|
||||
end
|
||||
|
||||
agg_opts
|
||||
|> Keyword.merge(specified)
|
||||
|> Keyword.put(:query, query)
|
||||
end
|
||||
end
|
||||
|
|
|
@ -1623,16 +1623,36 @@ defmodule Ash.Api do
|
|||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Runs an aggregate or aggregates over a resource query
|
||||
|
||||
If you pass an `%Ash.Query.Aggregate{}`, gotten from `Ash.Query.Aggregate.new()`,
|
||||
the query provided as the first argument to this function will not apply. For this
|
||||
reason, it is preferred that you pass in the tuple format, i.e
|
||||
|
||||
Prefer this:
|
||||
`Api.aggregate(query, {:count_of_things, :count})`
|
||||
|
||||
Over this:
|
||||
`Api.aggregate(query, Ash.Query.Aggregate.new(...))`
|
||||
|
||||
#{Spark.OptionsHelpers.docs(@aggregate_opts)}
|
||||
"""
|
||||
@callback aggregate(
|
||||
Ash.Query.t(),
|
||||
Ash.Api.aggregate() | list(Ash.Api.aggregate()),
|
||||
aggregate() | list(aggregate()),
|
||||
opts :: Keyword.t()
|
||||
) ::
|
||||
{:ok, any} | {:error, Ash.Error.t()}
|
||||
|
||||
@doc """
|
||||
Runs an aggregate or aggregates over a resource query
|
||||
|
||||
See `c:aggregate/3` for more.
|
||||
"""
|
||||
@callback aggregate!(
|
||||
Ash.Query.t(),
|
||||
Ash.Api.aggregate() | list(Ash.Api.aggregate()),
|
||||
aggregate() | list(aggregate()),
|
||||
opts :: Keyword.t()
|
||||
) ::
|
||||
any | no_return
|
||||
|
|
|
@ -81,15 +81,9 @@ defmodule Ash.Api.Interface do
|
|||
|
||||
{aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
|
||||
|
||||
case Ash.Query.Aggregate.new(query.resource, :count, :count, aggregate_opts) do
|
||||
{:ok, aggregate} ->
|
||||
case Api.aggregate(__MODULE__, query, aggregate, opts) do
|
||||
{:ok, %{count: count}} ->
|
||||
count
|
||||
|
||||
{:error, error} ->
|
||||
raise Ash.Error.to_error_class(error)
|
||||
end
|
||||
case Api.aggregate(__MODULE__, query, {:count, :count, aggregate_opts}, opts) do
|
||||
{:ok, %{count: count}} ->
|
||||
count
|
||||
|
||||
{:error, error} ->
|
||||
raise Ash.Error.to_error_class(error)
|
||||
|
@ -108,15 +102,9 @@ defmodule Ash.Api.Interface do
|
|||
|
||||
{aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
|
||||
|
||||
case Ash.Query.Aggregate.new(query.resource, :count, :count, aggregate_opts) do
|
||||
{:ok, aggregate} ->
|
||||
case Api.aggregate(__MODULE__, query, aggregate, opts) do
|
||||
{:ok, %{count: count}} ->
|
||||
{:ok, count}
|
||||
|
||||
{:error, error} ->
|
||||
{:error, Ash.Error.to_error_class(error)}
|
||||
end
|
||||
case Api.aggregate(__MODULE__, query, {:count, :count, aggregate_opts}, opts) do
|
||||
{:ok, %{count: count}} ->
|
||||
{:ok, count}
|
||||
|
||||
{:error, error} ->
|
||||
{:error, Ash.Error.to_error_class(error)}
|
||||
|
@ -135,15 +123,9 @@ defmodule Ash.Api.Interface do
|
|||
|
||||
{aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
|
||||
|
||||
case Ash.Query.Aggregate.new(query.resource, :exists, :exists, aggregate_opts) do
|
||||
{:ok, aggregate} ->
|
||||
case Api.aggregate(__MODULE__, query, aggregate, opts) do
|
||||
{:ok, %{exists: exists}} ->
|
||||
exists
|
||||
|
||||
{:error, error} ->
|
||||
raise Ash.Error.to_error_class(error)
|
||||
end
|
||||
case Api.aggregate(__MODULE__, query, {:exists, :exists, aggregate_opts}, opts) do
|
||||
{:ok, %{exists: exists}} ->
|
||||
exists
|
||||
|
||||
{:error, error} ->
|
||||
raise Ash.Error.to_error_class(error)
|
||||
|
@ -162,15 +144,9 @@ defmodule Ash.Api.Interface do
|
|||
|
||||
{aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
|
||||
|
||||
case Ash.Query.Aggregate.new(query.resource, :exists, :exists, aggregate_opts) do
|
||||
{:ok, aggregate} ->
|
||||
case Api.aggregate(__MODULE__, query, aggregate, opts) do
|
||||
{:ok, %{exists: exists}} ->
|
||||
{:ok, exists}
|
||||
|
||||
{:error, error} ->
|
||||
{:error, Ash.Error.to_error_class(error)}
|
||||
end
|
||||
case Api.aggregate(__MODULE__, query, {:exists, :exists, aggregate_opts}, opts) do
|
||||
{:ok, %{exists: exists}} ->
|
||||
{:ok, exists}
|
||||
|
||||
{:error, error} ->
|
||||
{:error, Ash.Error.to_error_class(error)}
|
||||
|
@ -190,20 +166,14 @@ defmodule Ash.Api.Interface do
|
|||
|
||||
{aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
|
||||
|
||||
case Ash.Query.Aggregate.new(
|
||||
query.resource,
|
||||
unquote(kind),
|
||||
unquote(kind),
|
||||
Keyword.put(aggregate_opts, :field, field)
|
||||
case Api.aggregate(
|
||||
__MODULE__,
|
||||
query,
|
||||
{unquote(kind), unquote(kind), Keyword.put(aggregate_opts, :field, field)},
|
||||
opts
|
||||
) do
|
||||
{:ok, aggregate} ->
|
||||
case Api.aggregate(__MODULE__, query, aggregate, opts) do
|
||||
{:ok, %{unquote(kind) => value}} ->
|
||||
{:ok, value}
|
||||
|
||||
{:error, error} ->
|
||||
{:error, Ash.Error.to_error_class(error)}
|
||||
end
|
||||
{:ok, %{unquote(kind) => value}} ->
|
||||
{:ok, value}
|
||||
|
||||
{:error, error} ->
|
||||
{:error, Ash.Error.to_error_class(error)}
|
||||
|
@ -223,20 +193,14 @@ defmodule Ash.Api.Interface do
|
|||
opts
|
||||
end
|
||||
|
||||
case Ash.Query.Aggregate.new(
|
||||
query.resource,
|
||||
unquote(kind),
|
||||
unquote(kind),
|
||||
Keyword.put(aggregate_opts, :field, field)
|
||||
case Api.aggregate(
|
||||
__MODULE__,
|
||||
query,
|
||||
{unquote(kind), unquote(kind), Keyword.put(aggregate_opts, :field, field)},
|
||||
opts
|
||||
) do
|
||||
{:ok, aggregate} ->
|
||||
case Api.aggregate(__MODULE__, query, aggregate, opts) do
|
||||
{:ok, %{unquote(kind) => value}} ->
|
||||
value
|
||||
|
||||
{:error, error} ->
|
||||
raise Ash.Error.to_error_class(error)
|
||||
end
|
||||
{:ok, %{unquote(kind) => value}} ->
|
||||
value
|
||||
|
||||
{:error, error} ->
|
||||
raise Ash.Error.to_error_class(error)
|
||||
|
|
Loading…
Reference in a new issue