defmodule Ecto.Adapters.DynamoDB do @moduledoc """ Ecto adapter for Amazon DynamoDB Currently for a fairly limited subset of Ecto, enough for basic operations. """ #NOTE: in ecto, Repo.get[!] ends up calling: #-> querable.get #-> queryable.one #-> queryable.all #-> queryable.execute #-> adapter.execute (possibly prepare somewhere in their too? trace.) @behaviour Ecto.Adapter #@behaviour Ecto.Adapter.Storage #@behaviour Ecto.Adapter.Migration defmacro __before_compile__(_env) do # Nothing to see here, yet... end use Bitwise, only_operators: true alias ExAws.Dynamo alias Ecto.Query.BooleanExpr # I don't think this is necessary: Probably under child_spec and ensure_all_started def start_link(repo, opts) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.start_link", %{"#{inspect __MODULE__}.start_link-params" => %{repo: repo, opts: opts}}) Agent.start_link fn -> [] end end ## Adapter behaviour - defined in lib/ecto/adapter.ex (in the ecto github repository) @doc """ Returns the childspec that starts the adapter process. """ def child_spec(repo, opts) do # TODO: need something here... # * Pull dynamo db connection options from config # * Start dynamo connector/aws libraries # we'll return our own start_link for now, but I don't think we actually need # an app here, we only need to ensure that our dependencies such as aws libs are started. # [:debug_requests, :access_key_id, :secret_access_key, :region, :dynamodb] |> Enum.map(fn key -> if opts[key] != nil, do: Application.put_env(:ex_aws, key, opts[key]) end) import Supervisor.Spec child_spec = worker(__MODULE__, [repo, opts]) ecto_dynamo_log(:debug, "#{inspect __MODULE__}.child_spec", %{"#{inspect __MODULE__}.child_spec-params" => %{repo: repo, child_spec: child_spec, opts: opts}}) child_spec end @doc """ Ensure all applications necessary to run the adapter are started. """ def ensure_all_started(repo, type) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.ensure_all_started", %{"#{inspect __MODULE__}.ensure_all_started-params" => %{type: type, repo: repo}}) with {:ok, _} = Application.ensure_all_started(:ecto_adapters_dynamodb) do {:ok, [repo]} end end def supports_ddl_transaction?, do: false def execute_ddl(repo, command, options) do Ecto.Adapters.DynamoDB.Migration.execute_ddl(repo, command, options) end # moved to transaction.ex in ecto 2.1.4 # def in_transaction?(_repo), do: false # # def rollback(_repo, _value), do: # raise BadFunctionError, message: "#{inspect __MODULE__} does not support transactions." @doc """ Called to autogenerate a value for id/embed_id/binary_id. Returns the autogenerated value, or nil if it must be autogenerated inside the storage or raise if not supported. For the Ecto type, `:id`, the adapter autogenerates a 128-bit integer For the Ecto type, `:embed_id`, the adapter autogenerates a string, using `Ecto.UUID.generate()` For the Ecto type, `:binary_id`, the adapter autogenerates a string, using `Ecto.UUID.generate()` """ @max_id ((1 <<< 128) - 1) # biggest possible int in 128 bits def autogenerate(:id), do: Enum.random(1..@max_id) def autogenerate(:embed_id), do: Ecto.UUID.generate() def autogenerate(:binary_id), do: Ecto.UUID.generate() @doc """ Returns the loaders for a given type. Rather than use the Ecto adapter loaders callback, the adapter builds on ExAws' decoding functionality, please see ExAws's `ExAws.Dynamo.decode_item` and the private function, `custom_decode`, in this module, which at this time only loads :utc_datetime and :naive_datetime. """ def loaders(_primitive, type), do: [type] @doc """ Returns the dumpers for a given type. We rely on ExAws encoding functionality during insertion and update to properly format types for DynamoDB. Please see ExAws `ExAws.Dynamo.update_item` and `ExAws.Dynamo.put_item` for specifics. Currently, we only modify :utc_datetime and :naive_datetime, appending the UTC offset, "Z", to the datetime string before passing to ExAws. """ def dumpers(:utc_datetime, datetime), do: [datetime, &to_iso_string/1] def dumpers(:naive_datetime, datetime), do: [datetime, &to_iso_string/1] def dumpers(_primitive, type), do: [type] # Add UTC offset # We are adding the offset here also for the :naive_datetime, this # assumes we are getting a UTC date (which does correspond with the # timestamps() macro but not necessarily with :naive_datetime in general) defp to_iso_string(datetime) do {:ok, (datetime |> Ecto.DateTime.cast! |> Ecto.DateTime.to_iso8601) <> "Z"} end @doc """ Commands invoked to prepare a query for `all`, `update_all` and `delete_all`. The returned result is given to `execute/6`. """ #@callback prepare(atom :: :all | :update_all | :delete_all, query :: Ecto.Query.t) :: # {:cache, prepared} | {:nocache, prepared} def prepare(:all, query) do # 'preparing' is more a SQL concept - Do we really need to do anything here or just pass the params through? ecto_dynamo_log(:debug, "#{inspect __MODULE__}.prepare: :all", %{"#{inspect __MODULE__}.prepare-params" => %{query: inspect(query, structs: false)}}) {:nocache, {:all, query}} end def prepare(:update_all, query) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.prepare: :update_all", %{"#{inspect __MODULE__}.prepare-params" => %{query: inspect(query, structs: false)}}) {:nocache, {:update_all, query}} end # do: {:cache, {System.unique_integer([:positive]), @conn.update_all(query)}} def prepare(:delete_all, query) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.prepare: :delete_all", %{"#{inspect __MODULE__}.prepare-params" => %{query: inspect(query, structs: false)}}) {:nocache, {:delete_all, query}} end # do: {:cache, {System.unique_integer([:positive]), @conn.delete_all(query)}} @doc """ Executes a previously prepared query. It must return a tuple containing the number of entries and the result set as a list of lists. The result set may also be `nil` if a particular operation does not support them. The `meta` field is a map containing some of the fields found in the `Ecto.Query` struct. It receives a process function that should be invoked for each selected field in the query result in order to convert them to the expected Ecto type. The `process` function will be nil if no result set is expected from the query. """ #@callback execute(repo, query_meta, query, params :: list(), process | nil, options) :: result when # result: {integer, [[term]] | nil} | no_return, # query: {:nocache, prepared} | # {:cached, (prepared -> :ok), cached} | # {:cache, (cached -> :ok), prepared} def execute(repo, meta, {:nocache, {func, prepared}}, params, process, opts) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.execute", %{"#{inspect __MODULE__}.execute-params" => %{repo: repo, meta: meta, prepared: prepared, params: params, process: process, opts: opts}}) {table, model} = prepared.from validate_where_clauses!(prepared) lookup_fields = extract_lookup_fields(prepared.wheres, params, []) limit_option = opts[:scan_limit] scan_limit = if is_integer(limit_option), do: [limit: limit_option], else: [] # Ecto migration does not know to specify 'scan: true' to retrieve the persisted migration versions # from line 34, file "deps/ecto/lib/ecto/migration/schema_migration.ex" migration_source = Keyword.get(repo.config, :migration_source, "schema_migrations") updated_opts = if table == migration_source do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.execute: table name corresponds with migration source: #{inspect migration_source}. Setting options for recursive scan.", %{}) Keyword.drop(opts, [:timeout, :log]) ++ [recursive: true] else Keyword.drop(opts, [:scan_limit, :limit]) ++ scan_limit end ecto_dynamo_log(:debug, "#{inspect __MODULE__}.execute: local variables", %{"#{inspect __MODULE__}.execute-vars" => %{table: table, lookup_fields: lookup_fields, scan_limit: scan_limit}}) case func do :delete_all -> delete_all(table, lookup_fields, updated_opts) :update_all -> update_all(table, lookup_fields, updated_opts, prepared.updates, params) :all -> ecto_dynamo_log(:info, "#{inspect __MODULE__}.execute: :all", %{"#{inspect __MODULE__}.execute-all-vars" => %{table: table, lookup_fields: lookup_fields, updated_opts: updated_opts}}) result = Ecto.Adapters.DynamoDB.Query.get_item(table, lookup_fields, updated_opts) ecto_dynamo_log(:debug, "#{inspect __MODULE__}.execute: all: result", %{"#{inspect __MODULE__}.execute-all-result" => inspect result}) if opts[:query_info_key], do: Ecto.Adapters.DynamoDB.QueryInfo.put(opts[:query_info_key], extract_query_info(result)) if result == %{} do # Empty map means "not found" {0, []} else sources = model.__schema__(:fields) |> Enum.into(%{}, fn f -> {model.__schema__(:field_source, f), f} end) cond do !result["Count"] and !result["Responses"] -> decoded = decode_item(result["Item"], model, sources, prepared.select) {1, [decoded]} true -> # batch_get_item returns "Responses" rather than "Items" results_to_decode = if result["Items"], do: result["Items"], else: result["Responses"][table] decoded = Enum.map(results_to_decode, &(decode_item(&1, model, sources, prepared.select))) {length(decoded), decoded} end end end end # delete_all allows for the recursive option, scanning through multiple pages defp delete_all(table, lookup_fields, opts) do ecto_dynamo_log(:info, "#{inspect __MODULE__}.delete_all", %{"#{inspect __MODULE__}.delete_all-params" => %{table: table, lookup_fields: lookup_fields, opts: opts}}) # select only the key {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(table) scan_or_query = Ecto.Adapters.DynamoDB.Query.scan_or_query?(table, lookup_fields) recursive = Ecto.Adapters.DynamoDB.Query.parse_recursive_option(scan_or_query, opts) updated_opts = prepare_recursive_opts(opts ++ [projection_expression: Enum.join(key_list, ", ")]) delete_all_recursive(table, lookup_fields, updated_opts, recursive, %{}, 0) end defp delete_all_recursive(table, lookup_fields, opts, recursive, query_info, total_processed) do # query the table for which records to delete fetch_result = Ecto.Adapters.DynamoDB.Query.get_item(table, lookup_fields, opts) ecto_dynamo_log(:debug, "#{inspect __MODULE__}.delete_all_recursive: fetch_result", %{"#{inspect __MODULE__}.delete_all_recursive-fetch_result" => inspect fetch_result}) items = case fetch_result do %{"Items" => fetch_items} -> fetch_items %{"Item" => item} -> [item] %{"Responses" => table_map} -> table_map[table] _ -> [] end prepared_data = for key_list <- Enum.map(items, &Map.to_list/1) do key_map = for {key, val_map} <- key_list, into: %{}, do: {key, Dynamo.Decoder.decode(val_map)} [delete_request: [key: key_map]] end unprocessed_items = if prepared_data != [] do batch_delete(table, prepared_data) else %{} end num_processed = length(prepared_data) - if !unprocessed_items[table], do: 0, else: length(unprocessed_items[table]) updated_query_info = Enum.reduce(fetch_result, query_info, fn({key, val}, acc) -> case key do "Count" -> Map.update(acc, key, val, fn x -> x + val end) "ScannedCount" -> Map.update(acc, key, val, fn x -> x + val end) "LastEvaluatedKey" -> Map.update(acc, key, val, fn _ -> fetch_result["LastEvaluatedKey"] end) _ -> acc end end) |> Map.update("UnprocessedItems", unprocessed_items, fn map -> if map == %{}, do: %{}, else: %{table => map[table] ++ unprocessed_items[table]} end) updated_recursive = Ecto.Adapters.DynamoDB.Query.update_recursive_option(recursive) if fetch_result["LastEvaluatedKey"] != nil and updated_recursive.continue do opts_with_offset = opts ++ [exclusive_start_key: fetch_result["LastEvaluatedKey"]] delete_all_recursive(table, lookup_fields, opts_with_offset, updated_recursive.new_value, updated_query_info, total_processed + num_processed) else # We're not retrying unprocessed items yet, but we are providing the relevant info in the QueryInfo agent if :query_info_key is supplied if opts[:query_info_key], do: Ecto.Adapters.DynamoDB.QueryInfo.put(opts[:query_info_key], updated_query_info) {num_processed + total_processed, nil} end end # returns unprocessed_items defp batch_delete(table, prepared_data) do batch_write_attempt = Dynamo.batch_write_item(%{table => prepared_data}) |> ExAws.request |> handle_error!(%{table: table, records: []}) batch_write_attempt["UnprocessedItems"] end defp update_all(table, lookup_fields, opts, updates, params) do ecto_dynamo_log(:info, "#{inspect __MODULE__}.update_all", %{"#{inspect __MODULE__}.update_all-params" => %{table: table, lookup_fields: lookup_fields, opts: opts}}) scan_or_query = Ecto.Adapters.DynamoDB.Query.scan_or_query?(table, lookup_fields) recursive = Ecto.Adapters.DynamoDB.Query.parse_recursive_option(scan_or_query, opts) key_list = Ecto.Adapters.DynamoDB.Info.primary_key!(table) ecto_dynamo_log(:debug, "#{inspect __MODULE__}.update_all: key_list", %{"#{inspect __MODULE__}.update_all-key_list" => inspect key_list}) # The remove statement must be constructed after finding pull-indexes, but it # also includes possibly removing nil fields, and since we have one handler for # both set and remove, we call it during the batch update process {update_expression, update_fields_sans_set_remove, set_remove_fields} = construct_update_expression(updates, params, opts) ecto_dynamo_log(:info, "#{inspect __MODULE__}.update_all: update fields", %{"#{inspect __MODULE__}.update_all-update_fields" => %{update_fields_sans_set_remove: inspect(update_fields_sans_set_remove), set_remove_fields: inspect(set_remove_fields)}}) attribute_names = construct_expression_attribute_names(update_fields_sans_set_remove) attribute_values = construct_expression_attribute_values(update_fields_sans_set_remove, opts) base_update_options = [expression_attribute_names: attribute_names, update_expression: update_expression, return_values: :all_new] updated_opts = prepare_recursive_opts(opts) update_options = maybe_add_attribute_values(base_update_options, attribute_values) pull_actions_without_index = Keyword.keys(set_remove_fields[:pull]) |> Enum.any?(fn x -> !Enum.member?(Keyword.keys(maybe_list(opts[:pull_indexes])), x) end) {new_update_options, new_set_remove_fields} = if pull_actions_without_index do {update_options, set_remove_fields} else merged_pull_indexes = Keyword.merge(set_remove_fields[:pull], maybe_list(opts[:pull_indexes])) opts_with_pull_indexes = Keyword.update(opts, :pull_indexes, merged_pull_indexes, fn _ -> merged_pull_indexes end) {update_batch_update_options(update_options, set_remove_fields, opts_with_pull_indexes), []} end update_all_recursive(table, lookup_fields, updated_opts, new_update_options, key_list, new_set_remove_fields, recursive, %{}, 0) end defp update_all_recursive(table, lookup_fields, opts, update_options, key_list, set_remove_fields, recursive, query_info, total_updated) do fetch_result = Ecto.Adapters.DynamoDB.Query.get_item(table, lookup_fields, opts) ecto_dynamo_log(:debug, "#{inspect __MODULE__}.update_all_recursive: fetch_result", %{"#{inspect __MODULE__}.update_all_recursive-fetch_result" => inspect(fetch_result)}) updated_query_info = case fetch_result do %{"Count" => last_count, "ScannedCount" => last_scanned_count} -> %{"Count" => last_count + Map.get(query_info, "Count", 0), "ScannedCount" => last_scanned_count + Map.get(query_info, "ScannedCount", 0), "LastEvaluatedKey" => Map.get(fetch_result, "LastEvaluatedKey")} _ -> query_info end items = case fetch_result do %{"Items" => fetch_items} -> fetch_items %{"Item" => item} -> [item] %{"Responses" => table_map} -> table_map[table] _ -> [] end num_updated = if items != [] do batch_update(table, items, key_list, update_options, set_remove_fields, opts) else 0 end updated_recursive = Ecto.Adapters.DynamoDB.Query.update_recursive_option(recursive) if fetch_result["LastEvaluatedKey"] != nil and updated_recursive.continue do opts_with_offset = opts ++ [exclusive_start_key: fetch_result["LastEvaluatedKey"]] update_all_recursive(table, lookup_fields, opts_with_offset, update_options, key_list, set_remove_fields, updated_recursive.new_value, updated_query_info, total_updated + num_updated) else if opts[:query_info_key], do: Ecto.Adapters.DynamoDB.QueryInfo.put(opts[:query_info_key], updated_query_info) {total_updated + num_updated, []} end end defp batch_update(table, items, key_list, update_options, set_remove_fields, opts) do Enum.reduce(items, 0, fn(result_to_update, acc) -> filters = get_key_values_dynamo_map(result_to_update, key_list) # we only update this on a case-by-case basis if pull actions # without specific indexes are specified options_with_set_and_remove = case set_remove_fields do [] -> update_options _ -> pull_fields_with_indexes = Enum.map(set_remove_fields[:pull], fn {field_atom, val} -> list = result_to_update[to_string(field_atom)] {field_atom, find_all_indexes_in_dynamodb_list(list, val)} end) merged_pull_indexes = Keyword.merge(pull_fields_with_indexes, maybe_list(opts[:pull_indexes])) opts_with_pull_indexes = Keyword.update(opts, :pull_indexes, merged_pull_indexes, fn _ -> merged_pull_indexes end) update_batch_update_options(update_options, set_remove_fields, opts_with_pull_indexes) end # 'options_with_set_and_remove' might not have the key, ':expression_attribute_values', # when there are only removal statements. record = if options_with_set_and_remove[:expression_attribute_values], do: [options_with_set_and_remove[:expression_attribute_values] |> Enum.into(%{})], else: [] if options_with_set_and_remove[:update_expression] |> String.trim != "" do Dynamo.update_item(table, filters, options_with_set_and_remove) |> ExAws.request |> handle_error!(%{table: table, records: record ++ []}) acc + 1 else acc end end) end defp update_batch_update_options(update_options, set_remove_fields, opts) do attribute_names = construct_expression_attribute_names(Keyword.values(set_remove_fields) |> List.flatten) set_and_push_fields = maybe_list(set_remove_fields[:set]) ++ maybe_list(set_remove_fields[:push]) opts_with_push = opts ++ Keyword.take(set_remove_fields, [:push]) attribute_values = construct_expression_attribute_values(set_and_push_fields, opts_with_push) set_statement = construct_set_statement(set_remove_fields[:set], opts_with_push) opts_for_construct_remove = Keyword.take(set_remove_fields, [:pull]) ++ Keyword.take(opts, [:pull_indexes, :remove_nil_fields]) remove_statement = construct_remove_statement(set_remove_fields[:set], opts_for_construct_remove) base_update_options = [expression_attribute_names: Map.merge(attribute_names, update_options[:expression_attribute_names]), update_expression: set_statement <> " " <> remove_statement <> " " <> update_options[:update_expression] |> String.trim, return_values: :all_new] maybe_add_attribute_values(base_update_options, attribute_values ++ maybe_list(update_options[:expression_attribute_values])) end # find indexes to remove for update :pull action defp find_all_indexes_in_dynamodb_list(dynamodb_list, target) do Dynamo.Decoder.decode(dynamodb_list) |> Enum.with_index() |> Enum.filter(fn {x, _} -> x == target end) |> Enum.map(fn {_, i} -> i end) end # During delete_all's and update_all's recursive # procedure, we want to keep the recursion in # the top-level, between actions, rather than # load all the results into memory and then act; # so we disable the recursion on get_item defp prepare_recursive_opts(opts) do opts |> Keyword.delete(:page_limit) |> Keyword.update(:recursive, false, fn _ -> false end) end @doc """ Inserts a single new struct in the data store. ## Autogenerate The primary key will be automatically included in `returning` if the field has type `:id` or `:binary_id` and no value was set by the developer or none was autogenerated by the adapter. """ #@callback insert(repo, schema_meta, fields, on_conflict, returning, options) :: # {:ok, fields} | {:invalid, constraints} | no_return # def insert(_,_,_,_,_) do def insert(repo, schema_meta, fields, on_conflict, returning, opts) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.insert", %{"#{inspect __MODULE__}.insert-params" => %{repo: repo, schema_meta: schema_meta, fields: fields, on_conflict: on_conflict, returning: returning, opts: opts}}) insert_nil_field_option = Keyword.get(opts, :insert_nil_fields, true) do_not_insert_nil_fields = insert_nil_field_option == false || Application.get_env(:ecto_adapters_dynamodb, :insert_nil_fields) == false {_, table} = schema_meta.source model = schema_meta.schema fields_map = Enum.into(fields, %{}) record = if do_not_insert_nil_fields, do: fields_map, else: build_record_map(model, fields_map) ecto_dynamo_log(:info, "#{inspect __MODULE__}.insert: local variables", %{"#{inspect __MODULE__}.insert-vars" => %{table: table, record: record}}) {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(table) hash_key = hd(key_list) on_conflict_action = elem(on_conflict, 0) options = case on_conflict_action do :replace_all -> [] _ -> attribute_names = for k <- key_list, into: %{}, do: {"##{k}", k} conditions = for k <- key_list, do: "attribute_not_exists(##{k})" condition_expression = Enum.join(conditions, " and ") [expression_attribute_names: attribute_names, condition_expression: condition_expression] end case Dynamo.put_item(table, record, options) |> ExAws.request |> handle_error!(%{table: table, records: [record]}) do {:error, "ConditionalCheckFailedException"} -> case on_conflict_action do # Per discussion with Jose Valim (https://github.com/elixir-ecto/ecto/issues/2378) # clarifying the adapter should return nothing if there is no `:returning` specified, # and what we thought was to be returned as a `nil` id, is only for cases where # "the field is autogenerated by the database" (https://hexdocs.pm/ecto/Ecto.Repo.html) :nothing -> {:ok, []} :raise -> # This constraint name yields the correct behavior in the case the user # has specified a unique constraint on the primary key in their schema: constraint_name = "#{table}_#{hash_key}_index" {:invalid, [unique: constraint_name]} end %{} -> {:ok, []} end end def insert_all(repo, schema_meta, field_list, fields, on_conflict, returning, opts) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.insert_all", %{"#{inspect __MODULE__}.insert_all-params" => %{repo: repo, schema_meta: schema_meta, field_list: field_list, fields: fields, on_conflict: on_conflict, returning: returning, opts: opts}}) insert_nil_field_option = Keyword.get(opts, :insert_nil_fields, true) do_not_insert_nil_fields = insert_nil_field_option == false || Application.get_env(:ecto_adapters_dynamodb, :insert_nil_fields) == false {_, table} = schema_meta.source model = schema_meta.schema prepared_fields = Enum.map(fields, fn(field_set) -> mapped_fields = Enum.into(field_set, %{}) record = if do_not_insert_nil_fields, do: mapped_fields, else: build_record_map(model, mapped_fields) [put_request: [item: record]] end) ecto_dynamo_log(:info, "#{inspect __MODULE__}.insert_all: local variables", %{"#{inspect __MODULE__}.insert_all-vars" => %{table: table, records: get_records_from_fields(prepared_fields)}}) batch_write(table, prepared_fields, opts) end # DynamoDB will reject an entire batch of insert_all() records if there are more than 25 requests. # https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchWriteItem.html # batch_write/3 will break the list into chunks of 25 items and insert each separately. defp batch_write(table, prepared_fields, opts) do batch_write_limit = 25 response_element = "UnprocessedItems" grouped_records = Enum.chunk_every(prepared_fields, batch_write_limit) num_batches = length grouped_records # Break the prepared_fields into chunks of at most 25 elements to be batch inserted, accumulating # the total count of records and appropriate results as it loops through the reduce. {total_processed, results} = grouped_records |> Stream.with_index |> Enum.reduce({0, []}, fn({field_group, i}, {running_total_processed, batch_write_results}) -> {total_batch_processed, batch_write_attempt} = handle_batch_write(field_group, table, response_element) # Log depth of 11 will capture the full data structure returned in any UnprocessedItems - https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchWriteItem.html ecto_dynamo_log(:debug, "#{inspect __MODULE__}.batch_write #{i + 1} of #{num_batches}: local variables", %{"#{inspect __MODULE__}.insert_all-batch_write" => %{table: table, field_group: field_group, results: batch_write_attempt}}, [depth: 11]) # We're not retrying unprocessed items yet, but we are providing the relevant info in the QueryInfo agent if :query_info_key is supplied if opts[:query_info_key] do query_info = extract_query_info(batch_write_attempt) Ecto.Adapters.DynamoDB.QueryInfo.update(opts[:query_info_key], [query_info], fn(list) -> list ++ [query_info] end) end {running_total_processed + total_batch_processed, batch_write_results ++ [batch_write_attempt]} end) result_body_for_log = %{table => Enum.flat_map(results, fn(res) -> res[response_element][table] || [] end)} ecto_dynamo_log(:info, "#{inspect __MODULE__}.batch_write: batch_write_attempt result", %{"#{inspect __MODULE__}.insert_all-batch_write" => inspect %{response_element => (if result_body_for_log[table] == [], do: %{}, else: result_body_for_log)}}) {total_processed, nil} end defp handle_batch_write(field_group, table, response_element) do results = Dynamo.batch_write_item(%{table => field_group}) |> ExAws.request |> handle_error!(%{table: table, records: get_records_from_fields(field_group)}) if results[response_element] == %{} do {length(field_group), results} else {length(field_group) - length(results[response_element][table]), results} end end defp get_records_from_fields(fields), do: Enum.map(fields, fn [put_request: [item: record]] -> record end) defp build_record_map(model, fields_to_insert) do # Ecto does not convert empty strings to nil before passing them # to Repo.insert_all, and ExAws will remove empty strings (as well as empty lists) # when building the insertion query but not nil values. We don't mind the removal # of empty lists since those cannot be inserted to indexed fields, but we'd like to # catch the removal of fields with empty strings by ExAws to support our option, :remove_nil_fields, # so we convert these to nil. fields = model.__schema__(:fields) sources = fields |> Enum.into(%{}, fn f -> {f, model.__schema__(:field_source, f)} end) empty_strings_to_nil = fields_to_insert |> Enum.map(fn {field, val} -> {field, (if val == "", do: nil, else: val)} end) |> Enum.into(%{}) model.__struct__ |> Map.delete(:__meta__) |> Map.from_struct |> Enum.reduce(%{}, fn {k, v}, acc -> Map.put(acc, Map.get(sources, k), v) end) |> Map.merge(empty_strings_to_nil) end # In testing, 'filters' contained only the primary key and value # TODO: handle cases of more than one tuple in 'filters'? def delete(repo, schema_meta, filters, opts) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.delete", %{"#{inspect __MODULE__}.delete-params" => %{repo: repo, schema_meta: schema_meta, filters: filters, opts: opts}}) {_, table} = schema_meta.source # We offer the :range_key option for tables with composite primary key # since Ecto will not provide the range_key value needed for the query. # If :range_key is not provided, check if the table has a composite # primary key and query for all the key values updated_filters = case opts[:range_key] do nil -> {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(table) if (length key_list) > 1 do updated_opts = opts ++ [projection_expression: Enum.join(key_list, ", ")] filters_as_strings = for {field, val} <- filters, do: {Atom.to_string(field), {val, :==}} fetch_result = Ecto.Adapters.DynamoDB.Query.get_item(table, filters_as_strings, updated_opts) items = case fetch_result do %{"Items" => fetch_items} -> fetch_items %{"Item" => item} -> [item] _ -> [] end if items == [], do: raise "__MODULE__.update error: no results found for record: #{inspect filters}" if (length items) > 1, do: raise "__MODULE__.update error: more than one result found for record: #{inspect filters}" for {field, key_map} <- Map.to_list(hd items) do [{_field_type, val}] = Map.to_list(key_map) {field, val} end else filters end range_key -> [range_key | filters] end attribute_names = construct_expression_attribute_names(keys_to_atoms(filters)) base_options = [expression_attribute_names: attribute_names] condition_expression = construct_condition_expression(filters) options = base_options ++ [condition_expression: condition_expression] # 'options' might not have the key, ':expression_attribute_values', when there are only removal statements record = if options[:expression_attribute_values], do: [options[:expression_attribute_values] |> Enum.into(%{})], else: [] case Dynamo.delete_item(table, updated_filters, options) |> ExAws.request |> handle_error!(%{table: table, records: record ++ []}) do %{} -> {:ok, []} {:error, "ConditionalCheckFailedException"} -> {:error, :stale} end end def update(repo, schema_meta, fields, filters, returning, opts) do ecto_dynamo_log(:debug, "#{inspect __MODULE__}.update", %{"#{inspect __MODULE__}.update-params" => %{repo: repo, schema_meta: schema_meta, fields: fields, filters: filters, returning: returning, opts: opts}}) {_, table} = schema_meta.source # We offer the :range_key option for tables with composite primary key # since Ecto will not provide the range_key value needed for the query. # If :range_key is not provided, check if the table has a composite # primary key and query for all the key values updated_filters = case opts[:range_key] do nil -> {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(table) if (length key_list) > 1 do updated_opts = opts ++ [projection_expression: Enum.join(key_list, ", ")] filters_as_strings = for {field, val} <- filters, do: {Atom.to_string(field), {val, :==}} fetch_result = Ecto.Adapters.DynamoDB.Query.get_item(table, filters_as_strings, updated_opts) items = case fetch_result do %{"Items" => fetch_items} -> fetch_items %{"Item" => item} -> [item] _ -> [] end if items == [], do: raise "__MODULE__.update error: no results found for record: #{inspect filters}" if (length items) > 1, do: raise "__MODULE__.update error: more than one result found for record: #{inspect filters}" for {field, key_map} <- Map.to_list(hd items) do [{_field_type, val}] = Map.to_list(key_map) {field, val} end else filters end range_key -> [range_key | filters] end update_expression = construct_update_expression(fields, opts) # add updated_filters to attribute_ names and values for condition_expression attribute_names = construct_expression_attribute_names(fields ++ keys_to_atoms(filters)) attribute_values = construct_expression_attribute_values(fields, opts) base_options = [expression_attribute_names: attribute_names, update_expression: update_expression] condition_expression = construct_condition_expression(filters) options = maybe_add_attribute_values(base_options, attribute_values) ++ [condition_expression: condition_expression] # 'options' might not have the key, ':expression_attribute_values', when there are only removal statements record = if options[:expression_attribute_values], do: [options[:expression_attribute_values] |> Enum.into(%{})], else: [] case Dynamo.update_item(table, updated_filters, options) |> ExAws.request |> handle_error!(%{table: table, records: record ++ []}) do %{} -> {:ok, []} {:error, "ConditionalCheckFailedException"} -> {:error, :stale} end end defp keys_to_atoms(list), do: for {k, v} <- list, do: {maybe_string_to_atom(k), v} defp maybe_string_to_atom(s), do: if is_binary(s), do: String.to_atom(s), else: s defp construct_condition_expression([{field, _val}] = _filters), do: "attribute_exists(##{to_string(field)})" defp extract_query_info(result), do: result |> Map.take(["Count", "ScannedCount", "LastEvaluatedKey", "UnprocessedItems", "UnprocessedKeys"]) # Used in update_all defp extract_update_params([], _action_atom, _params), do: [] defp extract_update_params([%{expr: key_list}], action_atom, params) do case key_list[action_atom] do nil -> [] action_list -> for s <- action_list do {field_atom, {:^, _, [idx]}} = s {field_atom, Enum.at(params,idx)} end end end defp extract_update_params([a], _action_atom, _params), do: error "#{inspect __MODULE__}.extract_update_params: Updates is either missing the :expr key or does not contain a struct or map: #{inspect a}" defp extract_update_params(unsupported, _action_atom, _params), do: error "#{inspect __MODULE__}.extract_update_params: unsupported parameter construction. #{inspect unsupported}" # Ecto does not support push pull for types other than array. # Therefore, we enable add and delete via opts defp extract_update_params(key_list, action_atom) do case key_list[action_atom] do nil -> [] action_list -> action_list end end # used in :update_all defp get_key_values_dynamo_map(dynamo_map, {:primary, keys}) do for k <- keys, do: {String.to_atom(k), Dynamo.Decoder.decode(dynamo_map[k])} end defp construct_expression_attribute_names(fields) do for {f, _} <- fields, into: %{}, do: {"##{Atom.to_string(f)}", Atom.to_string(f)} end defp construct_expression_attribute_values(fields, opts) do remove_rather_than_set_to_null = opts[:remove_nil_fields] || Application.get_env(:ecto_adapters_dynamodb, :remove_nil_fields_on_update) == true # If the value is nil and the :remove_nil_fields option is set, # we're removing this attribute, not updating it, so filter out any such fields: if remove_rather_than_set_to_null do for {k, v} <- fields, !is_nil(v), do: {k, format_val(k, v, opts)} else for {k, v} <- fields, do: {k, format_nil(k, v, opts)} end |> Enum.filter(fn {x, _} -> not Keyword.has_key?(maybe_list(opts[:pull]), x) end) end defp maybe_list(l) when is_list(l), do: l defp maybe_list(_), do: [] defp format_nil(_k, v, _opts) when is_nil(v), do: %{"NULL" => "true"} defp format_nil(k, v, opts), do: format_val(k, v, opts) defp format_val(k, v, opts) do case opts[:push][k] do nil -> v _ -> [v] end end # DynamoDB throws an error if we pass in an empty list for attribute values, # so we have to implement this stupid little helper function to avoid hurting # its feelings: defp maybe_add_attribute_values(options, []) do options end defp maybe_add_attribute_values(options, attribute_values) do [expression_attribute_values: attribute_values] ++ options end defp construct_update_expression(updates, params, opts) do to_set = extract_update_params(updates, :set, params) to_push = extract_update_params(updates, :push, params) to_pull = extract_update_params(updates, :pull, params) to_add = extract_update_params(opts, :add) ++ extract_update_params(updates, :inc, params) to_delete = extract_update_params(opts, :delete) {construct_add_statement(to_add, opts) <> " " <> construct_delete_statement(to_delete, opts) |> String.trim(), to_add ++ to_delete, [set: to_set, push: to_push, pull: to_pull]} end # The update callback supplies fields in the paramaters # whereas update_all includes a more complicated updates # structure defp construct_update_expression(fields, opts) do set_statement = construct_set_statement(fields, opts) rem_statement = construct_remove_statement(fields, opts) String.trim("#{set_statement} #{rem_statement}") end # fields::[{:field, val}] defp construct_set_statement(fields, opts) do remove_rather_than_set_to_null = opts[:remove_nil_fields] || Application.get_env(:ecto_adapters_dynamodb, :remove_nil_fields_on_update) == true set_clauses = for {key, val} <- fields, not (is_nil(val) and remove_rather_than_set_to_null) do key_str = Atom.to_string(key) "##{key_str}=:#{key_str}" end ++ case opts[:push] do nil -> [] push_list -> for {key, _val} <- push_list do key_str = Atom.to_string(key) if Enum.member?(maybe_list(opts[:prepend_to_list]), key), do: "##{key_str} = list_append(:#{key_str}, ##{key_str})", else: "##{key_str} = list_append(##{key_str}, :#{key_str})" end end case set_clauses do [] -> "" _ -> "SET " <> Enum.join(set_clauses, ", ") end end defp construct_remove_statement(fields, opts) do remove_rather_than_set_to_null = opts[:remove_nil_fields] || Application.get_env(:ecto_adapters_dynamodb, :remove_nil_fields_on_update) == true remove_clauses = if remove_rather_than_set_to_null do for {key, val} <- fields, is_nil(val), do: "##{Atom.to_string(key)}" else [] end # Ecto :pull update can be emulated provided # we are given an index to remove in opts[:pull_indexes] ++ cond do !opts[:pull_indexes] or (Keyword.values(opts[:pull_indexes]) |> List.flatten) == [] -> [] opts[:pull] == nil -> [] true -> for {key, _val} <- opts[:pull] do key_str = Atom.to_string(key) Enum.map(opts[:pull_indexes][key], fn index -> "##{key_str}[#{index}]" end) |> Enum.join(", ") end end case remove_clauses do [] -> "" _ -> "REMOVE " <> Enum.join(remove_clauses, ", ") end end # fields::[{:field, val}] defp construct_add_statement(fields, _opts) do add_clauses = for {key, _val} <- fields do key_str = Atom.to_string(key) "##{key_str} :#{key_str}" end case add_clauses do [] -> "" _ -> "ADD " <> Enum.join(add_clauses, ", ") end end defp construct_delete_statement(fields, _opts) do delete_clauses = for {key, _val} <- fields do key_str = Atom.to_string(key) "##{key_str} :#{key_str}" end case delete_clauses do [] -> "" _ -> "DELETE " <> Enum.join(delete_clauses, ", ") end end defp validate_where_clauses!(query) do for w <- query.wheres do validate_where_clause! w end end defp validate_where_clause!(%BooleanExpr{expr: {op, _, _}}) when op in [:==, :<, :>, :<=, :>=, :in], do: :ok defp validate_where_clause!(%BooleanExpr{expr: {logical_op, _, _}}) when logical_op in [:and, :or], do: :ok defp validate_where_clause!(%BooleanExpr{expr: {:is_nil, _, _}}), do: :ok defp validate_where_clause!(%BooleanExpr{expr: {:fragment, _, _}}), do: :ok defp validate_where_clause!(unsupported), do: error "unsupported where clause: #{inspect unsupported}" # We are parsing a nested, recursive structure of the general type: # %{:logical_op, list_of_clauses} | %{:conditional_op, field_and_value} defp extract_lookup_fields([], _params, lookup_fields), do: lookup_fields defp extract_lookup_fields([query | queries], params, lookup_fields) do # A logical operator tuple does not always have a parent 'expr' key. maybe_extract_from_expr = case query do %BooleanExpr{expr: expr} -> expr # TODO: could there be other cases? _ -> query end case maybe_extract_from_expr do # A logical operator points to a list of conditionals {op, _, [left, right]} when op in [:==, :<, :>, :<=, :>=, :in] -> {field, value} = get_op_clause(left, right, params) updated_lookup_fields = case List.keyfind(lookup_fields, field, 0) do # we assume the most ops we can apply to one field is two, otherwise this might throw an error {field, {old_val, old_op}} -> List.keyreplace(lookup_fields, field, 0, {field, {[value, old_val], [op, old_op]}}) _ -> [{field, {value, op}} | lookup_fields] end extract_lookup_fields(queries, params, updated_lookup_fields) # Logical operator expressions have more than one op clause # We are matching queries of the type: 'from(p in Person, where: p.email == "g@email.com" and p.first_name == "George")' # But not of the type: 'from(p in Person, where: [email: "g@email.com", first_name: "George"])' # # A logical operator is a member of a list {logical_op, _, clauses} when logical_op in [:and, :or] -> deeper_lookup_fields = extract_lookup_fields(clauses, params, []) extract_lookup_fields(queries, params, [{logical_op, deeper_lookup_fields} | lookup_fields]) {:fragment, _, raw_expr_mixed_list} -> parsed_fragment = parse_raw_expr_mixed_list(raw_expr_mixed_list, params) extract_lookup_fields(queries, params, [parsed_fragment | lookup_fields]) # We perform a post-query is_nil filter on indexed fields and have DynamoDB filter # for nil non-indexed fields (although post-query nil-filters on (missing) indexed # attributes could only find matches when the attributes are not the range part of # a queried partition key (hash part) since those would not return the sought records). {:is_nil, _, [arg]} -> {{:., _, [_, field_name]}, _, _} = arg # We give the nil value a string, "null", since it will be mapped as a DynamoDB attribute_expression_value extract_lookup_fields(queries, params, [{to_string(field_name), {"null", :is_nil}} | lookup_fields]) _ -> extract_lookup_fields(queries, params, lookup_fields) end end # Specific (as opposed to generalized) parsing for Ecto :fragment - the only use for it # so far is 'between' which is the only way to query 'between' on an indexed field since # those accept only single conditions. # # Example with values as strings: [raw: "", expr: {{:., [], [{:&, [], [0]}, :person_id]}, [], []}, raw: " between ", expr: "person:a", raw: " and ", expr: "person:f", raw: ""] # # Example with values as part of the string itself: [raw: "", expr: {{:., [], [{:&, [], [0]}, :person_id]}, [], []}, raw: " between person:a and person:f"] # # Example with values in params: [raw: "", expr: {{:., [], [{:&, [], [0]}, :person_id]}, [], []}, raw: " between ", expr: {:^, [], [0]}, raw: " and ", expr: {:^, [], [1]}, raw: ""] # defp parse_raw_expr_mixed_list(raw_expr_mixed_list, params) do # group the expression into fields, values, and operators, # only supporting the example with values in params case raw_expr_mixed_list do # between [raw: _, expr: {{:., [], [{:&, [], [0]}, field_atom]}, [], []}, raw: between_str, expr: {:^, [], [idx1]}, raw: and_str, expr: {:^, [], [idx2]}, raw: _] -> if not (Regex.match?(~r/^\s*between\s*and\s*$/i, between_str <> and_str)), do: parse_raw_expr_mixed_list_error(raw_expr_mixed_list) {to_string(field_atom), {[Enum.at(params, idx1), Enum.at(params, idx2)], :between}} # begins_with [raw: begins_with_str, expr: {{:., [], [{:&, [], [0]}, field_atom]}, [], []}, raw: comma_str, expr: {:^, [], [idx]}, raw: closing_parenthesis_str] -> if not (Regex.match?(~r/^\s*begins_with\(\s*,\s*\)\s*$/i, begins_with_str <> comma_str <> closing_parenthesis_str)), do: parse_raw_expr_mixed_list_error(raw_expr_mixed_list) {to_string(field_atom), {Enum.at(params, idx), :begins_with}} _ -> parse_raw_expr_mixed_list_error(raw_expr_mixed_list) end end defp parse_raw_expr_mixed_list_error(raw_expr_mixed_list), do: raise "#{inspect __MODULE__}.parse_raw_expr_mixed_list parse error. We currently only support the Ecto fragments of the form, 'where: fragment(\"? between ? and ?\", FIELD_AS_VARIABLE, VALUE_AS_VARIABLE, VALUE_AS_VARIABLE)'; and 'where: fragment(\"begins_with(?, ?)\", FIELD_AS_VARIABLE, VALUE_AS_VARIABLE)'. Received: #{inspect raw_expr_mixed_list}" defp get_op_clause(left, right, params) do field = left |> get_field |> Atom.to_string value = get_value(right, params) {field, value} end defp get_field({{:., _, [{:&, _, [0]}, field]}, _, []}), do: field defp get_field(other_clause) do error "Unsupported where clause, left hand side: #{other_clause}" end defp get_value({:^, _, [idx]}, params), do: Enum.at(params, idx) # Handle queries with interpolated values # ex. Repo.all from i in Item, where: i.id in ^item_ids defp get_value({:^, _, _}, params), do: params # Handle .all(query) QUERIES defp get_value(other_clause, _params), do: other_clause defp error(msg) do raise ArgumentError, message: msg end defp extract_select_fields(%Ecto.Query.SelectExpr{expr: expr} = _) do case expr do {_, _, [0]} -> [] {{:., _, [{_, _, _}, field]}, _, _} -> [field] {:{}, _, clauses} -> for {{_, _, [{_, _, _}, field]}, _, _} <- clauses, do: field end end # Decodes maps and datetime, seemingly unhandled by ExAws Dynamo decoder # (timestamps() corresponds with :naive_datetime) defp custom_decode(item, model, select) do selected_fields = extract_select_fields(select) case selected_fields do [] -> [Enum.reduce(model.__schema__(:fields), item, fn (field, acc) -> Map.update!(acc, field, fn val -> decode_type(model.__schema__(:type, field), val) end) end)] fields -> for field <- fields, do: decode_type(model.__schema__(:type, field), Map.get(item, field)) end end defp decode_item(item, model, sources, select) do item = Enum.reduce(item, %{}, fn {k, v}, acc -> key = to_string(Map.get(sources, String.to_atom(k))) Map.put(acc, key, v) end) %{"Item" => item} |> Dynamo.decode_item(as: model) |> custom_decode(model, select) end # This is used slightly differently # when handling select in custom_decode/2 defp decode_type(type, val) do if is_nil val do val else case type do :utc_datetime -> {:ok, dt, _offset} = DateTime.from_iso8601(val) dt :naive_datetime -> NaiveDateTime.from_iso8601!(val) {:embed, _} -> decode_embed(type, val) t when t in [Ecto.Adapters.DynamoDB.DynamoDBSet, MapSet] -> MapSet.new(val) _ -> val end end end defp decode_embed(type, val) do case Ecto.Adapters.SQL.load_embed(type, val) do {:ok, decoded_value} -> decoded_value :error -> ecto_dynamo_log(:info, "#{inspect __MODULE__}.decode_embed: failed to decode embedded value: #{inspect val}") nil end end # We found one instance where DynamoDB's error message could # be more instructive - when trying to set an indexed field to something # other than a string or number - so we're adding a more helpful message. # The parameter, 'params', has the type %{table: :string, records: [:map]} defp handle_error!(ex_aws_request_result, params) do case ex_aws_request_result do {:ok, result} -> result {:error, {error_name, _} = error} -> # Check for inappropriate insert into indexed field indexed_fields = Ecto.Adapters.DynamoDB.Info.indexed_attributes(params.table) # Repo.insert_all can present multiple records at once forbidden_insert_on_indexed_field = Enum.reduce(params.records, false, fn (record, acc) -> acc || Enum.any?(record, fn {field, val} -> [type] = ExAws.Dynamo.Encoder.encode(val) |> Map.keys # Ecto does not convert Empty strings to nil before passing them to Repo.update_all or # Repo.insert_all DynamoDB provides an instructive message during an update (forwarded by ExAws), # but less so for batch_write_item, so we catch the empty string as well. # Dynamo does not allow insertion of empty strings in any case. (Enum.member?(indexed_fields, to_string(field)) and not type in ["S", "N"]) || val == "" end) end) cond do # we use this error to check if an update or delete record does not exist error_name == "ConditionalCheckFailedException" -> {:error, error_name} forbidden_insert_on_indexed_field -> raise "The following request error could be related to attempting to insert an empty string or attempting to insert a type other than a string or number on an indexed field. Indexed fields: #{inspect indexed_fields}. Records: #{inspect params.records}.\n\nExAws Request Error! #{inspect error}" true -> raise ExAws.Error, message: "ExAws Request Error! #{inspect error}" end end end @doc """ Logs message to console and optionally to file. Log levels, colours and file path may be set in configuration (details in README.md). """ def ecto_dynamo_log(level, message, attributes \\ %{}, opts \\ []) do depth = opts[:depth] || 4 colours = Application.get_env(:ecto_adapters_dynamodb, :log_colours) d = DateTime.utc_now formatted_message = "#{d.year}-#{d.month}-#{d.day} #{d.hour}:#{d.minute}:#{d.second} UTC [Ecto dynamo #{level}] #{message}" {:ok, log_message} = Poison.encode(%{message: formatted_message, attributes: chisel(attributes, depth)}) log_path = Application.get_env(:ecto_adapters_dynamodb, :log_path) log_levels = Application.get_env(:ecto_adapters_dynamodb, :log_levels) || [:info] if level in log_levels do if Application.get_env(:ecto_adapters_dynamodb, :log_in_colour) do IO.ANSI.format([colours[level] || :normal, log_message], true) |> IO.puts else log_message |> IO.puts end if String.valid?(log_path) and Regex.match?(~r/\S/, log_path), do: log_pipe(log_path, log_message) end end defp chisel(str, _depth) when is_binary(str), do: str defp chisel(num, _depth) when is_number(num), do: num defp chisel(any, _depth) when (not is_map(any) and not is_list(any)), do: inspect any defp chisel(_, 0), do: "beyond_log_depth" defp chisel(%{__struct__: _} = struct, _depth), do: inspect struct defp chisel(map, depth) when is_map(map) do for {k, v} <- map, into: %{}, do: {k, chisel(v, depth - 1)} end defp chisel(list, depth) when is_list(list) do for e <- list, do: chisel(e, depth - 1) #Stream.with_index(list) |> Enum.reduce(%{}, fn({v,k}, acc)-> Map.put(acc, k, chisel(v, depth - 1)) end) end defp log_pipe(path, str) do {:ok, file} = File.open(path, [:append]) IO.binwrite(file, str) File.close(file) end end