View Source Dynamo

An Ecto-inspired Repo and schema DSL for Amazon DynamoDB.

Dynamo gives you typed schemas, composite key generation, a supervised Repo that owns credentials and retries, pure request builders you can unit-test, and a single normalised %Dynamo.Error{} for every failure.

defmodule MyApp.Order do
  use Dynamo.Schema

  item do
    table "app_data"

    field :user_id, :string, partition_key: true
    field :placed_at, :utc_datetime, sort_key: true
    field :order_id, :string, sort_key: true
    field :total, :decimal
    field :status, :string, default: "pending"
    field :version, :integer, default: 0

    global_secondary_index "StatusIndex", partition_key: :status, sort_key: :placed_at
  end
end

defmodule MyApp.Dynamo do
  use Dynamo.Repo, otp_app: :my_app
end

{:ok, order} = MyApp.Dynamo.put(%MyApp.Order{user_id: "u1", placed_at: DateTime.utc_now(), order_id: "o1", total: Decimal.new("9.99")})
{:ok, page}  = MyApp.Dynamo.query(%MyApp.Order{user_id: "u1"}, sort_key: {:gte, "2024-01-01"}, limit: 20)
{:ok, nil}   = MyApp.Dynamo.update(order, status: "shipped", version: {:increment, 1})

Installation

def deps do
  [
    {:dynamo, github: "bmalum/dynamo"},
    {:decimal, "~> 2.0"}   # optional, for the :decimal type
  ]
end

Requires Elixir 1.16+. decimal is optional: if you add it after dynamo was first compiled, run mix deps.compile dynamo --force so the :decimal type picks it up.

Upgrading from 0.1? The API changed completely. See CHANGELOG.md.

Setup

# lib/my_app/dynamo.ex
defmodule MyApp.Dynamo do
  use Dynamo.Repo, otp_app: :my_app
end

# config/runtime.exs
config :my_app, MyApp.Dynamo,
  region: System.get_env("AWS_REGION", "eu-central-1"),
  endpoint: System.get_env("DYNAMO_ENDPOINT")        # e.g. "http://localhost:8000" for DynamoDB Local

# lib/my_app/application.ex
children = [MyApp.Dynamo, ...]

Credentials resolve in the standard order: repo config → AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY → ECS container credentials → EC2 instance metadata (IMDSv2). Metadata credentials are cached in a supervised process and refreshed before expiry.

Other repo options: :max_retries (3), :base_backoff_ms, :max_backoff_ms, :receive_timeout, :connect_timeout, :req_options, :telemetry_prefix, :batch_retries (5).

Schemas

defmodule MyApp.User do
  use Dynamo.Schema,
    key_separator: "#",         # default
    partition_key_name: "pk",   # default
    sort_key_name: "sk"         # default

  item do
    table "app_data"
    entity_name "user"          # default: underscored module basename; or a parent schema module

    field :id, :string, partition_key: true
    field :email, :string, sort_key: true
    field :name, :string
    field :age, :integer, default: 0
    field :active, :boolean, default: true
    field :created_at, :utc_datetime
    field :balance, :decimal
    field :tags, :string_set
    field :profile, :map
    field :avatar, :binary

    global_secondary_index "EmailIndex", partition_key: :email, projection: :keys_only
  end
end

Field types

Field options: default:, partition_key:, sort_key:, required:.

Nil fields are omitted from written items (omit_nil: true, the default), so put never stores NULL attributes and load fills them from struct defaults. To remove an attribute from an existing item use update(item, field: :remove); update(item, field: nil) writes an explicit NULL.

Embedded schemas

defmodule MyApp.Address do
  use Dynamo.Schema, embedded: true      # no table, keys or indexes

  item do
    field :street, :string
    field :city, :string
    field :geo, {:list, :float}
  end
end

field :address, MyApp.Address
field :previous_addresses, {:list, MyApp.Address}

Embedded values keep their types on the way back (dates stay dates, decimals stay decimals). Any table schema can be embedded too; its key attributes are left out when nested.

Keys

Key fields are combined in declaration order with key_separator:

pk = "<entity_name>#<pk fields…>"      e.g. "user#u1"
sk = "<sk fields…>"                    e.g. "2024-01-15T10:00:00Z#o1"

Use partition_key [:a, :b] / sort_key [:c, :d] inside item instead of per-field flags if you prefer. Rules that keep keys round-trippable:

• Every partition key field must be non-nil → otherwise {:error, %Dynamo.Error{type: :validation_error}}.
• Sort key fields may be partially populated: generation stops at the first nil, producing a prefix for begins_with queries.
• A key value containing the separator is rejected (this is why the default is "#", not "-" – ISO dates contain dashes).

Existing tables that don't follow this layout: use Dynamo.Schema, raw_keys: true uses the partition/sort key fields as-is (native types, no entity prefix).

Working with schemas directly

{:ok, %Dynamo.Key{pk: %{"S" => "user#u1"}, sk: %{"S" => "[email protected]"}}} = Dynamo.Schema.primary_key(user)
{:ok, item}   = Dynamo.Schema.dump(user)          # DynamoDB item map incl. pk/sk
{:ok, user}   = Dynamo.Schema.load(MyApp.User, item)

Operations

All return {:ok, result} | {:error, %Dynamo.Error{}}; every function has a ! variant that raises.

alias MyApp.{Dynamo, User}

# write
{:ok, user} = Dynamo.put(user)
{:ok, user} = Dynamo.put(user, condition: "attribute_not_exists(pk)")
{:ok, old}  = Dynamo.put(user, return_values: :all_old)

# read
{:ok, %User{} | nil} = Dynamo.get(%User{id: "u1", email: "[email protected]"}, consistent: true)

# update – SET / REMOVE / ADD / DELETE built for you
{:ok, nil} = Dynamo.update(%User{id: "u1", email: "[email protected]"},
  name: "Jane",
  age: {:increment, 1},
  tags: {:add, MapSet.new(["vip"])},
  legacy_field: :remove,
  login_count: {:if_not_exists, 0}
)
{:ok, %User{}} = Dynamo.update(key, [age: {:add, 1}], return_values: :all_new)

# delete
{:ok, nil}     = Dynamo.delete(%User{id: "u1", email: "[email protected]"})
{:ok, %User{}} = Dynamo.delete(key, return_values: :all_old)

Expression options (condition:, filter:, projection:, names:, values:) are the same everywhere. condition: and filter: take either a raw string (with your own names:/values:, which are plain Elixir values encoded for you) or a keyword DSL that generates the placeholders:

Dynamo.put(user, condition: "#v = :v", names: %{"#v" => "version"}, values: %{v: 3})
Dynamo.put(user, condition: [version: 3])                                     # same thing
Dynamo.put(user, condition: [pk: :not_exists])
Dynamo.update(key, [age: {:increment, 1}], condition: [age: {:lt, 120}, active: true])
Dynamo.query(order, filter: {:or, [status: "open"], [total: {:gte, 100}]})
Dynamo.scan(User, filter: [tags: {:contains, "vip"}, {[:address, :city], "Berlin"}])   # nested path

Operators: value (=), {:eq | :ne | :lt | :lte | :gt | :gte, v}, {:between, lo, hi}, {:in, [..]}, {:begins_with, s}, {:contains, v}, :exists, :not_exists, {:attribute_type, "S"}, {:size, {:gt, n}}; combinators {:and, a, b}, {:or, a, b}, {:not, a}. Values are dumped with the field's schema type.

Querying

# all items in the partition
{:ok, %Dynamo.Page{items: items, last_evaluated_key: key}} = Dynamo.query(%Order{user_id: "u1"})

# sort key conditions
Dynamo.query(%Order{user_id: "u1"}, sort_key: {:begins_with, "2024-01"})
Dynamo.query(%Order{user_id: "u1"}, sort_key: {:between, "2024-01-01", "2024-02-01"})
Dynamo.query(%Order{user_id: "u1"}, sort_key: {:gte, "2024-06"}, descending: true, limit: 10)

# populated sort-key fields on the struct imply begins_with on the partial key
Dynamo.query(%Order{user_id: "u1", placed_at: ~U[2024-01-15 00:00:00Z]})

# pagination
{:ok, page1} = Dynamo.query(%Order{user_id: "u1"}, limit: 50)
{:ok, page2} = Dynamo.query(%Order{user_id: "u1"}, limit: 50, start_key: page1.last_evaluated_key)

# global secondary index (keys use the field's own type)
Dynamo.query(%Order{status: "shipped"}, index: "StatusIndex", sort_key: {:gt, ~U[2024-01-01 00:00:00Z]})

# filter / projection / count
Dynamo.query(%Order{user_id: "u1"}, filter: "#t > :min", names: %{"#t" => "total"}, values: %{min: 100})
Dynamo.query(%Order{user_id: "u1"}, projection: [:order_id, :total])
Dynamo.query(%Order{user_id: "u1"}, select: :count)

# everything, paginating for you (bounded by :limit)
{:ok, orders} = Dynamo.query_all(%Order{user_id: "u1"}, limit: 1_000)

%Dynamo.Page{} is enumerable: Enum.map(page, & &1.total).

Streaming and scans

# lazy query
Dynamo.stream(%Order{user_id: "u1"}, page_size: 100)
|> Stream.filter(&(&1.status == "open"))
|> Enum.take(10)

# lazy scan
Dynamo.stream(User) |> Enum.each(&migrate/1)

# truly parallel scan: segments fetched concurrently under the repo's Task.Supervisor
Dynamo.stream(User, segments: 8, max_concurrency: 4) |> Enum.count()

# one page
{:ok, %Dynamo.Page{}} = Dynamo.scan(User, limit: 100, filter: "attribute_exists(email)")

Streams raise %Dynamo.Error{} on failure (there is no other way to signal from inside a stream). Scans read every item – stream lazily, but don't scan casually.

Batches

# any number of items, chunked into 25, unprocessed items retried with backoff
{:ok, %{unprocessed: []}} = Dynamo.batch_write(users ++ [{:delete, old_user}])

# any number of keys (one schema), chunked into 100
{:ok, %{items: users, unprocessed_keys: []}} = Dynamo.batch_get([%User{id: "u1", email: "a"}, %User{id: "u2", email: "b"}])

Vector search

DynamoDB has native vector indexes (August 2026). Declare one on a {:vector, n} field and search it with an embedding:

defmodule MyApp.Doc do
  use Dynamo.Schema

  item do
    table "docs"
    field :id, :string, partition_key: true
    field :tenant, :string
    field :lang, :string
    field :title, :string
    field :embedding, {:vector, 1024}

    vector_index "Semantic",
      field: :embedding,
      distance: :cosine,                          # :cosine | :dot_product | :euclidean
      projection: :include, projected_attributes: [:title, :lang],
      partition_key: :tenant,                     # optional: scopes and scales searches
      filters: [:lang]                            # optional inline filter attributes
  end
end

{:ok, %{results: results}} =
  Dynamo.search_vectors(MyApp.Doc, "Semantic", query_embedding,
    top_k: 5,
    filter: [tenant: "acme", lang: "en"]          # partition key is required when declared
  )

for {%MyApp.Doc{title: t}, score} <- results, do: IO.puts("#{score}  #{t}")

mix dynamo.create_table --schema MyApp.Doc creates the index (on-demand billing is required). Results are {struct, score} pairs, most similar first; lower is closer for :cosine/:euclidean, higher for :dot_product. Only projected attributes are returned; key fields are recovered from the item key. Writes to the index are eventually consistent. SearchVectors is served from a separate AWS endpoint, which the client routes automatically (:search_endpoint to override).

Transactions

{:ok, _} =
  Dynamo.transaction(
    [
      {:update, %Account{id: "src"}, [balance: {:decrement, 100}], condition: "#b >= :amt", names: %{"#b" => "balance"}, values: %{amt: 100}},
      {:update, %Account{id: "dst"}, [balance: {:increment, 100}]},
      {:put, %Transfer{id: "t1", amount: 100}, condition: "attribute_not_exists(pk)"},
      {:check, %Account{id: "audit"}, "attribute_exists(pk)"},
      {:delete, %Hold{id: "h1"}}
    ],
    client_request_token: "transfer-t1"   # idempotency (10 min window)
  )

On cancellation the error carries per-item reasons:

{:error, %Dynamo.Error{type: :transaction_canceled, cancellation_reasons: [%{code: "ConditionalCheckFailed"}, %{code: "None"}, ...]}}

Errors

Every failure is a %Dynamo.Error{}:

case Dynamo.put(user, condition: "attribute_not_exists(pk)") do
  {:ok, user} -> ...
  {:error, %Dynamo.Error{type: :conditional_check_failed}} -> {:error, :already_exists}
  {:error, %Dynamo.Error{retryable?: true}} -> retry_later()
  {:error, %Dynamo.Error{} = e} -> Logger.error(Exception.message(e))
end

Fields: type, message, aws_type (e.g. "ResourceNotFoundException"), status, retryable?, cancellation_reasons, details.

Common types: :validation_error (local or AWS), :resource_not_found, :conditional_check_failed, :provisioned_throughput_exceeded, :throttled, :access_denied, :authentication_error, :transaction_canceled, :transaction_conflict, :item_collection_size_exceeded, :internal_server_error, :transport_error, :decode_error. Retryable errors are retried by the client automatically (:max_retries, jittered exponential backoff).

Retries and idempotency. Throttling and 5xx responses are safe to retry – DynamoDB did not apply the write. A transport timeout is different: the request may or may not have been applied, and the client retries it anyway (as the AWS SDKs do). Plain put/delete are idempotent so this is harmless; update with {:increment, n}/{:add, n} is not. For counters that must be exact, guard the update with a version condition (update(item, [count: {:increment, 1}, version: {:increment, 1}], condition: [version: item.version])) or use transaction/2 with client_request_token:, which DynamoDB de-duplicates for 10 minutes. Set max_retries: 0 on the repo to opt out of retries entirely.

Telemetry

The client emits [:dynamo, :request, :start | :stop | :exception] (prefix configurable) with metadata action, table, attempt, result, status, error, consumed_capacity and duration on stop:

:telemetry.attach("dynamo-log", [:dynamo, :request, :stop], fn _, %{duration: d}, meta, _ ->
  Logger.debug("dynamo #{meta.action} #{meta.table} #{System.convert_time_unit(d, :native, :millisecond)}ms #{meta.result}")
end, nil)

Add return_consumed_capacity: :total to any call to get capacity in the metadata.

Mix tasks

mix dynamo.create_table --schema MyApp.User --endpoint http://localhost:8000   # table + GSIs + vector indexes from the schema
mix dynamo.create_table my_table --partition-key id --no-sort-key --region eu-west-1
mix dynamo.list_tables --name-contains user
mix dynamo.delete_table my_table --force
mix dynamo.generate_schema legacy_table --module MyApp.Legacy                   # round-trip safe: detects key layout

All tasks load your application config, accept --repo MyApp.Dynamo to reuse a started repo's client, and exit non-zero on failure.

Testing your application

Point the repo at a fake via Req's plug option – no network, no mocks of library modules:

config :my_app, MyApp.Dynamo,
  region: "us-east-1", access_key_id: "test", secret_access_key: "test",
  req_options: [plug: {Req.Test, MyApp.Dynamo}]

Or run DynamoDB Local and set endpoint: "http://localhost:8000". The library's own test suite uses an in-memory fake (test/support/fake_dynamo.ex) you can borrow.

Design

• Schema – typed fields, one Dynamo.Schema.primary_key/1 used by every operation.
• Request – pure builders (Dynamo.Request, Dynamo.Update) that return {action, payload}; fully unit-tested against golden JSON.
• Client – SigV4, retries, telemetry, error normalisation. Nothing above it sees a raw AWS error.
• Repo – supervised owner of client + credentials; the only stateful piece.
• Fail loud – nil keys, separator collisions, unknown options and wrong types are errors, never sentinels.

See Agents.md for data-modelling guidance and the guides/ folder for worked examples.

Contributing

See CONTRIBUTING.md. Security issues: SECURITY.md.

License

MIT – see LICENSE.