DynamoDB Connector

Setup

libraryDependencies += "dev.zio" %% "zio-connect-dynamodb" % "0.4.4"

How to use it?

All available DynamoDBConnector combinators and operations are available in the package object zio.connect.dynamodb._ you will need to import that to get started.

Additionally, you must also configure and provide the underlying DynamoDB layer provided by zio-aws you can read more about how to configure it here

If you have default credentials in the system environment typically at ~/.aws/credentials or as env variables the following configuration will likely work.

import zio._
import zio.aws.netty.NettyHttpClient
import zio.aws.core.config.AwsConfig
import zio.aws.core.httpclient.HttpClient
import zio.connect.dynamodb._

lazy val httpClient: ZLayer[Any, Throwable, HttpClient] = NettyHttpClient.default
lazy val awsConfig: ZLayer[Any, Throwable, AwsConfig]   = httpClient >>> AwsConfig.default

Almost everything in this api requires the existence of a table, we utilize the models provided by zio-aws to create tables, requests, responses, and all other DynamoDB related types. These are typically modeled as new-types from zio-prelude or case classes.

Here's a create table request:

import zio.aws.dynamodb.model._
import zio.aws.dynamodb.model.primitives._

def createTableRequest(tableName: TableName): CreateTableRequest =
  CreateTableRequest(
    tableName = tableName,
    attributeDefinitions = List(
      AttributeDefinition(
        KeySchemaAttributeName("id"),
        ScalarAttributeType.S
      )
    ),
    keySchema = List(
      KeySchemaElement(KeySchemaAttributeName("id"), KeyType.HASH)
    ),
    provisionedThroughput = Some(
      ProvisionedThroughput(
        readCapacityUnits = PositiveLongObject(16L),
        writeCapacityUnits = PositiveLongObject(16L)
      )
    ),
    tableClass = TableClass.STANDARD
  )

DynamoDB is "schemaless" in the sense that put data of different shapes in different rows, but you must define a schema for the keys that the table depends on, these are called "partition keys" and "sort keys" or "hash" and "range" keys. Only a partition/hash key is required, and you should only define attributes for the fields which constitute your keySchema. A valid table definition must also declare a provisionedThroughput, but there are many other options available for creating tables, you can read more about tables here.

Once you have a table definition you can create a table using the createTable combinator:

val tableName = TableName("my-table")
val createTableAction: ZIO[DynamoDBConnector, AwsError, CreateTableResponse] = createTable(createTableRequest(tableName)) >>> createTable

Dynamo tables can take a moment to be created, so you'll want to have some kind of retry mechanism when performing subsequent operations on the table. To illustrate, let's put an item into the table:

val putItemAction: ZIO[DynamoDBConnector, AwsError, PutItemResponse] = 
  ZStream(PutItemRequest(tableName, Map("id" -> AttributeValue(s = StringAttributeValue("my-id"))))) >>> putItem

val putItemWithRetry: ZIO[DynamoDBConnector, AwsError, PutItemResponse] =
    putItemAction.retryWhile {
      case GenericAwsError(_: ResourceNotFoundException) => true
      case _ => false
    }

Let's say we want to get an item from the table, we can use the getItem combinator, note we need to provide the full key here to use it:

val key = Map("id" -> AttributeValue(s = StringAttributeValue("my-id")))
val getItemAction: ZIO[DynamoDBConnector, AwsError, GetItemResponse] = 
  getItem(GetItemRequest(tableName, key)) >>> getItem

Currently, entries from the table are returned as a Map[AttributeName, AttributeValue] this may change in the future.

And we can also delete an item from the table:

val deleteItemAction: ZIO[DynamoDBConnector, AwsError, DeleteItemResponse] =
deleteItem(DeleteItemRequest(tableName, key)) >>> deleteItem

In order to run a program involving the DynamoDBConnector you need to provide a live DynamoDB from zio-aws along with config and the live connector layer:

override def run: ZIO[ZIOAppArgs with Scope, Any, Any] =
     program.provide(awsConfig, DynamoDb.live, dynamoDBConnectorLiveLayer)

dynamoDBConnectorLiveLayer is a ZLayer that provides the LiveDynamoDBConnector

Operators

The following operations are available:

batchGetItem

Accepts a stream of BatchGetItemRequest and returns a Chunk of BatchGetItemResponse, if one of the tables in the request is not available, the entire request will fail. The response object will contain and unprocessedKeys field which can be used to retry the request.

val item1              = Map(AttributeName("id") -> AttributeValue(s = StringAttributeValue("key1")))
val item2              = Map(AttributeName("id") -> AttributeValue(s = StringAttributeValue("key2")))
val keysAndAttributes  = KeysAndAttributes(List(item1))
val keysAndAttributes2 = KeysAndAttributes(List(item2))

val batchGetItemRequest =
  BatchGetItemRequest(Map(tableName -> keysAndAttributes, tableName2 -> keysAndAttributes2))
  
val batchGetItemAction: ZIO[DynamoDBConnector, AwsError, Chunk[BatchGetItemResponse]] =
  batchGetItem(ZStream(batchGetItemRequest)) >>> batchGetItem

batchWriteItem

Accepts a stream of BatchWriteItemRequest and returns a Chunk of BatchWriteItemResponse, if one of the tables in the request is not available, the entire request will fail. This is used to simultaneously put and delete items on multiple tables, you cannot write and delete an item with the same key in the same table in a single request. The response object does have an unprocessedKeys field which can be used to retry the remaining requests.

val tableName = TableName("batchWriteItem1")
val item1     = Map(AttributeName("id") -> AttributeValue(s = StringAttributeValue("key1")))
val item2     = Map(AttributeName("id") -> AttributeValue(s = StringAttributeValue("key2")))
val writeRequests = List(
  WriteRequest(putRequest = PutRequest(item1)),
  WriteRequest(putRequest = PutRequest(item2))
)
val batchWriteItemRequest = BatchWriteItemRequest(
  Map(tableName -> writeRequests)
)

val batchWriteItemAction: ZIO[DynamoDBConnector, AwsError, Chunk[BatchWriteItemResponse]] =
  batchWriteItem(ZStream(batchWriteItemRequest)) >>> batchWriteItem

createTable

Accepts a stream of CreateTableRequest and returns Unit. Will fail with a ResourceInUseException if the table already exists.

def createTableRequest(tableName: TableName): CreateTableRequest =
  CreateTableRequest(
    tableName = tableName,
    attributeDefinitions = List(
      AttributeDefinition(
        KeySchemaAttributeName("id"),
        ScalarAttributeType.S
      )
    ),
    keySchema = List(
      KeySchemaElement(KeySchemaAttributeName("id"), KeyType.HASH)
    ),
    provisionedThroughput = Some(
      ProvisionedThroughput(
        readCapacityUnits = PositiveLongObject(16L),
        writeCapacityUnits = PositiveLongObject(16L)
      )
    ),
    tableClass = TableClass.STANDARD
  )
    
val createTableAction: ZIO[DynamoDBConnector, AwsError, Unit] =
    ZStream(createTableRequest(tableName)) >>> createTable

deleteItem

Accepts a stream of DeleteItemRequest and returns Unit. Will fail with a ResourceNotFoundException if the table does not exist, does not fail if the item does not exist.

val item1 = Map(AttributeName("id") -> AttributeValue(s = StringAttributeValue("key1")))
val deleteItemAction: ZIO[DynamoDBConnector, AwsError, Unit] = ZStream(DeleteItemRequest(tableName, item1)) >>> deleteItem

deleteTable

Accepts a stream of DeleteTableRequest and returns Unit. Will fail with ResourceNotFoundException if the table does not exist.

val deleteTableAction: ZIO[DynamoDBConnector, AwsError, Unit] = ZStream(DeleteTableRequest(tableName)) >>> deleteTable

describeTable

Accepts a stream of DescribeTableRequest and returns a Chunk of DescribeTableResponse. Will fail if the table does not exist.

val describeTableAction: ZIO[DynamoDBConnector, AwsError, Chunk[DescribeTableResponse]] = ZStream(DescribeTableRequest(tableName)) >>> describeTable

listTables

Takes a ListTableRequest and return a Stream of TableName, can also provide a limit to the number of tables returned.

val listTablesAction: ZIO[DynamoDBConnector, AwsError, TableName] = listTables(ListTablesRequest()).runCollect

getItem

Accepts a stream of GetItemRequest and returns a Chunk of GetItemResponse. Will fail with a ResourceNotFoundException if the table does not exist.

val item1 = Map(AttributeName("id") -> AttributeValue(s = StringAttributeValue("key1")))
val getItemAction: ZIO[DynamoDBConnector, AwsError, Chunk[GetItemResponse]] = ZStream(GetItemRequest(tableName, item1)) >>> getItem

putItem

Accepts a stream of PutItemRequest and returns Unit. Will fail with a ResourceNotFoundException if the table does not exist.

val item1 = Map(AttributeName("id") -> AttributeValue(s = StringAttributeValue("key1")))
val putItemAction: ZIO[DynamoDBConnector, AwsError, Unit] = ZStream(PutItemRequest(tableName, item1)) >>> putItem

query

Accepts a stream of QueryRequest and returns a Chunk of Map[AttributeName, AttributeValue]. Will fail with a ResourceNotFoundException if the table does not exist.

val tableName     = TableName("query1")
val keyExpression = KeyExpression("id = :id")
val expressionAttributeValues = Map(
  ExpressionAttributeValueVariable(":id") -> AttributeValue(s = StringAttributeValue("key1"))
)
val queryRequest = QueryRequest(
  tableName,
  keyConditionExpression = keyExpression,
  expressionAttributeValues = expressionAttributeValues
)

val queryAction: ZIO[DynamoDBConnector, AwsError, Chunk[Map[AttributeName, AttributeValue]]] =
  ZStream(queryRequest) >>> query

scan

Similar to query, but if you don't know the key, you can use scan to return some items in a table or filter by some non-key condition. Accepts a stream of ScanRequest returns a Chunk of Map[AttributeName, AttributeValue]. Fails if the table does not exist.

val tableName     = TableName("scan1")
val scanRequest = ScanRequest(tableName)

val scanAction: ZIO[DynamoDBConnector, AwsError, Chunk[Map[AttributeName, AttributeValue]]] =
  ZStream(scanRequest) >>> scan

tableExists

Given a TableName, returns a Boolean indicating if the table exists.

val tableExistsAction: ZIO[DynamoDBConnector, AwsError, Boolean] = ZStream(tableName) >>> tableExists

updateItem

Accepts a stream of UpdateItemRequest and returns Unit. Will fail with a ResourceNotFoundException if the table does not exist.

val tableName = TableName("updateItem1")
val item1     = Map(AttributeName("id") -> AttributeValue(s = StringAttributeValue("key1")))
val updateItemRequest = UpdateItemRequest(
  tableName,
  item1,
  Map(
    AttributeName("authorized") -> AttributeValueUpdate(
      AttributeValue(bool = BooleanAttributeValue(true)),
      AttributeAction.PUT
    )
  )
)

val updateItemAction: ZIO[DynamoDBConnector, AwsError, Unit] =
  ZStream(updateItemRequest) >>> updateItem

updateTable

Accepts a stream of UpdateTableRequest and returns Unit. Will fail with a ResourceNotFoundException if the table does not exist.

val tableName = TableName("updateTable1")
val updateTableRequest = UpdateTableRequest(
  tableName,
  provisionedThroughput = Some(
    ProvisionedThroughput(
      readCapacityUnits = PositiveLongObject(16L),
      writeCapacityUnits = PositiveLongObject(16L)
    )
  )
)

val updateTableAction: ZIO[DynamoDBConnector, AwsError, Unit] =
  ZStream(updateTableRequest) >>> updateTable