Packages
erlcloud
0.13.0
3.8.3
3.8.2
3.8.1
3.7.6
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
3.6.8
3.6.7
3.6.5
3.6.4
3.6.3
3.6.2
3.6.1
3.6.0
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.5
3.4.3
3.4.1
3.4.0
3.3.9
3.3.8
3.3.7
3.3.6
3.3.5
3.3.4
3.3.3
3.3.2
3.3.1
3.3.0
3.2.18
3.2.17
3.2.16
3.2.15
3.2.14
3.2.13
3.2.12
3.2.11
3.2.10
3.2.7
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
3.1.17
3.1.16
3.1.14
3.1.13
3.1.12
3.1.11
3.1.9
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.5
3.0.4
3.0.3
3.0.2
3.0.1
2.2.16
2.2.15
2.2.14
2.2.13
2.2.12
2.2.11
2.2.10
2.2.9
2.2.8
2.2.7
2.2.6
2.2.5
2.2.4
2.2.2
2.2.1
2.2.0
2.1.0
2.0.5
2.0.4
2.0.3
2.0.0
0.13.10
0.13.9
0.13.8
0.13.6
0.13.5
0.13.4
0.13.3
0.13.2
0.13.0
0.12.0
0.11.0
0.9.2
0.9.2-rc.1
0.9.1
0.9.0
AWS APIs library for Erlang
Current section
Files
Jump to
Current section
Files
src/erlcloud_ddb_util.erl
%% -*- mode: erlang;erlang-indent-level: 4;indent-tabs-mode: nil -*-
%% @author Ransom Richardson <ransom@ransomr.net>
%% @doc
%% Helpers for using DynamoDB from Erlang.
%%
%% This is a higher layer API that augments the operations supported
%% by erlcloud_ddb2. The functions in this file do not map directly to
%% DynamoDB operations. Instead they will perform multiple operations
%% in order to implement functionality that isn't available directly
%% using the DynamoDB API.
%%
%% @end
-module(erlcloud_ddb_util).
-include("erlcloud.hrl").
-include("erlcloud_ddb2.hrl").
-include("erlcloud_aws.hrl").
%%% DynamoDB Higher Layer API
-export([delete_all/2, delete_all/3, delete_all/4,
delete_hash_key/3, delete_hash_key/4, delete_hash_key/5,
get_all/2, get_all/3, get_all/4,
put_all/2, put_all/3, put_all/4,
q_all/2, q_all/3, q_all/4,
scan_all/1, scan_all/2, scan_all/3,
write_all/2, write_all/3, write_all/4
]).
-define(BATCH_WRITE_LIMIT, 25).
-define(BATCH_GET_LIMIT, 100).
-type conditions() :: erlcloud_ddb2:conditions().
-type ddb_opts() :: erlcloud_ddb2:ddb_opts().
-type expression() :: erlcloud_ddb2:expression().
-type hash_key() :: erlcloud_ddb2:in_attr().
-type in_item() :: erlcloud_ddb2:in_item().
-type key() :: erlcloud_ddb2:key().
-type out_item() :: erlcloud_ddb2:out_item().
-type range_key_name() :: erlcloud_ddb2:range_key_name().
-type table_name() :: erlcloud_ddb2:table_name().
-type items_return() :: {ok, [out_item()]} | {error, term()}.
default_config() -> erlcloud_aws:default_config().
%%%------------------------------------------------------------------------------
%%% delete_all
%%%------------------------------------------------------------------------------
-spec delete_all(table_name(), [key()]) -> ok | {error, term()}.
delete_all(Table, Keys) ->
delete_all(Table, Keys, [], default_config()).
-spec delete_all(table_name(), [key()], ddb_opts()) -> ok | {error, term()}.
delete_all(Table, Keys, Opts) ->
delete_all(Table, Keys, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more BatchWriteItem operations to delete all items.
%% Operations are performed in parallel. Writing to only one table is supported.
%%
%% ===Example===
%%
%% `
%% ok =
%% erlcloud_ddb_util:delete_all(
%% [{<<"Forum">>,
%% [{<<"Name">>, {s, <<"Amazon DynamoDB">>}},
%% {<<"Name">>, {s, <<"Amazon RDS">>}},
%% {<<"Name">>, {s, <<"Amazon Redshift">>}},
%% {<<"Name">>, {s, <<"Amazon ElastiCache">>}}
%% ]}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec delete_all(table_name(), [key()], ddb_opts(), aws_config()) -> ok | {error, term()}.
delete_all(Table, Keys, Opts, Config) ->
write_all(Table, [{delete, Key} || Key <- Keys], Opts, Config).
%%%------------------------------------------------------------------------------
%%% delete_hash_key
%%%------------------------------------------------------------------------------
-spec delete_hash_key(table_name(), hash_key(), range_key_name()) -> ok | {error, term()}.
delete_hash_key(Table, HashKey, RangeKeyName) ->
delete_hash_key(Table, HashKey, RangeKeyName, [], default_config()).
-spec delete_hash_key(table_name(), hash_key(), range_key_name(), ddb_opts()) -> ok | {error, term()}.
delete_hash_key(Table, HashKey, RangeKeyName, Opts) ->
delete_hash_key(Table, HashKey, RangeKeyName, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Delete all items with the specified table. Table must be a
%% hash-and-range primary key table. Opts is currently ignored and is
%% provided for future enhancements. This method is not transacted.
%%
%% ===Example===
%%
%% `
%% ok = erlcloud_ddb_util:delete_hash_key(<<"tn">>, {<<"hash-key-name">>, <<"hash-key-value">>}, <<"range-key-name">>, [])),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec delete_hash_key(table_name(), hash_key(), range_key_name(), ddb_opts(), aws_config()) -> ok | {error, term()}.
delete_hash_key(Table, HashKey, RangeKeyName, Opts, Config) ->
case erlcloud_ddb2:q(Table, HashKey,
[{consistent_read, true},
{limit, ?BATCH_WRITE_LIMIT},
{attributes_to_get, [RangeKeyName]},
{out, typed_record}],
Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_q{count = 0}} ->
ok;
{ok, QResult} ->
case erlcloud_ddb2:batch_write_item(
[{Table, [{delete, [HashKey, RangeKey]} || [RangeKey] <- QResult#ddb2_q.items]}],
[{out, record}], Config) of
{error, Reason} ->
{error, Reason};
{ok, BatchResult} ->
if QResult#ddb2_q.last_evaluated_key == undefined andalso
BatchResult#ddb2_batch_write_item.unprocessed_items == [] ->
%% No more work to do
ok;
true ->
%% Some stuff was unprocessed - keep going
delete_hash_key(Table, HashKey, RangeKeyName, Opts, Config)
end
end
end.
%%%------------------------------------------------------------------------------
%%% get_all
%%%------------------------------------------------------------------------------
-type get_all_opts() :: erlcloud_ddb2:batch_get_item_request_item_opts().
-spec get_all(table_name(), [key()]) -> items_return().
get_all(Table, Keys) ->
get_all(Table, Keys, [], default_config()).
-spec get_all(table_name(), [key()], get_all_opts()) -> items_return().
get_all(Table, Keys, Opts) ->
get_all(Table, Keys, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more BatchGetItem operations to get all matching
%% items. Operations are performed in parallel. Order may not be preserved.
%% Getting from only one table is supported.
%%
%% ===Example===
%%
%% `
%% {ok, Items} =
%% erlcloud_ddb_util:get_all(
%% <<"Forum">>,
%% [{<<"Name">>, {s, <<"Amazon DynamoDB">>}},
%% {<<"Name">>, {s, <<"Amazon RDS">>}},
%% {<<"Name">>, {s, <<"Amazon Redshift">>}}],
%% [{projection_expression, <<"Name, Threads, Messages, Views">>}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec get_all(table_name(), [key()], get_all_opts(), aws_config()) -> items_return().
get_all(Table, Keys, Opts, Config) when length(Keys) =< ?BATCH_GET_LIMIT ->
batch_get_retry([{Table, Keys, Opts}], Config, []);
get_all(Table, Keys, Opts, Config) ->
BatchList = chop(?BATCH_GET_LIMIT, Keys),
Results = pmap_unordered(
fun(Batch) ->
%% try/catch to prevent hang forever if there is an exception
try
batch_get_retry([{Table, Batch, Opts}], Config, [])
catch
Type:Ex ->
{error, {Type, Ex}}
end
end,
BatchList),
lists:foldl(fun parfold/2, {ok, []}, Results).
-spec batch_get_retry([erlcloud_ddb2:batch_get_item_request_item()], aws_config(), [out_item()]) -> items_return().
batch_get_retry(RequestItems, Config, Acc) ->
case erlcloud_ddb2:batch_get_item(RequestItems, [{out, record}], Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_batch_get_item{unprocessed_keys = [],
responses = [#ddb2_batch_get_item_response{items = Items}]}} ->
{ok, Items ++ Acc};
{ok, #ddb2_batch_get_item{unprocessed_keys = Unprocessed,
responses = [#ddb2_batch_get_item_response{items = Items}]}} ->
batch_get_retry(Unprocessed, Config, Items ++ Acc)
end.
%%%------------------------------------------------------------------------------
%%% put_all
%%%------------------------------------------------------------------------------
-spec put_all(table_name(), [in_item()]) -> ok | {error, term()}.
put_all(Table, Items) ->
put_all(Table, Items, [], default_config()).
-spec put_all(table_name(), [in_item()], ddb_opts()) -> ok | {error, term()}.
put_all(Table, Items, Opts) ->
put_all(Table, Items, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more BatchWriteItem operations to put all items.
%% Operations are performed in parallel. Writing to only one table is supported.
%%
%% ===Example===
%%
%% `
%% ok =
%% erlcloud_ddb_util:put_all(
%% [{<<"Forum">>,
%% [[{<<"Name">>, {s, <<"Amazon DynamoDB">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}],
%% [{<<"Name">>, {s, <<"Amazon RDS">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}],
%% [{<<"Name">>, {s, <<"Amazon Redshift">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}],
%% [{<<"Name">>, {s, <<"Amazon ElastiCache">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]
%% ]}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec put_all(table_name(), [in_item()], ddb_opts(), aws_config()) -> ok | {error, term()}.
put_all(Table, Items, Opts, Config) ->
write_all(Table, [{put, Item} || Item <- Items], Opts, Config).
%%%------------------------------------------------------------------------------
%%% q_all
%%%------------------------------------------------------------------------------
-type q_all_opts() :: erlcloud_ddb2:q_opts().
-spec q_all(table_name(), conditions() | expression()) -> items_return().
q_all(Table, KeyConditionsOrExpression) ->
q_all(Table, KeyConditionsOrExpression, [], default_config()).
-spec q_all(table_name(), conditions() | expression(), q_all_opts()) -> items_return().
q_all(Table, KeyConditionsOrExpression, Opts) ->
q_all(Table, KeyConditionsOrExpression, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more Query operations to get all matching items.
%%
%% ===Example===
%%
%% `
%% {ok, Items} =
%% erlcloud_ddb_util:q_all(
%% <<"Thread">>,
%% <<"ForumName = :n AND LastPostDateTime BETWEEN :t1 AND :t2">>,
%% [{expression_attribute_values,
%% [{<<":n">>, <<"Amazon DynamoDB">>},
%% {<<":t1">>, <<"20130101">>},
%% {<<":t2">>, <<"20130115">>}]},
%% {index_name, <<"LastPostIndex">>},
%% {select, all_attributes},
%% {consistent_read, true}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec q_all(table_name(), conditions() | expression(), q_all_opts(), aws_config()) -> items_return().
q_all(Table, KeyConditionsOrExpression, Opts, Config) ->
q_all(Table, KeyConditionsOrExpression, Opts, Config, [], undefined).
-spec q_all(table_name(), conditions() | expression(), q_all_opts(), aws_config(), [[out_item()]], key() | undefined)
-> items_return().
q_all(Table, KeyConditionsOrExpression, Opts, Config, Acc, StartKey) ->
case erlcloud_ddb2:q(Table, KeyConditionsOrExpression,
[{exclusive_start_key, StartKey}, {out, record} | Opts],
Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_q{last_evaluated_key = undefined, items = Items}} ->
{ok, flatreverse([Items | Acc])};
{ok, #ddb2_q{last_evaluated_key = LastKey, items = Items}} ->
q_all(Table, KeyConditionsOrExpression, Opts, Config, [Items | Acc], LastKey)
end.
%%%------------------------------------------------------------------------------
%%% scan_all
%%%------------------------------------------------------------------------------
-type scan_all_opts() :: erlcloud_ddb2:scan_opts().
-spec scan_all(table_name()) -> items_return().
scan_all(Table) ->
scan_all(Table, [], default_config()).
-spec scan_all(table_name(), scan_all_opts()) -> items_return().
scan_all(Table, Opts) ->
scan_all(Table, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more Scan operations to get all matching items.
%%
%% ===Example===
%%
%% `
%% {ok, Items} =
%% erlcloud_ddb_util:scan_all(
%% <<"Thread">>,
%% [{segment, 0},
%% {total_segments, 4}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec scan_all(table_name(), scan_all_opts(), aws_config()) -> items_return().
scan_all(Table, Opts, Config) ->
scan_all(Table, Opts, Config, [], undefined).
-spec scan_all(table_name(), scan_all_opts(), aws_config(), [[out_item()]], key() | undefined)
-> items_return().
scan_all(Table, Opts, Config, Acc, StartKey) ->
case erlcloud_ddb2:scan(Table,
[{exclusive_start_key, StartKey}, {out, record} | Opts],
Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_scan{last_evaluated_key = undefined, items = Items}} ->
{ok, flatreverse([Items | Acc])};
{ok, #ddb2_scan{last_evaluated_key = LastKey, items = Items}} ->
scan_all(Table, Opts, Config, [Items | Acc], LastKey)
end.
%%%------------------------------------------------------------------------------
%%% write_all
%%%------------------------------------------------------------------------------
-type write_all_item() :: erlcloud_ddb2:batch_write_item_request().
-spec write_all(table_name(), [write_all_item()]) -> ok | {error, term()}.
write_all(Table, Items) ->
write_all(Table, Items, [], default_config()).
-spec write_all(table_name(), [write_all_item()], ddb_opts()) -> ok | {error, term()}.
write_all(Table, Items, Opts) ->
write_all(Table, Items, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more BatchWriteItem operations to put or delete all items.
%% Operations are performed in parallel. Writing to only one table is supported.
%%
%% ===Example===
%%
%% `
%% ok =
%% erlcloud_ddb_util:write_all(
%% [{<<"Forum">>,
%% [{put, [{<<"Name">>, {s, <<"Amazon DynamoDB">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]},
%% {put, [{<<"Name">>, {s, <<"Amazon RDS">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]},
%% {put, [{<<"Name">>, {s, <<"Amazon Redshift">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]},
%% {put, [{<<"Name">>, {s, <<"Amazon ElastiCache">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]}
%% ]}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec write_all(table_name(), [write_all_item()], ddb_opts(), aws_config()) -> ok | {error, term()}.
write_all(Table, Items, _Opts, Config) when length(Items) =< ?BATCH_WRITE_LIMIT ->
batch_write_retry([{Table, Items}], Config);
write_all(Table, Items, _Opts, Config) ->
BatchList = chop(?BATCH_WRITE_LIMIT, Items),
Results = pmap_unordered(
fun(Batch) ->
%% try/catch to prevent hang forever if there is an exception
try
batch_write_retry([{Table, Batch}], Config)
catch
Type:Ex ->
{error, {Type, Ex}}
end
end,
BatchList),
write_all_result(Results).
-spec batch_write_retry([erlcloud_ddb2:batch_write_item_request_item()], aws_config()) -> ok | {error, term()}.
batch_write_retry(RequestItems, Config) ->
case erlcloud_ddb2:batch_write_item(RequestItems, [{out, record}], Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_batch_write_item{unprocessed_items = []}} ->
ok;
{ok, #ddb2_batch_write_item{unprocessed_items = Unprocessed}} ->
batch_write_retry(Unprocessed, Config)
end.
write_all_result([ok | T]) ->
write_all_result(T);
write_all_result([{error, Reason} | _]) ->
{error, Reason};
write_all_result([]) ->
ok.
%%%------------------------------------------------------------------------------
%%% Internal Functions
%%%------------------------------------------------------------------------------
%% Reverses a list of lists and flattens one level
flatreverse(List) ->
lists:foldl(fun(I, A) -> I ++ A end, [], List).
%% fold a set of results from parallel operations producing lists
parfold(_, {error, Reason}) ->
{error, Reason};
parfold({error, Reason}, _) ->
{error, Reason};
parfold({ok, I}, {ok, A}) ->
{ok, I ++ A}.
%% parallel map implementation. See Armstrong's Programming Erlang and
%% http://bc.tech.coop/blog/070601.html
pmap_unordered(F, L) ->
Parent = self(),
Ref = make_ref(),
Pids = [spawn(fun() -> Parent ! {Ref, F(X)} end) || X <- L],
[receive {Ref, Result} -> Result end || _ <- Pids].
%% creates a list of list each of which has N or fewer elements
chop(N, List) ->
chop(N, List, []).
chop(_, [], Acc) ->
lists:reverse(Acc);
chop(N, List, Acc) ->
{H, T} = safe_split(N, List),
chop(N, T, [H | Acc]).
%% lists:split throws if N is larger than the list, safe_split doesn't
safe_split(N, List) ->
safe_split(N, List, []).
safe_split(0, List, Acc) ->
{lists:reverse(Acc), List};
safe_split(_, [], Acc) ->
{lists:reverse(Acc), []};
safe_split(N, [H|T], Acc) ->
safe_split(N - 1, T, [H | Acc]).