defmodule AshPostgres do @using_opts_schema Ashton.schema( opts: [ repo: :atom ], required: [:repo], describe: [ repo: "The repo that will be used to fetch your data. See the `Ecto.Repo` documentation for more" ], constraints: [ repo: {&AshPostgres.postgres_repo?/1, "must be using the postgres adapter"} ] ) @moduledoc """ A postgres data layer that levereges Ecto's postgres tools. To use it, add `use AshPostgres, repo: MyRepo` to your resource, after `use Ash.Resource` #{Ashton.document(@using_opts_schema)} """ @behaviour Ash.DataLayer defmacro __using__(opts) do quote bind_quoted: [opts: opts] do opts = AshPostgres.validate_using_opts(__MODULE__, opts) @data_layer AshPostgres @repo opts[:repo] def repo() do @repo end end end def validate_using_opts(mod, opts) do case Ashton.validate(opts, @using_opts_schema) do {:ok, opts} -> opts {:error, [{key, message} | _]} -> raise Ash.Error.ResourceDslError, resource: mod, using: __MODULE__, option: key, message: message end end def postgres_repo?(repo) do repo.__adapter__() == Ecto.Adapters.Postgres end def repo(resource) do resource.repo() end import Ecto.Query, only: [from: 2] @impl true def can?(:query_async), do: false def can?(:transact), do: false @impl true def limit(query, limit, _resource) do {:ok, from(row in query, limit: ^limit)} end @impl true def offset(query, offset, _resource) do {:ok, from(row in query, offset: ^offset)} end @impl true def run_query(query, resource) do {:ok, repo(resource).all(query)} end @impl true def resource_to_query(resource), do: Ecto.Queryable.to_query(resource) @impl true def create(resource, changeset) do repo = repo(resource) repo.transaction(fn -> # changeset = cast_assocs(changeset, repo, resource, ) result = try do changeset |> repo.insert() |> case do {:ok, result} -> result {:error, changeset} -> repo.rollback(changeset) end rescue e -> repo.rollback(e) end case changeset do %{__after_action__: [_ | _] = after_action_hooks} -> Enum.each(after_action_hooks, fn hook -> case hook.(changeset, result, repo) do :ok -> :ok {:error, error} -> repo.rollback(error) :error -> repo.rollback(:error) end end) result _ -> result end end) end @impl true def sort(query, sort, _resource) do {:ok, from(row in query, order_by: ^sort )} end @impl true def filter(query, filter, resource) do Enum.reduce(filter, {:ok, query}, fn _, {:error, error} -> {:error, error} {key, value}, {:ok, query} -> do_filter(query, key, value, resource) end) end defp do_filter( query, :from_related, {records, %{cardinality: :many_to_many} = relationship}, _resource ) do ids = Enum.map(records, &Map.get(&1, relationship.source_field)) {:ok, from(row in query, join: join_row in ^relationship.through, on: field(join_row, ^relationship.destination_field_on_join_table) == field(row, ^relationship.destination_field), where: field(join_row, ^relationship.source_field_on_join_table) in ^ids, select_merge: %{__related_id__: field(join_row, ^relationship.source_field_on_join_table)} )} end defp do_filter(query, :from_related, {records, relationship}, _resource) do ids = Enum.map(records, &Map.get(&1, relationship.source_field)) {:ok, from(row in query, where: field(row, ^relationship.destination_field) in ^ids )} end defp do_filter(query, key, value, resource) do cond do attr = Ash.attribute(resource, key) -> filter_attribute(query, attr, value, resource) rel = Ash.relationship(resource, key) -> filter_relationship(query, rel, value, resource) true -> {:error, "No such filter"} end end defp filter_attribute(query, attribute, value, _resource) do query = from(row in query, where: field(row, ^attribute.name) == ^value ) {:ok, query} end # Only supports a single id for now defp filter_relationship(query, %{name: name}, id, _resource) do query = from(row in query, join: related in assoc(row, ^name), where: related.id == ^id ) {:ok, query} end @impl true def can_query_async?(resource) do repo(resource).in_transaction?() end # def cast_assocs(changeset, repo, resource, relationships) do # Enum.reduce(relationships, changeset, fn {relationship, value}, changeset -> # case Ash.relationship(resource, relationship) do # %{type: :belongs_to, source_field: source_field} -> # belongs_to_assoc_update(changeset, source_field, value) # %{type: :has_one} = rel -> # has_one_assoc_update(changeset, repo, rel, value) # %{type: :has_many} = rel -> # has_many_assoc_update(changeset, rel, value) # %{type: :many_to_many} = rel -> # many_to_many_assoc_update(changeset, rel, value, repo) # _ -> # changeset # end # end) # end # defp has_one_assoc_update( # changeset, # repo, # %{ # destination: destination, # destination_field: destination_field, # source_field: source_field # }, # identifier # ) do # Ecto.Changeset.prepare_changes(changeset, fn changeset -> # value = # case identifier do # %{id: id} -> id # nil -> nil # _ -> raise "what" # end # query = # from(row in destination, # where: # field(row, ^destination_field) == ^Ecto.Changeset.get_field(changeset, source_field) # ) # repo.update_all(query, set: [{destination_field, value}]) # changeset # end) # end # defp belongs_to_assoc_update(changeset, source_field, %{id: id}) do # Ecto.Changeset.cast(changeset, %{source_field => id}, [source_field]) # end # defp belongs_to_assoc_update(changeset, source_field, nil) do # Ecto.Changeset.cast(changeset, %{source_field => nil}, [source_field]) # end # defp has_many_assoc_update( # changeset, # %{ # destination: destination, # destination_field: destination_field, # source_field: source_field # }, # values # ) do # ids = values |> Enum.map(&Map.get(&1, :id)) |> Enum.reject(&is_nil/1) # add_after_action_hook(changeset, fn _changeset, result, repo -> # field_value = Map.get(result, source_field) # query = # from(row in destination, # where: row.id in ^ids # ) # repo.update_all(query, set: [{destination_field, field_value}]) # :ok # end) # end # defp many_to_many_assoc_update( # changeset, # %{ # through: through, # source_field: source_field, # source_field_on_join_table: source_field_on_join_table, # destination_field_on_join_table: destination_field_on_join_table, # join_table_fields: join_table_fields # }, # values, # repo # ) do # ids = values |> Enum.map(&Map.get(&1, :id)) |> Enum.reject(&is_nil/1) # source_id = # Ecto.Changeset.get_field( # changeset, # source_field # ) # changeset # |> Ecto.Changeset.prepare_changes(fn changeset -> # delete_now_unrelated_ids( # repo, # source_id, # through, # source_field_on_join_table, # destination_field_on_join_table, # ids # ) # changeset # end) # |> add_after_action_hook(fn _changeset, _result, repo -> # case upsert_join_table_rows( # repo, # source_id, # through, # values, # source_field_on_join_table, # destination_field_on_join_table, # join_table_fields # ) do # :ok -> # :ok # end # end) # end # defp delete_now_unrelated_ids( # repo, # source_id, # through, # source_field_on_join_table, # destination_field_on_join_table, # ids # ) do # query = # from(join_row in through, # where: field(join_row, ^destination_field_on_join_table) not in ^ids, # where: field(join_row, ^source_field_on_join_table) == ^source_id # ) # repo.delete_all(query) # end # defp upsert_join_table_rows( # repo, # source_id, # through, # values, # source_field_on_join_table, # destination_field_on_join_table, # join_table_fields # ) do # values # |> Enum.map(fn fields -> # update_fields = Map.delete(fields, :id) # cond do # is_nil(Map.get(fields, :id)) -> # {:error, "No destination id specified"} # is_bitstring(through) -> # {:error, "Cannot update the fields of a join table only specified as a table name"} # Map.keys(update_fields) -- join_table_fields != [] -> # {:error, "no such fields to update"} # true -> # attributes = # update_fields # |> Map.put(source_field_on_join_table, source_id) # |> Map.put(destination_field_on_join_table, Map.get(fields, :id)) # changeset = # through # |> struct() # |> Ecto.Changeset.cast( # attributes, # join_table_fields ++ [source_field_on_join_table, destination_field_on_join_table] # ) # repo.insert(changeset, # on_conflict: :replace_all_except_primary_key, # conflict_target: [source_field_on_join_table, destination_field_on_join_table] # ) # end # end) # |> Enum.filter(fn result -> match?({:error, _}, result) end) # |> case do # [] -> :ok # errors -> {:error, errors} # end # end # defp add_after_action_hook(changeset, hook) do # changeset # |> Map.put_new(:__after_action__, []) # |> Map.update!(:__after_action__, fn list -> [hook | list] end) # end # Copied from an older file, to be added as more functionality is added back in # @impl true # def update(%resource{} = record, _action, attributes, relationships, _params) do # @repo.transaction(fn -> # changeset = # record # |> Ecto.Changeset.cast(attributes, Map.keys(attributes)) # |> AshPostgres.DataLayer.cast_assocs(@repo, resource, relationships) # result = # case @repo.update(changeset) do # {:ok, result} -> result # {:error, changeset} -> @repo.rollback(changeset) # end # case changeset do # %{__after_action__: [_ | _] = after_action_hooks} -> # Enum.each(after_action_hooks, fn hook -> # case hook.(changeset, result, @repo) do # :ok -> :ok # {:error, error} -> @repo.rollback(error) # :error -> @repo.rollback(:error) # end # end) # result # _other -> # result # end # end) # end # @impl true # def append_related(record, relationship, resource_identifiers) do # @repo.transaction(fn -> # AshPostgres.DataLayer.append_related(@repo, record, relationship, resource_identifiers) # end) # end # @impl true # def delete_related(record, relationship, resource_identifiers) do # @repo.transaction(fn -> # AshPostgres.DataLayer.delete_related(@repo, record, relationship, resource_identifiers) # end) # end # @impl true # def replace_related(record, relationship, resource_identifiers) do # @repo.transaction(fn -> # AshPostgres.DataLayer.replace_related(@repo, record, relationship, resource_identifiers) # end) # end # @impl true # def delete(record, _action, _params) do # @repo.delete(record) # end # @doc false # def replace_related( # repo, # record, # %{ # type: :many_to_many, # through: through, # source_field: source_field, # source_field_on_join_table: source_field_on_join_table, # destination_field_on_join_table: destination_field_on_join_table # }, # identifiers # ) do # ids = identifiers |> Enum.map(&Map.get(&1, :id)) |> Enum.reject(&is_nil/1) # source_id = Map.get(record, source_field) # delete_now_unrelated_ids( # repo, # source_id, # through, # source_field_on_join_table, # destination_field_on_join_table, # ids # ) # upsert_join_table_rows( # repo, # source_id, # through, # identifiers, # source_field_on_join_table, # destination_field_on_join_table # ) # record # end # def replace_related( # repo, # record, # %{ # type: :has_many, # source_field: source_field, # destination: destination, # destination_field: destination_field # }, # identifiers # ) do # ids = identifiers |> Enum.map(&Map.get(&1, :id)) |> Enum.reject(&is_nil/1) # field_value = Map.get(record, source_field) # query = # from(row in destination, # where: row.id in ^ids # ) # repo.update_all(query, set: [{destination_field, field_value}]) # record # end # def replace_related( # repo, # record, # %{ # type: :belongs_to, # source_field: source_field # }, # identifier # ) do # value = # case identifier do # %{id: id} -> id # nil -> nil # _ -> raise "what do" # end # record # |> Ecto.Changeset.cast(%{source_field => value}, [source_field]) # |> repo.update() # |> case do # {:ok, record} -> record # {:error, error} -> repo.rollback(error) # end # end # def replace_related( # repo, # record, # %{ # type: :has_one, # source_field: source_field, # destination_field: destination_field, # destination: destination # }, # identifier # ) do # value = # case identifier do # %{id: id} -> id # nil -> nil # _ -> raise "what" # end # query = # from(row in destination, # where: field(row, ^destination_field) == ^Map.get(record, source_field) # ) # repo.update_all(query, set: [{destination_field, value}]) # record # end # @doc false # def append_related(repo, record, %{type: :many_to_many} = relationship, identifiers) do # source_id = Map.get(record, relationship.source_field) # upsert_join_table_rows( # repo, # source_id, # relationship.through, # identifiers, # relationship.source_field_on_join_table, # relationship.destination_field_on_join_table # ) # record # end # def append_related( # repo, # record, # %{type: :has_many, destination: destination, destination_field: destination_field}, # identifiers # ) do # ids = # identifiers # |> Enum.map(&Map.get(&1, :id)) # |> Enum.reject(&is_nil/1) # query = # from(related in destination, # where: related.id in ^ids, # where: field(related, ^destination_field) != ^record.id # ) # repo.update_all(query, set: [{destination_field, record.id}]) # end # @doc false # def delete_related(repo, record, %{type: :many_to_many} = relationship, identifiers) do # source_id = Map.get(record, relationship.source_field) # ids = # identifiers # |> Enum.map(&Map.get(&1, :id)) # |> Enum.reject(&is_nil/1) # delete_related_ids( # repo, # source_id, # relationship.through, # relationship.source_field_on_join_table, # relationship.destination_field_on_join_table, # ids # ) # record # end # def delete_related( # repo, # record, # %{type: :has_many, destination: destination, destination_field: destination_field}, # identifiers # ) do # ids = # identifiers # |> Enum.map(&Map.get(&1, :id)) # |> Enum.reject(&is_nil/1) # query = # from(related in destination, # where: related.id in ^ids, # where: field(related, ^destination_field) != ^record.id # ) # # TODO: Validate the a delete_related action doesn't exist for has_many relationships # # where the destination field is not nullable. That will only ever error. # repo.update_all(query, set: [{destination_field, record.id}]) # end # defp delete_related_ids( # repo, # source_id, # through, # source_field_on_join_table, # destination_field_on_join_table, # ids # ) do # query = # from(join_row in through, # where: field(join_row, ^destination_field_on_join_table) in ^ids, # where: field(join_row, ^source_field_on_join_table) == ^source_id # ) # repo.delete_all(query) # end # def add_through_schema_fields(repo, resource, includes) when is_list(includes) do # Enum.flat_map(includes, fn {rel, further} -> # case Ash.relationship(resource, rel) do # %{type: :many_to_many, destination: destination} = relationship -> # [ # {rel, &fetch_and_add_through_row(&1, repo, relationship)}, # {rel, add_through_schema_fields(repo, destination, further)} # ] # %{destination: destination} -> # [ # {rel, add_through_schema_fields(repo, destination, further)} # ] # end # end) # end # def add_through_schema_fields(_repo, _resource, includes), do: includes # defp fetch_and_add_through_row(source_ids, repo, relationship) do # query = # from(join_row in relationship.through, # where: # type(field(join_row, ^relationship.source_field_on_join_table), :binary_id) in ^source_ids, # join: destination_row in ^relationship.destination, # on: # type(field(join_row, ^relationship.destination_field_on_join_table), :binary_id) == # field(destination_row, ^relationship.destination_field) # ) # query # |> add_select(relationship) # |> repo.all() # |> Enum.map(fn {destination_row, join_row} -> # {Map.get(join_row, relationship.source_field_on_join_table), # Map.put(destination_row, :__join_row__, join_row)} # end) # end # defp add_select(query, relationship) do # case relationship.through do # string when is_bitstring(string) -> # from([join_row, destination_row] in query, # select: # {destination_row, # %{ # ^relationship.source_field_on_join_table => # type(field(join_row, ^relationship.source_field_on_join_table), :binary_id), # ^relationship.destination_field_on_join_table => # type(field(join_row, ^relationship.destination_field_on_join_table), :binary_id) # }} # ) # module when is_atom(module) -> # from([join_row, destination_row] in query, # select: {destination_row, join_row} # ) # end # end end