diff --git a/moto/dynamodb/exceptions.py b/moto/dynamodb/exceptions.py
--- a/moto/dynamodb/exceptions.py
+++ b/moto/dynamodb/exceptions.py
@@ -1,15 +1,24 @@
+import json
+from moto.core.exceptions import JsonRESTError
 from moto.dynamodb.limits import HASH_KEY_MAX_LENGTH, RANGE_KEY_MAX_LENGTH
 
 
-class InvalidIndexNameError(ValueError):
+class DynamodbException(JsonRESTError):
     pass
 
 
-class MockValidationException(ValueError):
+class MockValidationException(DynamodbException):
+    error_type = "com.amazonaws.dynamodb.v20111205#ValidationException"
+
     def __init__(self, message):
+        super().__init__(MockValidationException.error_type, message=message)
         self.exception_msg = message
 
 
+class InvalidIndexNameError(MockValidationException):
+    pass
+
+
 class InvalidUpdateExpressionInvalidDocumentPath(MockValidationException):
     invalid_update_expression_msg = (
         "The document path provided in the update expression is invalid for update"
@@ -183,19 +192,37 @@ def __init__(self):
         super().__init__(self.inc_data_type_msg)
 
 
-class ConditionalCheckFailed(ValueError):
-    msg = "The conditional request failed"
+class ConditionalCheckFailed(DynamodbException):
+    error_type = "com.amazonaws.dynamodb.v20111205#ConditionalCheckFailedException"
 
-    def __init__(self):
-        super().__init__(self.msg)
+    def __init__(self, msg=None):
+        super().__init__(
+            ConditionalCheckFailed.error_type, msg or "The conditional request failed"
+        )
 
 
-class TransactionCanceledException(ValueError):
+class TransactionCanceledException(DynamodbException):
     cancel_reason_msg = "Transaction cancelled, please refer cancellation reasons for specific reasons [{}]"
+    error_type = "com.amazonaws.dynamodb.v20120810#TransactionCanceledException"
 
     def __init__(self, errors):
-        msg = self.cancel_reason_msg.format(", ".join([str(err) for err in errors]))
-        super().__init__(msg)
+        msg = self.cancel_reason_msg.format(
+            ", ".join([str(code) for code, _ in errors])
+        )
+        super().__init__(
+            error_type=TransactionCanceledException.error_type, message=msg
+        )
+        reasons = [
+            {"Code": code, "Message": message} if code else {"Code": "None"}
+            for code, message in errors
+        ]
+        self.description = json.dumps(
+            {
+                "__type": TransactionCanceledException.error_type,
+                "CancellationReasons": reasons,
+                "Message": msg,
+            }
+        )
 
 
 class MultipleTransactionsException(MockValidationException):
@@ -240,3 +267,46 @@ class TooManyAddClauses(InvalidUpdateExpression):
 
     def __init__(self):
         super().__init__(self.msg)
+
+
+class ResourceNotFoundException(JsonRESTError):
+    def __init__(self, msg=None):
+        err = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
+        super().__init__(err, msg or "Requested resource not found")
+
+
+class TableNotFoundException(JsonRESTError):
+    def __init__(self, name):
+        err = "com.amazonaws.dynamodb.v20111205#TableNotFoundException"
+        msg = "Table not found: {}".format(name)
+        super().__init__(err, msg)
+
+
+class SourceTableNotFoundException(JsonRESTError):
+    def __init__(self, source_table_name):
+        er = "com.amazonaws.dynamodb.v20111205#SourceTableNotFoundException"
+        super().__init__(er, "Source table not found: %s" % source_table_name)
+
+
+class BackupNotFoundException(JsonRESTError):
+    def __init__(self, backup_arn):
+        er = "com.amazonaws.dynamodb.v20111205#BackupNotFoundException"
+        super().__init__(er, "Backup not found: %s" % backup_arn)
+
+
+class TableAlreadyExistsException(JsonRESTError):
+    def __init__(self, target_table_name):
+        er = "com.amazonaws.dynamodb.v20111205#TableAlreadyExistsException"
+        super().__init__(er, "Table already exists: %s" % target_table_name)
+
+
+class ResourceInUseException(JsonRESTError):
+    def __init__(self):
+        er = "com.amazonaws.dynamodb.v20111205#ResourceInUseException"
+        super().__init__(er, "Resource in use")
+
+
+class StreamAlreadyEnabledException(JsonRESTError):
+    def __init__(self):
+        er = "com.amazonaws.dynamodb.v20111205#ResourceInUseException"
+        super().__init__(er, "Cannot enable stream")
diff --git a/moto/dynamodb/models/__init__.py b/moto/dynamodb/models/__init__.py
--- a/moto/dynamodb/models/__init__.py
+++ b/moto/dynamodb/models/__init__.py
@@ -25,6 +25,14 @@
     InvalidAttributeTypeError,
     MultipleTransactionsException,
     TooManyTransactionsException,
+    TableNotFoundException,
+    ResourceNotFoundException,
+    SourceTableNotFoundException,
+    TableAlreadyExistsException,
+    BackupNotFoundException,
+    ResourceInUseException,
+    StreamAlreadyEnabledException,
+    MockValidationException,
 )
 from moto.dynamodb.models.utilities import bytesize
 from moto.dynamodb.models.dynamo_type import DynamoType
@@ -648,7 +656,7 @@ def put_item(
         overwrite=False,
     ):
         if self.hash_key_attr not in item_attrs.keys():
-            raise KeyError(
+            raise MockValidationException(
                 "One or more parameter values were invalid: Missing the key "
                 + self.hash_key_attr
                 + " in the item"
@@ -656,7 +664,7 @@ def put_item(
         hash_value = DynamoType(item_attrs.get(self.hash_key_attr))
         if self.has_range_key:
             if self.range_key_attr not in item_attrs.keys():
-                raise KeyError(
+                raise MockValidationException(
                     "One or more parameter values were invalid: Missing the key "
                     + self.range_key_attr
                     + " in the item"
@@ -725,7 +733,7 @@ def has_range_key(self):
 
     def get_item(self, hash_key, range_key=None, projection_expression=None):
         if self.has_range_key and not range_key:
-            raise ValueError(
+            raise MockValidationException(
                 "Table has a range key, but no range key was passed into get_item"
             )
         try:
@@ -780,7 +788,7 @@ def query(
             all_indexes = self.all_indexes()
             indexes_by_name = dict((i.name, i) for i in all_indexes)
             if index_name not in indexes_by_name:
-                raise ValueError(
+                raise MockValidationException(
                     "Invalid index: %s for table: %s. Available indexes are: %s"
                     % (index_name, self.name, ", ".join(indexes_by_name.keys()))
                 )
@@ -791,7 +799,9 @@ def query(
                     key for key in index.schema if key["KeyType"] == "HASH"
                 ][0]
             except IndexError:
-                raise ValueError("Missing Hash Key. KeySchema: %s" % index.name)
+                raise MockValidationException(
+                    "Missing Hash Key. KeySchema: %s" % index.name
+                )
 
             try:
                 index_range_key = [
@@ -904,11 +914,13 @@ def all_items(self):
     def all_indexes(self):
         return (self.global_indexes or []) + (self.indexes or [])
 
-    def get_index(self, index_name, err=None):
+    def get_index(self, index_name, error_if_not=False):
         all_indexes = self.all_indexes()
         indexes_by_name = dict((i.name, i) for i in all_indexes)
-        if err and index_name not in indexes_by_name:
-            raise err
+        if error_if_not and index_name not in indexes_by_name:
+            raise InvalidIndexNameError(
+                "The table does not have the specified index: %s" % index_name
+            )
         return indexes_by_name[index_name]
 
     def has_idx_items(self, index_name):
@@ -938,10 +950,7 @@ def scan(
         scanned_count = 0
 
         if index_name:
-            err = InvalidIndexNameError(
-                "The table does not have the specified index: %s" % index_name
-            )
-            self.get_index(index_name, err)
+            self.get_index(index_name, error_if_not=True)
             items = self.has_idx_items(index_name)
         else:
             items = self.all_items()
@@ -1182,12 +1191,14 @@ def default_vpc_endpoint_service(service_region, zones):
 
     def create_table(self, name, **params):
         if name in self.tables:
-            return None
+            raise ResourceInUseException
         table = Table(name, region=self.region_name, **params)
         self.tables[name] = table
         return table
 
     def delete_table(self, name):
+        if name not in self.tables:
+            raise ResourceNotFoundException
         return self.tables.pop(name, None)
 
     def describe_endpoints(self):
@@ -1211,11 +1222,10 @@ def untag_resource(self, table_arn, tag_keys):
                 ]
 
     def list_tags_of_resource(self, table_arn):
-        required_table = None
         for table in self.tables:
             if self.tables[table].table_arn == table_arn:
-                required_table = self.tables[table]
-        return required_table.tags
+                return self.tables[table].tags
+        raise ResourceNotFoundException
 
     def list_tables(self, limit, exclusive_start_table_name):
         all_tables = list(self.tables.keys())
@@ -1240,7 +1250,7 @@ def list_tables(self, limit, exclusive_start_table_name):
         return tables, None
 
     def describe_table(self, name):
-        table = self.tables[name]
+        table = self.get_table(name)
         return table.describe(base_key="Table")
 
     def update_table(
@@ -1281,7 +1291,7 @@ def update_table_streams(self, name, stream_specification):
             stream_specification.get("StreamEnabled")
             or stream_specification.get("StreamViewType")
         ) and table.latest_stream_label:
-            raise ValueError("Table already has stream enabled")
+            raise StreamAlreadyEnabledException
         table.set_stream_specification(stream_specification)
         return table
 
@@ -1338,9 +1348,7 @@ def put_item(
         expression_attribute_values=None,
         overwrite=False,
     ):
-        table = self.tables.get(table_name)
-        if not table:
-            return None
+        table = self.get_table(table_name)
         return table.put_item(
             item_attrs,
             expected,
@@ -1377,22 +1385,37 @@ def get_keys_value(self, table, keys):
         if table.hash_key_attr not in keys or (
             table.has_range_key and table.range_key_attr not in keys
         ):
-            raise ValueError(
-                "Table has a range key, but no range key was passed into get_item"
-            )
+            # "Table has a range key, but no range key was passed into get_item"
+            raise MockValidationException("Validation Exception")
         hash_key = DynamoType(keys[table.hash_key_attr])
         range_key = (
             DynamoType(keys[table.range_key_attr]) if table.has_range_key else None
         )
         return hash_key, range_key
 
+    def get_schema(self, table_name, index_name):
+        table = self.get_table(table_name)
+        if index_name:
+            all_indexes = (table.global_indexes or []) + (table.indexes or [])
+            indexes_by_name = dict((i.name, i) for i in all_indexes)
+            if index_name not in indexes_by_name:
+                raise ResourceNotFoundException(
+                    "Invalid index: {} for table: {}. Available indexes are: {}".format(
+                        index_name, table_name, ", ".join(indexes_by_name.keys())
+                    )
+                )
+
+            return indexes_by_name[index_name].schema
+        else:
+            return table.schema
+
     def get_table(self, table_name):
+        if table_name not in self.tables:
+            raise ResourceNotFoundException()
         return self.tables.get(table_name)
 
     def get_item(self, table_name, keys, projection_expression=None):
         table = self.get_table(table_name)
-        if not table:
-            raise ValueError("No table found")
         hash_key, range_key = self.get_keys_value(table, keys)
         return table.get_item(hash_key, range_key, projection_expression)
 
@@ -1412,9 +1435,7 @@ def query(
         filter_expression=None,
         **filter_kwargs,
     ):
-        table = self.tables.get(table_name)
-        if not table:
-            return None, None
+        table = self.get_table(table_name)
 
         hash_key = DynamoType(hash_key_dict)
         range_values = [DynamoType(range_value) for range_value in range_value_dicts]
@@ -1448,9 +1469,7 @@ def scan(
         index_name,
         projection_expression,
     ):
-        table = self.tables.get(table_name)
-        if not table:
-            return None, None, None
+        table = self.get_table(table_name)
 
         scan_filters = {}
         for key, (comparison_operator, comparison_values) in filters.items():
@@ -1582,8 +1601,6 @@ def delete_item(
         condition_expression=None,
     ):
         table = self.get_table(table_name)
-        if not table:
-            return None
 
         hash_value, range_value = self.get_keys_value(table, key)
         item = table.get_item(hash_value, range_value)
@@ -1719,25 +1736,31 @@ def check_unicity(table_name, key):
                     )
                 else:
                     raise ValueError
-                errors.append(None)
+                errors.append((None, None))
             except MultipleTransactionsException:
                 # Rollback to the original state, and reraise the error
                 self.tables = original_table_state
                 raise MultipleTransactionsException()
             except Exception as e:  # noqa: E722 Do not use bare except
-                errors.append(type(e).__name__)
-        if any(errors):
+                errors.append((type(e).__name__, e.message))
+        if set(errors) != set([(None, None)]):
             # Rollback to the original state, and reraise the errors
             self.tables = original_table_state
             raise TransactionCanceledException(errors)
 
     def describe_continuous_backups(self, table_name):
-        table = self.get_table(table_name)
+        try:
+            table = self.get_table(table_name)
+        except ResourceNotFoundException:
+            raise TableNotFoundException(table_name)
 
         return table.continuous_backups
 
     def update_continuous_backups(self, table_name, point_in_time_spec):
-        table = self.get_table(table_name)
+        try:
+            table = self.get_table(table_name)
+        except ResourceNotFoundException:
+            raise TableNotFoundException(table_name)
 
         if (
             point_in_time_spec["PointInTimeRecoveryEnabled"]
@@ -1759,6 +1782,8 @@ def update_continuous_backups(self, table_name, point_in_time_spec):
         return table.continuous_backups
 
     def get_backup(self, backup_arn):
+        if backup_arn not in self.backups:
+            raise BackupNotFoundException(backup_arn)
         return self.backups.get(backup_arn)
 
     def list_backups(self, table_name):
@@ -1768,9 +1793,10 @@ def list_backups(self, table_name):
         return backups
 
     def create_backup(self, table_name, backup_name):
-        table = self.get_table(table_name)
-        if table is None:
-            raise KeyError()
+        try:
+            table = self.get_table(table_name)
+        except ResourceNotFoundException:
+            raise TableNotFoundException(table_name)
         backup = Backup(self, backup_name, table)
         self.backups[backup.arn] = backup
         return backup
@@ -1791,11 +1817,8 @@ def describe_backup(self, backup_arn):
 
     def restore_table_from_backup(self, target_table_name, backup_arn):
         backup = self.get_backup(backup_arn)
-        if backup is None:
-            raise KeyError()
-        existing_table = self.get_table(target_table_name)
-        if existing_table is not None:
-            raise ValueError()
+        if target_table_name in self.tables:
+            raise TableAlreadyExistsException(target_table_name)
         new_table = RestoredTable(
             target_table_name, region=self.region_name, backup=backup
         )
@@ -1808,12 +1831,12 @@ def restore_table_to_point_in_time(self, target_table_name, source_table_name):
         copy all items from the source without respect to other arguments.
         """
 
-        source = self.get_table(source_table_name)
-        if source is None:
-            raise KeyError()
-        existing_table = self.get_table(target_table_name)
-        if existing_table is not None:
-            raise ValueError()
+        try:
+            source = self.get_table(source_table_name)
+        except ResourceNotFoundException:
+            raise SourceTableNotFoundException(source_table_name)
+        if target_table_name in self.tables:
+            raise TableAlreadyExistsException(target_table_name)
         new_table = RestoredPITTable(
             target_table_name, region=self.region_name, source=source
         )
diff --git a/moto/dynamodb/responses.py b/moto/dynamodb/responses.py
--- a/moto/dynamodb/responses.py
+++ b/moto/dynamodb/responses.py
@@ -8,9 +8,9 @@
 from moto.core.responses import BaseResponse
 from moto.core.utils import camelcase_to_underscores, amz_crc32, amzn_request_id
 from .exceptions import (
-    InvalidIndexNameError,
     MockValidationException,
-    TransactionCanceledException,
+    ResourceNotFoundException,
+    ConditionalCheckFailed,
 )
 from moto.dynamodb.models import dynamodb_backends, dynamo_json_dump
 
@@ -111,13 +111,6 @@ def get_endpoint_name(self, headers):
         if match:
             return match.split(".")[1]
 
-    def error(self, type_, message, status=400):
-        return (
-            status,
-            self.response_headers,
-            dynamo_json_dump({"__type": type_, "message": message}),
-        )
-
     @property
     def dynamodb_backend(self):
         """
@@ -165,19 +158,16 @@ def create_table(self):
         # check billing mode and get the throughput
         if "BillingMode" in body.keys() and body["BillingMode"] == "PAY_PER_REQUEST":
             if "ProvisionedThroughput" in body.keys():
-                er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-                return self.error(
-                    er,
-                    "ProvisionedThroughput cannot be specified when BillingMode is PAY_PER_REQUEST",
+                raise MockValidationException(
+                    "ProvisionedThroughput cannot be specified when BillingMode is PAY_PER_REQUEST"
                 )
             throughput = None
             billing_mode = "PAY_PER_REQUEST"
         else:  # Provisioned (default billing mode)
             throughput = body.get("ProvisionedThroughput")
             if throughput is None:
-                return self.error(
-                    "ValidationException",
-                    "One or more parameter values were invalid: ReadCapacityUnits and WriteCapacityUnits must both be specified when BillingMode is PROVISIONED",
+                raise MockValidationException(
+                    "One or more parameter values were invalid: ReadCapacityUnits and WriteCapacityUnits must both be specified when BillingMode is PROVISIONED"
                 )
             billing_mode = "PROVISIONED"
         # getting ServerSideEncryption details
@@ -189,16 +179,14 @@ def create_table(self):
         # getting the indexes
         global_indexes = body.get("GlobalSecondaryIndexes")
         if global_indexes == []:
-            return self.error(
-                "ValidationException",
-                "One or more parameter values were invalid: List of GlobalSecondaryIndexes is empty",
+            raise MockValidationException(
+                "One or more parameter values were invalid: List of GlobalSecondaryIndexes is empty"
             )
         global_indexes = global_indexes or []
         local_secondary_indexes = body.get("LocalSecondaryIndexes")
         if local_secondary_indexes == []:
-            return self.error(
-                "ValidationException",
-                "One or more parameter values were invalid: List of LocalSecondaryIndexes is empty",
+            raise MockValidationException(
+                "One or more parameter values were invalid: List of LocalSecondaryIndexes is empty"
             )
         local_secondary_indexes = local_secondary_indexes or []
         # Verify AttributeDefinitions list all
@@ -241,76 +229,62 @@ def create_table(self):
             sse_specification=sse_spec,
             tags=tags,
         )
-        if table is not None:
-            return dynamo_json_dump(table.describe())
-        else:
-            er = "com.amazonaws.dynamodb.v20111205#ResourceInUseException"
-            return self.error(er, "Resource in use")
+        return dynamo_json_dump(table.describe())
 
     def _throw_attr_error(self, actual_attrs, expected_attrs, indexes):
         def dump_list(list_):
             return str(list_).replace("'", "")
 
-        er = "com.amazonaws.dynamodb.v20111205#ValidationException"
         err_head = "One or more parameter values were invalid: "
         if len(actual_attrs) > len(expected_attrs):
             if indexes:
-                return self.error(
-                    er,
+                raise MockValidationException(
                     err_head
                     + "Some AttributeDefinitions are not used. AttributeDefinitions: "
                     + dump_list(actual_attrs)
                     + ", keys used: "
-                    + dump_list(expected_attrs),
+                    + dump_list(expected_attrs)
                 )
             else:
-                return self.error(
-                    er,
+                raise MockValidationException(
                     err_head
-                    + "Number of attributes in KeySchema does not exactly match number of attributes defined in AttributeDefinitions",
+                    + "Number of attributes in KeySchema does not exactly match number of attributes defined in AttributeDefinitions"
                 )
         elif len(actual_attrs) < len(expected_attrs):
             if indexes:
-                return self.error(
-                    er,
+                raise MockValidationException(
                     err_head
                     + "Some index key attributes are not defined in AttributeDefinitions. Keys: "
                     + dump_list(list(set(expected_attrs) - set(actual_attrs)))
                     + ", AttributeDefinitions: "
-                    + dump_list(actual_attrs),
+                    + dump_list(actual_attrs)
                 )
             else:
-                return self.error(
-                    er, "Invalid KeySchema: Some index key attribute have no definition"
+                raise MockValidationException(
+                    "Invalid KeySchema: Some index key attribute have no definition"
                 )
         else:
             if indexes:
-                return self.error(
-                    er,
+                raise MockValidationException(
                     err_head
                     + "Some index key attributes are not defined in AttributeDefinitions. Keys: "
                     + dump_list(list(set(expected_attrs) - set(actual_attrs)))
                     + ", AttributeDefinitions: "
-                    + dump_list(actual_attrs),
+                    + dump_list(actual_attrs)
                 )
             else:
-                return self.error(
-                    er,
+                raise MockValidationException(
                     err_head
                     + "Some index key attributes are not defined in AttributeDefinitions. Keys: "
                     + dump_list(expected_attrs)
                     + ", AttributeDefinitions: "
-                    + dump_list(actual_attrs),
+                    + dump_list(actual_attrs)
                 )
 
     def delete_table(self):
         name = self.body["TableName"]
         table = self.dynamodb_backend.delete_table(name)
-        if table is not None:
-            return dynamo_json_dump(table.describe())
-        else:
-            er = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
-            return self.error(er, "Requested resource not found")
+        return dynamo_json_dump(table.describe())
 
     def describe_endpoints(self):
         response = {"Endpoints": self.dynamodb_backend.describe_endpoints()}
@@ -329,26 +303,22 @@ def untag_resource(self):
         return ""
 
     def list_tags_of_resource(self):
-        try:
-            table_arn = self.body["ResourceArn"]
-            all_tags = self.dynamodb_backend.list_tags_of_resource(table_arn)
-            all_tag_keys = [tag["Key"] for tag in all_tags]
-            marker = self.body.get("NextToken")
-            if marker:
-                start = all_tag_keys.index(marker) + 1
-            else:
-                start = 0
-            max_items = 10  # there is no default, but using 10 to make testing easier
-            tags_resp = all_tags[start : start + max_items]
-            next_marker = None
-            if len(all_tags) > start + max_items:
-                next_marker = tags_resp[-1]["Key"]
-            if next_marker:
-                return json.dumps({"Tags": tags_resp, "NextToken": next_marker})
-            return json.dumps({"Tags": tags_resp})
-        except AttributeError:
-            er = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
-            return self.error(er, "Requested resource not found")
+        table_arn = self.body["ResourceArn"]
+        all_tags = self.dynamodb_backend.list_tags_of_resource(table_arn)
+        all_tag_keys = [tag["Key"] for tag in all_tags]
+        marker = self.body.get("NextToken")
+        if marker:
+            start = all_tag_keys.index(marker) + 1
+        else:
+            start = 0
+        max_items = 10  # there is no default, but using 10 to make testing easier
+        tags_resp = all_tags[start : start + max_items]
+        next_marker = None
+        if len(all_tags) > start + max_items:
+            next_marker = tags_resp[-1]["Key"]
+        if next_marker:
+            return json.dumps({"Tags": tags_resp, "NextToken": next_marker})
+        return json.dumps({"Tags": tags_resp})
 
     def update_table(self):
         name = self.body["TableName"]
@@ -357,28 +327,20 @@ def update_table(self):
         throughput = self.body.get("ProvisionedThroughput", None)
         billing_mode = self.body.get("BillingMode", None)
         stream_spec = self.body.get("StreamSpecification", None)
-        try:
-            table = self.dynamodb_backend.update_table(
-                name=name,
-                attr_definitions=attr_definitions,
-                global_index=global_index,
-                throughput=throughput,
-                billing_mode=billing_mode,
-                stream_spec=stream_spec,
-            )
-            return dynamo_json_dump(table.describe())
-        except ValueError:
-            er = "com.amazonaws.dynamodb.v20111205#ResourceInUseException"
-            return self.error(er, "Cannot enable stream")
+        table = self.dynamodb_backend.update_table(
+            name=name,
+            attr_definitions=attr_definitions,
+            global_index=global_index,
+            throughput=throughput,
+            billing_mode=billing_mode,
+            stream_spec=stream_spec,
+        )
+        return dynamo_json_dump(table.describe())
 
     def describe_table(self):
         name = self.body["TableName"]
-        try:
-            table = self.dynamodb_backend.describe_table(name)
-            return dynamo_json_dump(table)
-        except KeyError:
-            er = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
-            return self.error(er, "Requested resource not found")
+        table = self.dynamodb_backend.describe_table(name)
+        return dynamo_json_dump(table)
 
     @include_consumed_capacity()
     def put_item(self):
@@ -387,8 +349,7 @@ def put_item(self):
         return_values = self.body.get("ReturnValues", "NONE")
 
         if return_values not in ("ALL_OLD", "NONE"):
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, "Return values set to invalid value")
+            raise MockValidationException("Return values set to invalid value")
 
         if put_has_empty_keys(item, self.dynamodb_backend.get_table(name)):
             return get_empty_str_error()
@@ -415,36 +376,22 @@ def put_item(self):
         if condition_expression:
             overwrite = False
 
-        try:
-            result = self.dynamodb_backend.put_item(
-                name,
-                item,
-                expected,
-                condition_expression,
-                expression_attribute_names,
-                expression_attribute_values,
-                overwrite,
-            )
-        except MockValidationException as mve:
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, mve.exception_msg)
-        except KeyError as ke:
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, ke.args[0])
-        except ValueError as ve:
-            er = "com.amazonaws.dynamodb.v20111205#ConditionalCheckFailedException"
-            return self.error(er, str(ve))
-
-        if result:
-            item_dict = result.to_json()
-            if return_values == "ALL_OLD":
-                item_dict["Attributes"] = existing_attributes
-            else:
-                item_dict.pop("Attributes", None)
-            return dynamo_json_dump(item_dict)
+        result = self.dynamodb_backend.put_item(
+            name,
+            item,
+            expected,
+            condition_expression,
+            expression_attribute_names,
+            expression_attribute_values,
+            overwrite,
+        )
+
+        item_dict = result.to_json()
+        if return_values == "ALL_OLD":
+            item_dict["Attributes"] = existing_attributes
         else:
-            er = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
-            return self.error(er, "Requested resource not found")
+            item_dict.pop("Attributes", None)
+        return dynamo_json_dump(item_dict)
 
     def batch_write_item(self):
         table_batches = self.body["RequestItems"]
@@ -455,15 +402,10 @@ def batch_write_item(self):
                 request = list(table_request.values())[0]
                 if request_type == "PutRequest":
                     item = request["Item"]
-                    res = self.dynamodb_backend.put_item(table_name, item)
-                    if not res:
-                        return self.error(
-                            "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException",
-                            "Requested resource not found",
-                        )
+                    self.dynamodb_backend.put_item(table_name, item)
                 elif request_type == "DeleteRequest":
                     keys = request["Key"]
-                    item = self.dynamodb_backend.delete_item(table_name, keys)
+                    self.dynamodb_backend.delete_item(table_name, keys)
 
         response = {
             "ConsumedCapacity": [
@@ -483,36 +425,26 @@ def batch_write_item(self):
     @include_consumed_capacity(0.5)
     def get_item(self):
         name = self.body["TableName"]
-        table = self.dynamodb_backend.get_table(name)
-        if table is None:
-            return self.error(
-                "com.amazonaws.dynamodb.v20120810#ResourceNotFoundException",
-                "Requested resource not found",
-            )
+        self.dynamodb_backend.get_table(name)
         key = self.body["Key"]
         projection_expression = self.body.get("ProjectionExpression")
         expression_attribute_names = self.body.get("ExpressionAttributeNames")
         if expression_attribute_names == {}:
             if projection_expression is None:
-                er = "ValidationException"
-                return self.error(
-                    er,
-                    "ExpressionAttributeNames can only be specified when using expressions",
+                raise MockValidationException(
+                    "ExpressionAttributeNames can only be specified when using expressions"
                 )
             else:
-                er = "ValidationException"
-                return self.error(er, "ExpressionAttributeNames must not be empty")
+                raise MockValidationException(
+                    "ExpressionAttributeNames must not be empty"
+                )
 
         expression_attribute_names = expression_attribute_names or {}
         projection_expression = self._adjust_projection_expression(
             projection_expression, expression_attribute_names
         )
 
-        try:
-            item = self.dynamodb_backend.get_item(name, key, projection_expression)
-        except ValueError:
-            er = "com.amazon.coral.validate#ValidationException"
-            return self.error(er, "Validation Exception")
+        item = self.dynamodb_backend.get_item(name, key, projection_expression)
         if item:
             item_dict = item.describe_attrs(attributes=None)
             return dynamo_json_dump(item_dict)
@@ -529,27 +461,26 @@ def batch_get_item(self):
         # Scenario 1: We're requesting more than a 100 keys from a single table
         for table_name, table_request in table_batches.items():
             if len(table_request["Keys"]) > 100:
-                return self.error(
-                    "com.amazonaws.dynamodb.v20111205#ValidationException",
+                raise MockValidationException(
                     "1 validation error detected: Value at 'requestItems."
                     + table_name
-                    + ".member.keys' failed to satisfy constraint: Member must have length less than or equal to 100",
+                    + ".member.keys' failed to satisfy constraint: Member must have length less than or equal to 100"
                 )
         # Scenario 2: We're requesting more than a 100 keys across all tables
         nr_of_keys_across_all_tables = sum(
             [len(req["Keys"]) for _, req in table_batches.items()]
         )
         if nr_of_keys_across_all_tables > 100:
-            return self.error(
-                "com.amazonaws.dynamodb.v20111205#ValidationException",
-                "Too many items requested for the BatchGetItem call",
+            raise MockValidationException(
+                "Too many items requested for the BatchGetItem call"
             )
 
         for table_name, table_request in table_batches.items():
             keys = table_request["Keys"]
             if self._contains_duplicates(keys):
-                er = "com.amazon.coral.validate#ValidationException"
-                return self.error(er, "Provided list of item keys contains duplicates")
+                raise MockValidationException(
+                    "Provided list of item keys contains duplicates"
+                )
             attributes_to_get = table_request.get("AttributesToGet")
             projection_expression = table_request.get("ProjectionExpression")
             expression_attribute_names = table_request.get(
@@ -562,15 +493,9 @@ def batch_get_item(self):
 
             results["Responses"][table_name] = []
             for key in keys:
-                try:
-                    item = self.dynamodb_backend.get_item(
-                        table_name, key, projection_expression
-                    )
-                except ValueError:
-                    return self.error(
-                        "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException",
-                        "Requested resource not found",
-                    )
+                item = self.dynamodb_backend.get_item(
+                    table_name, key, projection_expression
+                )
                 if item:
                     item_describe = item.describe_attrs(attributes_to_get)
                     results["Responses"][table_name].append(item_describe["Item"])
@@ -607,31 +532,10 @@ def query(self):
         if key_condition_expression:
             value_alias_map = self.body.get("ExpressionAttributeValues", {})
 
-            table = self.dynamodb_backend.get_table(name)
-
-            # If table does not exist
-            if table is None:
-                return self.error(
-                    "com.amazonaws.dynamodb.v20120810#ResourceNotFoundException",
-                    "Requested resource not found",
-                )
-
             index_name = self.body.get("IndexName")
-            if index_name:
-                all_indexes = (table.global_indexes or []) + (table.indexes or [])
-                indexes_by_name = dict((i.name, i) for i in all_indexes)
-                if index_name not in indexes_by_name:
-                    er = "com.amazonaws.dynamodb.v20120810#ResourceNotFoundException"
-                    return self.error(
-                        er,
-                        "Invalid index: {} for table: {}. Available indexes are: {}".format(
-                            index_name, name, ", ".join(indexes_by_name.keys())
-                        ),
-                    )
-
-                index = indexes_by_name[index_name].schema
-            else:
-                index = table.schema
+            schema = self.dynamodb_backend.get_schema(
+                table_name=name, index_name=index_name
+            )
 
             reverse_attribute_lookup = dict(
                 (v, k) for k, v in self.body.get("ExpressionAttributeNames", {}).items()
@@ -642,7 +546,7 @@ def query(self):
                     " AND ", key_condition_expression, maxsplit=1, flags=re.IGNORECASE
                 )
 
-                index_hash_key = [key for key in index if key["KeyType"] == "HASH"][0]
+                index_hash_key = [key for key in schema if key["KeyType"] == "HASH"][0]
                 hash_key_var = reverse_attribute_lookup.get(
                     index_hash_key["AttributeName"], index_hash_key["AttributeName"]
                 )
@@ -656,11 +560,10 @@ def query(self):
                     (None, None),
                 )
                 if hash_key_expression is None:
-                    return self.error(
-                        "ValidationException",
+                    raise MockValidationException(
                         "Query condition missed key schema element: {}".format(
                             hash_key_var
-                        ),
+                        )
                     )
                 hash_key_expression = hash_key_expression.strip("()")
                 expressions.pop(i)
@@ -698,11 +601,10 @@ def query(self):
                             "begins_with"
                         )
                     ]
-                    return self.error(
-                        "com.amazonaws.dynamodb.v20111205#ValidationException",
+                    raise MockValidationException(
                         "Invalid KeyConditionExpression: Invalid function name; function: {}".format(
                             function_used
-                        ),
+                        )
                     )
                 else:
                     # [range_key, =, x]
@@ -713,14 +615,13 @@ def query(self):
                     supplied_range_key, supplied_range_key
                 )
                 range_keys = [
-                    k["AttributeName"] for k in index if k["KeyType"] == "RANGE"
+                    k["AttributeName"] for k in schema if k["KeyType"] == "RANGE"
                 ]
                 if supplied_range_key not in range_keys:
-                    return self.error(
-                        "ValidationException",
+                    raise MockValidationException(
                         "Query condition missed key schema element: {}".format(
                             range_keys[0]
-                        ),
+                        )
                     )
             else:
                 hash_key_expression = key_condition_expression.strip("()")
@@ -728,10 +629,7 @@ def query(self):
                 range_values = []
 
             if not re.search("[^<>]=", hash_key_expression):
-                return self.error(
-                    "com.amazonaws.dynamodb.v20111205#ValidationException",
-                    "Query key condition not supported",
-                )
+                raise MockValidationException("Query key condition not supported")
             hash_key_value_alias = hash_key_expression.split("=")[1].strip()
             # Temporary fix until we get proper KeyConditionExpression function
             hash_key = value_alias_map.get(
@@ -743,9 +641,8 @@ def query(self):
             query_filters = self.body.get("QueryFilter")
 
             if not (key_conditions or query_filters):
-                return self.error(
-                    "com.amazonaws.dynamodb.v20111205#ValidationException",
-                    "Either KeyConditions or QueryFilter should be present",
+                raise MockValidationException(
+                    "Either KeyConditions or QueryFilter should be present"
                 )
 
             if key_conditions:
@@ -759,16 +656,14 @@ def query(self):
                     if key not in (hash_key_name, range_key_name):
                         filter_kwargs[key] = value
                 if hash_key_name is None:
-                    er = "'com.amazonaws.dynamodb.v20120810#ResourceNotFoundException"
-                    return self.error(er, "Requested resource not found")
+                    raise ResourceNotFoundException
                 hash_key = key_conditions[hash_key_name]["AttributeValueList"][0]
                 if len(key_conditions) == 1:
                     range_comparison = None
                     range_values = []
                 else:
                     if range_key_name is None and not filter_kwargs:
-                        er = "com.amazon.coral.validate#ValidationException"
-                        return self.error(er, "Validation Exception")
+                        raise MockValidationException("Validation Exception")
                     else:
                         range_condition = key_conditions.get(range_key_name)
                         if range_condition:
@@ -798,9 +693,6 @@ def query(self):
             filter_expression=filter_expression,
             **filter_kwargs
         )
-        if items is None:
-            er = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
-            return self.error(er, "Requested resource not found")
 
         result = {
             "Count": len(items),
@@ -867,21 +759,8 @@ def scan(self):
                 index_name,
                 projection_expression,
             )
-        except InvalidIndexNameError as err:
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, str(err))
         except ValueError as err:
-            er = "com.amazonaws.dynamodb.v20111205#ValidationError"
-            return self.error(er, "Bad Filter Expression: {0}".format(err))
-        except Exception as err:
-            er = "com.amazonaws.dynamodb.v20111205#InternalFailure"
-            return self.error(er, "Internal error. {0}".format(err))
-
-        # Items should be a list, at least an empty one. Is None if table does not exist.
-        # Should really check this at the beginning
-        if items is None:
-            er = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
-            return self.error(er, "Requested resource not found")
+            raise MockValidationException("Bad Filter Expression: {0}".format(err))
 
         result = {
             "Count": len(items),
@@ -897,14 +776,13 @@ def delete_item(self):
         key = self.body["Key"]
         return_values = self.body.get("ReturnValues", "NONE")
         if return_values not in ("ALL_OLD", "NONE"):
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, "Return values set to invalid value")
-
-        table = self.dynamodb_backend.get_table(name)
-        if not table:
-            er = "com.amazonaws.dynamodb.v20120810#ConditionalCheckFailedException"
-            return self.error(
-                er, "A condition specified in the operation could not be evaluated."
+            raise MockValidationException("Return values set to invalid value")
+
+        try:
+            self.dynamodb_backend.get_table(name)
+        except ResourceNotFoundException:
+            raise ConditionalCheckFailed(
+                "A condition specified in the operation could not be evaluated."
             )
 
         # Attempt to parse simple ConditionExpressions into an Expected
@@ -913,19 +791,13 @@ def delete_item(self):
         expression_attribute_names = self.body.get("ExpressionAttributeNames", {})
         expression_attribute_values = self.body.get("ExpressionAttributeValues", {})
 
-        try:
-            item = self.dynamodb_backend.delete_item(
-                name,
-                key,
-                expression_attribute_names,
-                expression_attribute_values,
-                condition_expression,
-            )
-        except ValueError:
-            er = "com.amazonaws.dynamodb.v20111205#ConditionalCheckFailedException"
-            return self.error(
-                er, "A condition specified in the operation could not be evaluated."
-            )
+        item = self.dynamodb_backend.delete_item(
+            name,
+            key,
+            expression_attribute_names,
+            expression_attribute_values,
+            condition_expression,
+        )
 
         if item and return_values == "ALL_OLD":
             item_dict = item.to_json()
@@ -941,19 +813,11 @@ def update_item(self):
         update_expression = self.body.get("UpdateExpression", "").strip()
         attribute_updates = self.body.get("AttributeUpdates")
         if update_expression and attribute_updates:
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(
-                er,
-                "Can not use both expression and non-expression parameters in the same request: Non-expression parameters: {AttributeUpdates} Expression parameters: {UpdateExpression}",
+            raise MockValidationException(
+                "Can not use both expression and non-expression parameters in the same request: Non-expression parameters: {AttributeUpdates} Expression parameters: {UpdateExpression}"
             )
         # We need to copy the item in order to avoid it being modified by the update_item operation
-        try:
-            existing_item = copy.deepcopy(self.dynamodb_backend.get_item(name, key))
-        except ValueError:
-            return self.error(
-                "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException",
-                "Requested resource not found",
-            )
+        existing_item = copy.deepcopy(self.dynamodb_backend.get_item(name, key))
         if existing_item:
             existing_attributes = existing_item.to_json()["Attributes"]
         else:
@@ -966,8 +830,7 @@ def update_item(self):
             "UPDATED_OLD",
             "UPDATED_NEW",
         ):
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, "Return values set to invalid value")
+            raise MockValidationException("Return values set to invalid value")
 
         if "Expected" in self.body:
             expected = self.body["Expected"]
@@ -980,26 +843,16 @@ def update_item(self):
         expression_attribute_names = self.body.get("ExpressionAttributeNames", {})
         expression_attribute_values = self.body.get("ExpressionAttributeValues", {})
 
-        try:
-            item = self.dynamodb_backend.update_item(
-                name,
-                key,
-                update_expression=update_expression,
-                attribute_updates=attribute_updates,
-                expression_attribute_names=expression_attribute_names,
-                expression_attribute_values=expression_attribute_values,
-                expected=expected,
-                condition_expression=condition_expression,
-            )
-        except MockValidationException as mve:
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, mve.exception_msg)
-        except ValueError:
-            er = "com.amazonaws.dynamodb.v20111205#ConditionalCheckFailedException"
-            return self.error(er, "The conditional request failed")
-        except TypeError:
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, "Validation Exception")
+        item = self.dynamodb_backend.update_item(
+            name,
+            key,
+            update_expression=update_expression,
+            attribute_updates=attribute_updates,
+            expression_attribute_names=expression_attribute_names,
+            expression_attribute_values=expression_attribute_values,
+            expected=expected,
+            condition_expression=condition_expression,
+        )
 
         item_dict = item.to_json()
         item_dict["ConsumedCapacity"] = {"TableName": name, "CapacityUnits": 0.5}
@@ -1100,7 +953,7 @@ def transact_get_items(self):
                 % TRANSACTION_MAX_ITEMS
             )
 
-            return self.error("ValidationException", msg)
+            raise MockValidationException(msg)
 
         ret_consumed_capacity = self.body.get("ReturnConsumedCapacity", "NONE")
         consumed_capacity = dict()
@@ -1109,11 +962,7 @@ def transact_get_items(self):
 
             table_name = transact_item["Get"]["TableName"]
             key = transact_item["Get"]["Key"]
-            try:
-                item = self.dynamodb_backend.get_item(table_name, key)
-            except ValueError:
-                er = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
-                return self.error(er, "Requested resource not found")
+            item = self.dynamodb_backend.get_item(table_name, key)
 
             if not item:
                 responses.append({})
@@ -1145,26 +994,13 @@ def transact_get_items(self):
 
     def transact_write_items(self):
         transact_items = self.body["TransactItems"]
-        try:
-            self.dynamodb_backend.transact_write_items(transact_items)
-        except TransactionCanceledException as e:
-            er = "com.amazonaws.dynamodb.v20111205#TransactionCanceledException"
-            return self.error(er, str(e))
-        except MockValidationException as mve:
-            er = "com.amazonaws.dynamodb.v20111205#ValidationException"
-            return self.error(er, mve.exception_msg)
+        self.dynamodb_backend.transact_write_items(transact_items)
         response = {"ConsumedCapacity": [], "ItemCollectionMetrics": {}}
         return dynamo_json_dump(response)
 
     def describe_continuous_backups(self):
         name = self.body["TableName"]
 
-        if self.dynamodb_backend.get_table(name) is None:
-            return self.error(
-                "com.amazonaws.dynamodb.v20111205#TableNotFoundException",
-                "Table not found: {}".format(name),
-            )
-
         response = self.dynamodb_backend.describe_continuous_backups(name)
 
         return json.dumps({"ContinuousBackupsDescription": response})
@@ -1173,12 +1009,6 @@ def update_continuous_backups(self):
         name = self.body["TableName"]
         point_in_time_spec = self.body["PointInTimeRecoverySpecification"]
 
-        if self.dynamodb_backend.get_table(name) is None:
-            return self.error(
-                "com.amazonaws.dynamodb.v20111205#TableNotFoundException",
-                "Table not found: {}".format(name),
-            )
-
         response = self.dynamodb_backend.update_continuous_backups(
             name, point_in_time_spec
         )
@@ -1196,64 +1026,38 @@ def create_backup(self):
         body = self.body
         table_name = body.get("TableName")
         backup_name = body.get("BackupName")
-        try:
-            backup = self.dynamodb_backend.create_backup(table_name, backup_name)
-            response = {"BackupDetails": backup.details}
-            return dynamo_json_dump(response)
-        except KeyError:
-            er = "com.amazonaws.dynamodb.v20111205#TableNotFoundException"
-            return self.error(er, "Table not found: %s" % table_name)
+        backup = self.dynamodb_backend.create_backup(table_name, backup_name)
+        response = {"BackupDetails": backup.details}
+        return dynamo_json_dump(response)
 
     def delete_backup(self):
         body = self.body
         backup_arn = body.get("BackupArn")
-        try:
-            backup = self.dynamodb_backend.delete_backup(backup_arn)
-            response = {"BackupDescription": backup.description}
-            return dynamo_json_dump(response)
-        except KeyError:
-            er = "com.amazonaws.dynamodb.v20111205#BackupNotFoundException"
-            return self.error(er, "Backup not found: %s" % backup_arn)
+        backup = self.dynamodb_backend.delete_backup(backup_arn)
+        response = {"BackupDescription": backup.description}
+        return dynamo_json_dump(response)
 
     def describe_backup(self):
         body = self.body
         backup_arn = body.get("BackupArn")
-        try:
-            backup = self.dynamodb_backend.describe_backup(backup_arn)
-            response = {"BackupDescription": backup.description}
-            return dynamo_json_dump(response)
-        except KeyError:
-            er = "com.amazonaws.dynamodb.v20111205#BackupNotFoundException"
-            return self.error(er, "Backup not found: %s" % backup_arn)
+        backup = self.dynamodb_backend.describe_backup(backup_arn)
+        response = {"BackupDescription": backup.description}
+        return dynamo_json_dump(response)
 
     def restore_table_from_backup(self):
         body = self.body
         target_table_name = body.get("TargetTableName")
         backup_arn = body.get("BackupArn")
-        try:
-            restored_table = self.dynamodb_backend.restore_table_from_backup(
-                target_table_name, backup_arn
-            )
-            return dynamo_json_dump(restored_table.describe())
-        except KeyError:
-            er = "com.amazonaws.dynamodb.v20111205#BackupNotFoundException"
-            return self.error(er, "Backup not found: %s" % backup_arn)
-        except ValueError:
-            er = "com.amazonaws.dynamodb.v20111205#TableAlreadyExistsException"
-            return self.error(er, "Table already exists: %s" % target_table_name)
+        restored_table = self.dynamodb_backend.restore_table_from_backup(
+            target_table_name, backup_arn
+        )
+        return dynamo_json_dump(restored_table.describe())
 
     def restore_table_to_point_in_time(self):
         body = self.body
         target_table_name = body.get("TargetTableName")
         source_table_name = body.get("SourceTableName")
-        try:
-            restored_table = self.dynamodb_backend.restore_table_to_point_in_time(
-                target_table_name, source_table_name
-            )
-            return dynamo_json_dump(restored_table.describe())
-        except KeyError:
-            er = "com.amazonaws.dynamodb.v20111205#SourceTableNotFoundException"
-            return self.error(er, "Source table not found: %s" % source_table_name)
-        except ValueError:
-            er = "com.amazonaws.dynamodb.v20111205#TableAlreadyExistsException"
-            return self.error(er, "Table already exists: %s" % target_table_name)
+        restored_table = self.dynamodb_backend.restore_table_to_point_in_time(
+            target_table_name, source_table_name
+        )
+        return dynamo_json_dump(restored_table.describe())

Hi @Gabe1203, welcome to Moto! Will mark it as an enhancement to extend the error handling here.
diff --git a/tests/test_dynamodb/test_dynamodb.py b/tests/test_dynamodb/test_dynamodb.py
--- a/tests/test_dynamodb/test_dynamodb.py
+++ b/tests/test_dynamodb/test_dynamodb.py
@@ -6,7 +6,7 @@
 from boto3.dynamodb.conditions import Attr, Key
 import re
 import sure  # noqa # pylint: disable=unused-import
-from moto import mock_dynamodb
+from moto import mock_dynamodb, settings
 from moto.dynamodb import dynamodb_backends
 from botocore.exceptions import ClientError
 
@@ -399,14 +399,13 @@ def test_put_item_with_streams():
             "Data": {"M": {"Key1": {"S": "Value1"}, "Key2": {"S": "Value2"}}},
         }
     )
-    table = dynamodb_backends["us-west-2"].get_table(name)
-    if not table:
-        # There is no way to access stream data over the API, so this part can't run in server-tests mode.
-        return
-    len(table.stream_shard.items).should.be.equal(1)
-    stream_record = table.stream_shard.items[0].record
-    stream_record["eventName"].should.be.equal("INSERT")
-    stream_record["dynamodb"]["SizeBytes"].should.be.equal(447)
+
+    if not settings.TEST_SERVER_MODE:
+        table = dynamodb_backends["us-west-2"].get_table(name)
+        len(table.stream_shard.items).should.be.equal(1)
+        stream_record = table.stream_shard.items[0].record
+        stream_record["eventName"].should.be.equal("INSERT")
+        stream_record["dynamodb"]["SizeBytes"].should.be.equal(447)
 
 
 @mock_dynamodb
@@ -1597,12 +1596,9 @@ def test_bad_scan_filter():
     table = dynamodb.Table("test1")
 
     # Bad expression
-    try:
+    with pytest.raises(ClientError) as exc:
         table.scan(FilterExpression="client test")
-    except ClientError as err:
-        err.response["Error"]["Code"].should.equal("ValidationError")
-    else:
-        raise RuntimeError("Should have raised ResourceInUseException")
+    exc.value.response["Error"]["Code"].should.equal("ValidationException")
 
 
 @mock_dynamodb
@@ -3870,6 +3866,12 @@ def test_transact_write_items_put_conditional_expressions():
         )
     # Assert the exception is correct
     ex.value.response["Error"]["Code"].should.equal("TransactionCanceledException")
+    reasons = ex.value.response["CancellationReasons"]
+    reasons.should.have.length_of(5)
+    reasons.should.contain(
+        {"Code": "ConditionalCheckFailed", "Message": "The conditional request failed"}
+    )
+    reasons.should.contain({"Code": "None"})
     ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
     # Assert all are present
     items = dynamodb.scan(TableName="test-table")["Items"]

DynamoDB - Provide full cancellation reasons for write_transactions failue
When using boto3 and `write_transactions`, if the transaction fails we normally get an error response that includes the full cancellation reasons: 

```
"CancellationReasons": [
            {
                "Code": "None"
            },
            {
                "Code": "ConditionalCheckFailed",
                "Message": "The conditional request failed"
            },
            {
                "Code": "None"
            }
        ]
```

when using moto (version 2.3.0) this object is not returned on transaction failure, we just get the error message: 
`"Error with transaction: An error occurred (TransactionCanceledException) when calling the TransactWriteItems operation: Transaction cancelled, please refer cancellation reasons for specific reasons [None, ConditionalCheckFailed, None]"`

Would be great to get the full cancellation reasons so we can better test the our transaction error handling. 

Thanks! 


tests/test_dynamodb/test_dynamodb.py::test_bad_scan_filter tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_put_conditional_expressions
tests/test_dynamodb/test_dynamodb.py::test_remove_list_index__remove_existing_nested_index tests/test_dynamodb/test_dynamodb.py::test_basic_projection_expressions_using_query_with_attr_expression_names tests/test_dynamodb/test_dynamodb.py::test_describe_backup_for_non_existent_backup_raises_error tests/test_dynamodb/test_dynamodb.py::test_update_item_with_list tests/test_dynamodb/test_dynamodb.py::test_describe_continuous_backups_errors tests/test_dynamodb/test_dynamodb.py::test_describe_missing_table_boto3 tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_delete tests/test_dynamodb/test_dynamodb.py::test_remove_list_index__remove_existing_index tests/test_dynamodb/test_dynamodb.py::test_dynamodb_update_item_fails_on_string_sets tests/test_dynamodb/test_dynamodb.py::test_projection_expression_execution_order tests/test_dynamodb/test_dynamodb.py::test_update_item_with_attribute_in_right_hand_side_and_operation tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_conditioncheck_passes tests/test_dynamodb/test_dynamodb.py::test_basic_projection_expression_using_get_item tests/test_dynamodb/test_dynamodb.py::test_list_table_tags_empty tests/test_dynamodb/test_dynamodb.py::test_create_backup_for_non_existent_table_raises_error tests/test_dynamodb/test_dynamodb.py::test_update_expression_with_plus_in_attribute_name tests/test_dynamodb/test_dynamodb.py::test_scan_filter2 tests/test_dynamodb/test_dynamodb.py::test_describe_backup tests/test_dynamodb/test_dynamodb.py::test_batch_write_item tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_delete_with_successful_condition_expression tests/test_dynamodb/test_dynamodb.py::test_query_invalid_table tests/test_dynamodb/test_dynamodb.py::test_delete_backup tests/test_dynamodb/test_dynamodb.py::test_nested_projection_expression_using_query tests/test_dynamodb/test_dynamodb.py::test_transact_get_items_should_return_empty_map_for_non_existent_item tests/test_dynamodb/test_dynamodb.py::test_update_supports_list_append_maps tests/test_dynamodb/test_dynamodb.py::test_update_supports_nested_update_if_nested_value_not_exists tests/test_dynamodb/test_dynamodb.py::test_index_with_unknown_attributes_should_fail tests/test_dynamodb/test_dynamodb.py::test_update_supports_nested_list_append tests/test_dynamodb/test_dynamodb.py::test_update_item_atomic_counter_from_zero tests/test_dynamodb/test_dynamodb.py::test_update_item_if_original_value_is_none tests/test_dynamodb/test_dynamodb.py::test_filter_expression_execution_order tests/test_dynamodb/test_dynamodb.py::test_delete_item tests/test_dynamodb/test_dynamodb.py::test_list_tables_paginated tests/test_dynamodb/test_dynamodb.py::test_query_gsi_with_range_key tests/test_dynamodb/test_dynamodb.py::test_valid_transact_get_items tests/test_dynamodb/test_dynamodb.py::test_describe_continuous_backups tests/test_dynamodb/test_dynamodb.py::test_remove_list_index__remove_existing_double_nested_index tests/test_dynamodb/test_dynamodb.py::test_non_existing_attribute_should_raise_exception tests/test_dynamodb/test_dynamodb.py::test_scan_by_non_exists_index tests/test_dynamodb/test_dynamodb.py::test_update_supports_list_append_with_nested_if_not_exists_operation tests/test_dynamodb/test_dynamodb.py::test_put_empty_item tests/test_dynamodb/test_dynamodb.py::test_gsi_lastevaluatedkey tests/test_dynamodb/test_dynamodb.py::test_query_catches_when_no_filters tests/test_dynamodb/test_dynamodb.py::test_restore_table_to_point_in_time_raises_error_when_dest_exist tests/test_dynamodb/test_dynamodb.py::test_update_supports_complex_expression_attribute_values tests/test_dynamodb/test_dynamodb.py::test_update_expression_with_numeric_literal_instead_of_value tests/test_dynamodb/test_dynamodb.py::test_list_tables_boto3[multiple-tables] tests/test_dynamodb/test_dynamodb.py::test_batch_items_throws_exception_when_requesting_100_items_across_all_tables tests/test_dynamodb/test_dynamodb.py::test_update_list_index__set_nested_index_out_of_range tests/test_dynamodb/test_dynamodb.py::test_update_nested_item_if_original_value_is_none tests/test_dynamodb/test_dynamodb.py::test_update_supports_multiple_levels_nested_list_append tests/test_dynamodb/test_dynamodb.py::test_list_tables_boto3[one-table] tests/test_dynamodb/test_dynamodb.py::test_put_item_nonexisting_range_key tests/test_dynamodb/test_dynamodb.py::test_list_backups tests/test_dynamodb/test_dynamodb.py::test_query_filter_overlapping_expression_prefixes tests/test_dynamodb/test_dynamodb.py::test_batch_items_with_basic_projection_expression tests/test_dynamodb/test_dynamodb.py::test_source_and_restored_table_items_are_not_linked tests/test_dynamodb/test_dynamodb.py::test_update_expression_with_minus_in_attribute_name tests/test_dynamodb/test_dynamodb.py::test_restore_table_from_backup_raises_error_when_table_already_exists tests/test_dynamodb/test_dynamodb.py::test_put_item_nonexisting_hash_key tests/test_dynamodb/test_dynamodb.py::test_restore_table_to_point_in_time_raises_error_when_source_not_exist tests/test_dynamodb/test_dynamodb.py::test_invalid_transact_get_items tests/test_dynamodb/test_dynamodb.py::test_basic_projection_expression_using_get_item_with_attr_expression_names tests/test_dynamodb/test_dynamodb.py::test_batch_items_should_throw_exception_for_duplicate_request tests/test_dynamodb/test_dynamodb.py::test_update_continuous_backups tests/test_dynamodb/test_dynamodb.py::test_update_item_atomic_counter tests/test_dynamodb/test_dynamodb.py::test_nested_projection_expression_using_query_with_attr_expression_names tests/test_dynamodb/test_dynamodb.py::test_item_add_empty_string_range_key_exception tests/test_dynamodb/test_dynamodb.py::test_get_item_for_non_existent_table_raises_error tests/test_dynamodb/test_dynamodb.py::test_batch_items_returns_all tests/test_dynamodb/test_dynamodb.py::test_scan_filter tests/test_dynamodb/test_dynamodb.py::test_error_when_providing_expression_and_nonexpression_params tests/test_dynamodb/test_dynamodb.py::test_update_return_attributes tests/test_dynamodb/test_dynamodb.py::test_item_size_is_under_400KB tests/test_dynamodb/test_dynamodb.py::test_multiple_updates tests/test_dynamodb/test_dynamodb.py::test_scan_filter4 tests/test_dynamodb/test_dynamodb.py::test_nested_projection_expression_using_get_item_with_attr_expression tests/test_dynamodb/test_dynamodb.py::test_update_expression_with_space_in_attribute_name tests/test_dynamodb/test_dynamodb.py::test_create_multiple_backups_with_same_name tests/test_dynamodb/test_dynamodb.py::test_gsi_projection_type_keys_only tests/test_dynamodb/test_dynamodb.py::test_update_list_index__set_existing_nested_index tests/test_dynamodb/test_dynamodb.py::test_set_attribute_is_dropped_if_empty_after_update_expression[use tests/test_dynamodb/test_dynamodb.py::test_batch_items_throws_exception_when_requesting_100_items_for_single_table tests/test_dynamodb/test_dynamodb.py::test_restore_table_from_non_existent_backup_raises_error tests/test_dynamodb/test_dynamodb.py::test_query_by_non_exists_index tests/test_dynamodb/test_dynamodb.py::test_gsi_key_cannot_be_empty tests/test_dynamodb/test_dynamodb.py::test_remove_list_index__remove_index_out_of_range tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_fails_with_transaction_canceled_exception tests/test_dynamodb/test_dynamodb.py::test_update_if_not_exists tests/test_dynamodb/test_dynamodb.py::test_remove_top_level_attribute tests/test_dynamodb/test_dynamodb.py::test_restore_table_from_backup tests/test_dynamodb/test_dynamodb.py::test_nested_projection_expression_using_scan_with_attr_expression_names tests/test_dynamodb/test_dynamodb.py::test_batch_items_with_basic_projection_expression_and_attr_expression_names tests/test_dynamodb/test_dynamodb.py::test_update_non_existing_item_raises_error_and_does_not_contain_item_afterwards tests/test_dynamodb/test_dynamodb.py::test_allow_update_to_item_with_different_type tests/test_dynamodb/test_dynamodb.py::test_describe_limits tests/test_dynamodb/test_dynamodb.py::test_sorted_query_with_numerical_sort_key tests/test_dynamodb/test_dynamodb.py::test_dynamodb_max_1mb_limit tests/test_dynamodb/test_dynamodb.py::test_update_continuous_backups_errors tests/test_dynamodb/test_dynamodb.py::test_list_backups_for_non_existent_table tests/test_dynamodb/test_dynamodb.py::test_duplicate_create tests/test_dynamodb/test_dynamodb.py::test_summing_up_2_strings_raises_exception tests/test_dynamodb/test_dynamodb.py::test_update_item_atomic_counter_return_values tests/test_dynamodb/test_dynamodb.py::test_put_item_with_special_chars tests/test_dynamodb/test_dynamodb.py::test_query_missing_expr_names tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_update_with_failed_condition_expression tests/test_dynamodb/test_dynamodb.py::test_describe_endpoints[eu-central-1] tests/test_dynamodb/test_dynamodb.py::test_create_backup tests/test_dynamodb/test_dynamodb.py::test_query_filter tests/test_dynamodb/test_dynamodb.py::test_list_tables_boto3[no-table] tests/test_dynamodb/test_dynamodb.py::test_attribute_item_delete tests/test_dynamodb/test_dynamodb.py::test_update_list_index__set_existing_index tests/test_dynamodb/test_dynamodb.py::test_nested_projection_expression_using_get_item tests/test_dynamodb/test_dynamodb.py::test_nested_projection_expression_using_scan tests/test_dynamodb/test_dynamodb.py::test_update_item_on_map tests/test_dynamodb/test_dynamodb.py::test_restore_table_to_point_in_time tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_put tests/test_dynamodb/test_dynamodb.py::test_basic_projection_expressions_using_query tests/test_dynamodb/test_dynamodb.py::test_list_not_found_table_tags tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_update tests/test_dynamodb/test_dynamodb.py::test_update_return_updated_new_attributes_when_same tests/test_dynamodb/test_dynamodb.py::test_gsi_verify_negative_number_order tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_conditioncheck_fails tests/test_dynamodb/test_dynamodb.py::test_set_ttl tests/test_dynamodb/test_dynamodb.py::test_scan_filter_should_not_return_non_existing_attributes tests/test_dynamodb/test_dynamodb.py::test_update_supports_list_append tests/test_dynamodb/test_dynamodb.py::test_put_return_attributes tests/test_dynamodb/test_dynamodb.py::test_lsi_projection_type_keys_only tests/test_dynamodb/test_dynamodb.py::test_update_list_index__set_double_nested_index tests/test_dynamodb/test_dynamodb.py::test_gsi_key_can_be_updated tests/test_dynamodb/test_dynamodb.py::test_put_item_with_streams tests/test_dynamodb/test_dynamodb.py::test_list_tables_exclusive_start_table_name_empty tests/test_dynamodb/test_dynamodb.py::test_update_item_with_no_action_passed_with_list tests/test_dynamodb/test_dynamodb.py::test_update_item_with_empty_string_attr_no_exception tests/test_dynamodb/test_dynamodb.py::test_transact_write_items_delete_with_failed_condition_expression tests/test_dynamodb/test_dynamodb.py::test_basic_projection_expressions_using_scan tests/test_dynamodb/test_dynamodb.py::test_list_table_tags tests/test_dynamodb/test_dynamodb.py::test_update_supports_nested_list_append_onto_another_list tests/test_dynamodb/test_dynamodb.py::test_update_list_index__set_index_out_of_range tests/test_dynamodb/test_dynamodb.py::test_filter_expression tests/test_dynamodb/test_dynamodb.py::test_basic_projection_expressions_using_scan_with_attr_expression_names tests/test_dynamodb/test_dynamodb.py::test_item_add_empty_string_hash_key_exception tests/test_dynamodb/test_dynamodb.py::test_item_add_empty_string_attr_no_exception tests/test_dynamodb/test_dynamodb.py::test_update_item_add_to_non_existent_set tests/test_dynamodb/test_dynamodb.py::test_gsi_projection_type_include tests/test_dynamodb/test_dynamodb.py::test_query_global_secondary_index_when_created_via_update_table_resource tests/test_dynamodb/test_dynamodb.py::test_update_item_add_to_non_existent_number_set tests/test_dynamodb/test_dynamodb.py::test_update_expression_with_multiple_set_clauses_must_be_comma_separated tests/test_dynamodb/test_dynamodb.py::test_scan_filter3 tests/test_dynamodb/test_dynamodb.py::test_update_list_index__set_index_of_a_string tests/test_dynamodb/test_dynamodb.py::test_update_supports_list_append_with_nested_if_not_exists_operation_and_property_already_exists tests/test_dynamodb/test_dynamodb.py::test_describe_endpoints[ap-south-1] tests/test_dynamodb/test_dynamodb.py::test_delete_table tests/test_dynamodb/test_dynamodb.py::test_update_item_add_to_list_using_legacy_attribute_updates tests/test_dynamodb/test_dynamodb.py::test_remove_top_level_attribute_non_existent tests/test_dynamodb/test_dynamodb.py::test_delete_non_existent_backup_raises_error tests/test_dynamodb/test_dynamodb.py::test_update_item_with_attribute_in_right_hand_side tests/test_dynamodb/test_dynamodb.py::test_list_table_tags_paginated