Series/2.x unify errors (#373)

build.sbt

@@ -182,14 +182,16 @@ lazy val zioDynamodb = module("zio-dynamodb", "dynamodb")
         val gets       = (1 to i).map(p => s"${lowerAlpha(p)} <- get[${upperAlpha(p)}](field$p)").mkString("\n      ")
         s"""def as[$tparams, $returnType](
            |    $params
-           |  )(fn: ($ftypes) => $returnType): Either[DynamoDBError, $returnType] =
+           |  )(fn: ($ftypes) => $returnType): Either[ItemError, $returnType] =
            |    for {
            |      $gets
            |    } yield fn($fparams)""".stripMargin
       }
       IO.write(
         file,
         s"""package zio.dynamodb
+           |
+           |import zio.dynamodb.DynamoDBError.ItemError
            |
            |private[dynamodb] trait GeneratedFromAttributeValueAs { this: AttrMap =>
            |
@@ -212,14 +214,16 @@ lazy val zioDynamodb = module("zio-dynamodb", "dynamodb")
         val gets       = (1 to i).map(p => s"${lowerAlpha(p)} <- get[${upperAlpha(p)}](field$p)").mkString("\n      ")
         s"""def as[$tparams, $returnType](
            |    $params
-           |  )(fn: ($ftypes) => $returnType): Either[DynamoDBError, $returnType] =
+           |  )(fn: ($ftypes) => $returnType): Either[ItemError, $returnType] =
            |    for {
            |      $gets
            |    } yield fn($fparams)""".stripMargin
       }
       IO.write(
         file,
         s"""package zio.dynamodb
+           |
+           |import zio.dynamodb.DynamoDBError.ItemError
            |
            |private[dynamodb] trait GeneratedFromAttributeValueAs { this: AttrMap =>
            |

docs/transactions.md

@@ -43,7 +43,7 @@ val enrollAvi = (putAvi zip putClasses).transaction
 
 ## Transaction Failures
 
-DynamoDBQueries using the `.transaction` method will fail at runtime if there are invalid transaction actions such as creating a table, scanning for items, or querying. The [DynamoDB documentation] has a limited number of actions that can be performed for either a read or a write transaction. There is a `.safeTransaction` method that is also available that will return `Either[Throwable, DynamoDBQuery[A]]`.
+DynamoDBQueries using the `.transaction` method will fail at runtime if there are invalid transaction actions such as creating a table, scanning for items, or querying. The [DynamoDB documentation] has a limited number of actions that can be performed for either a read or a write transaction. There is a `.safeTransaction` method that is also available that will return `Either[DynamoDBError.TransactionError, DynamoDBQuery[A]]`.
 
 There are more examples in our [integration tests](../dynamodb/src/it/scala/zio/dynamodb/LiveSpec.scala).
 

dynamodb/src/it/scala/zio/dynamodb/LiveSpec.scala

@@ -1,6 +1,6 @@
 package zio.dynamodb
 
-import software.amazon.awssdk.services.dynamodb.model.{ DynamoDbException, IdempotentParameterMismatchException }
+import software.amazon.awssdk.services.dynamodb.model.IdempotentParameterMismatchException
 import zio.dynamodb.UpdateExpression.Action.SetAction
 import zio.dynamodb.UpdateExpression.SetOperand
 import zio._
@@ -147,7 +147,13 @@ object LiveSpec extends DynamoDBLocalSpec {
     }
 
   private def assertDynamoDbException(substring: String): Assertion[Any] =
-    isSubtype[DynamoDbException](hasMessage(containsString(substring)))
+    isSubtype[DynamoDBError.AWSError](
+      hasField(
+        "cause",
+        _.cause,
+        hasMessage(containsString(substring))
+      )
+    )
 
   private val conditionAlwaysTrue = ConditionExpression.Equals(
     ConditionExpression.Operand.ValueOperand(AttributeValue(id)),
@@ -1400,7 +1406,15 @@ object LiveSpec extends DynamoDBLocalSpec {
               } yield ()
 
               assertZIO(program.exit)(
-                fails(isSubtype[IdempotentParameterMismatchException](Assertion.anything))
+                fails(
+                  isSubtype[DynamoDBError.AWSError](
+                    hasField(
+                      "cause",
+                      _.cause,
+                      isSubtype[IdempotentParameterMismatchException](anything)
+                    )
+                  )
+                )
               )
             }
           }

dynamodb/src/it/scala/zio/dynamodb/TypeSafeApiCrudSpec.scala

@@ -2,10 +2,14 @@ package zio.dynamodb
 
 import zio.schema.{ DeriveSchema, Schema }
 import zio.test._
+import zio.test.assertTrue
 import zio.test.Assertion._
+import zio.dynamodb.DynamoDBError.ItemError
 import zio.dynamodb.DynamoDBQuery.{ deleteFrom, forEach, get, put, scanAll, update }
-import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException
 import zio.Chunk
+import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException
+import zio.stream.ZStream
+import zio.ZIO
 
 object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
 
@@ -54,7 +58,24 @@ object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
             _    <- put(tableName, person).execute
             exit <- put(tableName, person).where(Person.id.notExists).execute.exit
           } yield exit
-          assertZIO(exit)(fails(isSubtype[ConditionalCheckFailedException](anything)))
+          assertZIO(exit)(fails(isConditionalCheckFailedException))
+        }
+      },
+      test("map error from condition expression that id not exists fails when item exists") {
+        withSingleIdKeyTable { tableName =>
+          val person = Person("1", "Smith", Some("John"), 21)
+          val exit   = for {
+            _    <- put(tableName, person).execute
+            exit <- put(tableName, person)
+                      .where(Person.id.notExists)
+                      .execute
+                      .mapError {
+                        case DynamoDBError.AWSError(_: ConditionalCheckFailedException) => 42
+                        case _                                                          => 0
+                      }
+                      .exit
+          } yield (exit)
+          assertZIO(exit)(fails(equalTo(42)))
         }
       },
       test("with condition expression that id exists when there is a item succeeds") {
@@ -80,9 +101,45 @@ object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
             p <- get(tableName)(Person.id.partitionKey === "1").execute.absolve
           } yield assertTrue(p == personUpdated)
         }
+      },
+      test("with forEach, catching a BatchError and resuming processing") {
+        withSingleIdKeyTable { tableName =>
+          type FailureWrapper = Either[String, Option[Person]]
+          val person1                                                                = Person("1", "Smith", Some("John"), 21)
+          val person2                                                                = Person("2", "Brown", None, 42)
+          val inputStream                                                            = ZStream(person1, person2)
+          val outputStream: ZStream[DynamoDBExecutor, DynamoDBError, FailureWrapper] = inputStream
+            .grouped(2)
+            .mapZIO { chunk =>
+              val batchWriteItem = DynamoDBQuery
+                .forEach(chunk)(a => put(tableName, a))
+                .map(Chunk.fromIterable)
+              for {
+                r <- ZIO.environment[DynamoDBExecutor]
+                b <- batchWriteItem.execute.provideEnvironment(r).map(_.map(Right(_))).catchSome {
+                       // example of catching a BatchError and resuming processing
+                       case DynamoDBError.BatchError.WriteError(map) => ZIO.succeed(Chunk(Left(map.toString)))
+                     }
+              } yield b
+            }
+            .flattenChunks
+
+          for {
+            xs <- outputStream.runCollect
+          } yield assertTrue(xs == Chunk(Right(None), Right(None)))
+        }
       }
     )
 
+  def isConditionalCheckFailedException: Assertion[Any] =
+    isSubtype[DynamoDBError.AWSError](
+      hasField(
+        "cause",
+        _.cause,
+        isSubtype[ConditionalCheckFailedException](anything)
+      )
+    )
+
   private val updateSuite = suite("update")(
     test("sets a single field with an update expression when item exists") {
       withSingleIdKeyTable { tableName =>
@@ -115,7 +172,7 @@ object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
             .where(Person.id.exists)
             .execute
             .exit
-        assertZIO(exit)(fails(isSubtype[ConditionalCheckFailedException](anything)))
+        assertZIO(exit)(fails(isConditionalCheckFailedException))
       }
     },
     test("set's a single field with an update plus a condition expression that addressSet contains an element") {
@@ -239,7 +296,7 @@ object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
           .where(Person.id === "1")
           .execute
           .exit
-        assertZIO(exit)(fails(isSubtype[ConditionalCheckFailedException](anything)))
+        assertZIO(exit)(fails(isConditionalCheckFailedException))
       }
     },
     test(
@@ -564,7 +621,7 @@ object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
           _ <- deleteFrom(tableName)(Person.id.partitionKey === "1").where(Person.id.exists).execute
           p <- get(tableName)(Person.id.partitionKey === "1").execute
         } yield assertTrue(
-          p == Left(DynamoDBError.ValueNotFound("value with key AttrMap(Map(id -> String(1))) not found"))
+          p == Left(ItemError.ValueNotFound("value with key AttrMap(Map(id -> String(1))) not found"))
         )
       }
     },
@@ -573,7 +630,7 @@ object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
     ) {
       withSingleIdKeyTable { tableName =>
         assertZIO(deleteFrom(tableName)(Person.id.partitionKey === "1").where(Person.id.exists).execute.exit)(
-          fails(isSubtype[ConditionalCheckFailedException](anything))
+          fails(isConditionalCheckFailedException)
         )
       }
     },
@@ -589,7 +646,7 @@ object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
                  .execute
           p <- get(tableName)(Person.id.partitionKey === "1").execute
         } yield assertTrue(
-          p == Left(DynamoDBError.ValueNotFound("value with key AttrMap(Map(id -> String(1))) not found"))
+          p == Left(ItemError.ValueNotFound("value with key AttrMap(Map(id -> String(1))) not found"))
         )
       }
     }
@@ -616,8 +673,8 @@ object TypeSafeApiCrudSpec extends DynamoDBLocalSpec {
           people <- forEach(Chunk(person1, person2))(p => get(tableName)(Person.id.partitionKey === p.id)).execute
         } yield assertTrue(
           people == List(
-            Left(DynamoDBError.ValueNotFound("value with key AttrMap(Map(id -> String(1))) not found")),
-            Left(DynamoDBError.ValueNotFound("value with key AttrMap(Map(id -> String(2))) not found"))
+            Left(ItemError.ValueNotFound("value with key AttrMap(Map(id -> String(1))) not found")),
+            Left(ItemError.ValueNotFound("value with key AttrMap(Map(id -> String(2))) not found"))
           )
         )
       }

dynamodb/src/main/scala/zio/dynamodb/AttrMap.scala

@@ -1,11 +1,12 @@
 package zio.dynamodb
 
-import zio.dynamodb.DynamoDBError.DecodingError
+import zio.dynamodb.DynamoDBError.ItemError.DecodingError
+import zio.dynamodb.DynamoDBError.ItemError
 import zio.prelude.ForEachOps
 
 final case class AttrMap(map: Map[String, AttributeValue]) extends GeneratedFromAttributeValueAs { self =>
 
-  def get[A](field: String)(implicit ev: FromAttributeValue[A]): Either[DynamoDBError, A] =
+  def get[A](field: String)(implicit ev: FromAttributeValue[A]): Either[ItemError, A] =
     map
       .get(field)
       .toRight(DecodingError(s"field '$field' not found"))
@@ -17,26 +18,28 @@ final case class AttrMap(map: Map[String, AttributeValue]) extends GeneratedFrom
       case _            => Right(None)
     }
 
-  def getItem[A](field: String)(f: AttrMap => Either[DynamoDBError, A]): Either[DynamoDBError, A] =
+  def getItem[A](field: String)(f: AttrMap => Either[ItemError, A]): Either[ItemError, A] =
     get[Item](field).flatMap(item => f(item))
 
   // convenience method so that user does not have to transform between an Option and an Either
   def getOptionalItem[A](
     field: String
-  )(f: AttrMap => Either[DynamoDBError, A]): Either[DynamoDBError, Option[A]] =
+  )(f: AttrMap => Either[ItemError, A]): Either[ItemError, Option[A]] =
     getOptional[Item](field).flatMap(
-      _.fold[Either[DynamoDBError, Option[A]]](Right(None))(item => f(item).map(Some(_)))
+      _.fold[Either[ItemError, Option[A]]](Right(None))(item => f(item).map(Some(_)))
     )
 
   // convenience method so that user does not have to transform between a List and an Either
-  def getIterableItem[A](field: String)(f: AttrMap => Either[DynamoDBError, A]): Either[DynamoDBError, Iterable[A]] =
-    get[Iterable[Item]](field).flatMap[DynamoDBError, Iterable[A]](xs => xs.forEach(f))
+  def getIterableItem[A](
+    field: String
+  )(f: AttrMap => Either[ItemError, A]): Either[ItemError, Iterable[A]] =
+    get[Iterable[Item]](field).flatMap[ItemError, Iterable[A]](xs => xs.forEach(f))
 
   // convenience method so that user does not have to transform between an Option, List and an Either
   def getOptionalIterableItem[A](
     field: String
-  )(f: AttrMap => Either[DynamoDBError, A]): Either[DynamoDBError, Option[Iterable[A]]] = {
-    def maybeTransform(maybeItems: Option[Iterable[Item]]): Either[DynamoDBError, Option[Iterable[A]]] =
+  )(f: AttrMap => Either[ItemError, A]): Either[ItemError, Option[Iterable[A]]] = {
+    def maybeTransform(maybeItems: Option[Iterable[Item]]): Either[ItemError, Option[Iterable[A]]] =
       maybeItems match {
         case None     => Right(None)
         case Some(xs) => xs.forEach(f).map(Some(_))

dynamodb/src/main/scala/zio/dynamodb/AttributeValue.scala

@@ -2,13 +2,14 @@ package zio.dynamodb
 
 import zio.dynamodb.ConditionExpression.Operand._
 import zio.dynamodb.ConditionExpression._
+import zio.dynamodb.DynamoDBError.ItemError
 import zio.schema.Schema
 import scala.collection.immutable.Set
 
 sealed trait AttributeValue { self =>
   type ScalaType
 
-  def decode[A](implicit schema: Schema[A]): Either[DynamoDBError, A] = Codec.decoder(schema)(self)
+  def decode[A](implicit schema: Schema[A]): Either[ItemError, A] = Codec.decoder(schema)(self)
 
   def ===[From](that: Operand.Size[From, ScalaType]): ConditionExpression[From]                  = Equals(ValueOperand(self), that)
   def <>[From](that: Operand.Size[From, ScalaType]): ConditionExpression[From]                   = NotEqual(ValueOperand(self), that)

dynamodb/src/main/scala/zio/dynamodb/Codec.scala

@@ -1,7 +1,8 @@
 package zio.dynamodb
 
 import zio.dynamodb.Annotations._
-import zio.dynamodb.DynamoDBError.DecodingError
+import zio.dynamodb.DynamoDBError.ItemError.DecodingError
+import zio.dynamodb.DynamoDBError.ItemError
 import zio.prelude.{ FlipOps, ForEachOps }
 import zio.schema.Schema.{ Optional, Primitive }
 import zio.schema.annotation.caseName
@@ -643,7 +644,7 @@ private[dynamodb] object Codec {
           (av: AttributeValue) => javaTimeStringParser(av)(ZoneOffset.of(_))
       }
 
-    private def javaTimeStringParser[A](av: AttributeValue)(unsafeParse: String => A): Either[DynamoDBError, A] =
+    private def javaTimeStringParser[A](av: AttributeValue)(unsafeParse: String => A): Either[ItemError, A] =
       FromAttributeValue.stringFromAttributeValue.fromAttributeValue(av).flatMap { s =>
         val stringOrA = Try(unsafeParse(s)).toEither.left
           .map(e => DecodingError(s"error parsing string '$s': ${e.getMessage}"))
@@ -781,7 +782,7 @@ private[dynamodb] object Codec {
       (av: AttributeValue) => {
         av match {
           case AttributeValue.Map(map) =>
-            val xs: Iterable[Either[DynamoDBError, (String, V)]] = map.map {
+            val xs: Iterable[Either[ItemError, (String, V)]] = map.map {
               case (k, v) =>
                 dec(v) match {
                   case Right(decV) => Right((k.value, decV))
@@ -824,7 +825,7 @@ private[dynamodb] object Codec {
           case AttributeValue.Map(map) =>
             // default enum encoding uses a Map with a single entry that denotes the type
             // TODO: think about being stricter and rejecting Maps with > 1 entry ???
-            map.toList.headOption.fold[Either[DynamoDBError, Z]](Left(DecodingError(s"map $av is empty"))) {
+            map.toList.headOption.fold[Either[ItemError, Z]](Left(DecodingError(s"map $av is empty"))) {
               case (AttributeValue.String(subtype), av) =>
                 cases.find { c =>
                   maybeCaseName(c.annotations).fold(c.id == subtype)(_ == subtype)
@@ -844,13 +845,13 @@ private[dynamodb] object Codec {
       discriminator: String,
       cases: Schema.Case[Z, _]*
     ): Decoder[Z] = { (av: AttributeValue) =>
-      def findCase(value: String): Either[DynamoDBError, Schema.Case[Z, _]] =
+      def findCase(value: String): Either[ItemError, Schema.Case[Z, _]] =
         cases.find {
           case Schema.Case(_, _, _, _, _, Chunk(caseName(const))) => const == value
           case Schema.Case(id, _, _, _, _, _)                     => id == value
         }.toRight(DecodingError(s"type name '$value' not found in schema cases"))
 
-      def decode(id: String): Either[DynamoDBError, Z] =
+      def decode(id: String): Either[ItemError, Z] =
         findCase(id).flatMap { c =>
           val dec = decoder(c.schema)
           dec(av).map(_.asInstanceOf[Z])
@@ -878,16 +879,16 @@ private[dynamodb] object Codec {
               .filter(_.isRight)
 
           rights.toList match {
-            case Nil      => Left(DynamoDBError.DecodingError(s"All sub type decoders failed for $av"))
+            case Nil      => Left(ItemError.DecodingError(s"All sub type decoders failed for $av"))
             case a :: Nil => a.map(_.asInstanceOf[Z])
             case _        =>
-              Left(DynamoDBError.DecodingError(s"More than one sub type decoder succeeded for $av"))
+              Left(ItemError.DecodingError(s"More than one sub type decoder succeeded for $av"))
           }
 
         case AttributeValue.Map(map)                        =>
           map
             .get(AttributeValue.String(discriminator))
-            .fold[Either[DynamoDBError, Z]](
+            .fold[Either[ItemError, Z]](
               Left(DecodingError(s"map $av does not contain discriminator field '$discriminator'"))
             ) {
               case AttributeValue.String(typeName) =>
@@ -903,16 +904,16 @@ private[dynamodb] object Codec {
     private[dynamodb] def decodeFields(
       av: AttributeValue,
       fields: Schema.Field[_, _]*
-    ): Either[DynamoDBError, List[Any]] =
+    ): Either[ItemError, List[Any]] =
       av match {
         case AttributeValue.Map(map) =>
           fields.toList.forEach {
             case Schema.Field(key, schema, _, _, _, _) =>
-              val dec                                = decoder(schema)
-              val k                                  = key // @fieldName is respected by the zio-schema macro
-              val maybeValue                         = map.get(AttributeValue.String(k))
-              val maybeDecoder                       = maybeValue.map(dec).toRight(DecodingError(s"field '$k' not found in $av"))
-              val either: Either[DynamoDBError, Any] = for {
+              val dec                            = decoder(schema)
+              val k                              = key // @fieldName is respected by the zio-schema macro
+              val maybeValue                     = map.get(AttributeValue.String(k))
+              val maybeDecoder                   = maybeValue.map(dec).toRight(DecodingError(s"field '$k' not found in $av"))
+              val either: Either[ItemError, Any] = for {
                 decoder <- maybeDecoder
                 decoded <- decoder
               } yield decoded