Repository: getmoto/moto
Pull request #7672 (2024-05-05 21:01:03)

DynamoDB: import_table()/describe_import()

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/7672?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
Attention: Patch coverage is `95.91837%` with `4 lines` in your changes are missing coverage. Please review.
> Project coverage is 94.29%. Comparing base [(`8ab84c1`)](https://app.codecov.io/gh/getmoto/moto/commit/8ab84c1990a6ce0ccf57f2b2f36536949104e0c3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`f5eed6e`)](https://app.codecov.io/gh/getmoto/moto/pull/7672?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

| [Files](https://app.codecov.io/gh/getmoto/moto/pull/7672?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Patch % | Lines |
|---|---|---|
| [moto/dynamodb/models/table\_import.py](https://app.codecov.io/gh/getmoto/moto/pull/7672?src=pr&el=tree&filepath=moto%2Fdynamodb%2Fmodels%2Ftable_import.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#diff-bW90by9keW5hbW9kYi9tb2RlbHMvdGFibGVfaW1wb3J0LnB5) | 94.28% | [4 Missing :warning: ](https://app.codecov.io/gh/getmoto/moto/pull/7672?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) |

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master    #7672   +/-   ##
=======================================
  Coverage   94.29%   94.29%           
=======================================
  Files        1086     1087    +1     
  Lines       92294    92392   +98     
=======================================
+ Hits        87030    87123   +93     
- Misses       5264     5269    +5     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/7672/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/7672/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `29.45% <15.30%> (-0.02%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/7672/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.26% <95.91%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/7672?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).


Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 8b4411847..2bd7ccbcb 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -2024,7 +2024,7 @@
 
 ## dynamodb
 <details>
-<summary>56% implemented</summary>
+<summary>59% implemented</summary>
 
 - [X] batch_execute_statement
 - [X] batch_get_item
@@ -2043,7 +2043,7 @@
 - [ ] describe_export
 - [ ] describe_global_table
 - [ ] describe_global_table_settings
-- [ ] describe_import
+- [X] describe_import
 - [ ] describe_kinesis_streaming_destination
 - [ ] describe_limits
 - [X] describe_table
@@ -2056,7 +2056,7 @@
 - [ ] export_table_to_point_in_time
 - [X] get_item
 - [ ] get_resource_policy
-- [ ] import_table
+- [X] import_table
 - [X] list_backups
 - [ ] list_contributor_insights
 - [ ] list_exports
diff --git a/docs/docs/services/dynamodb.rst b/docs/docs/services/dynamodb.rst
index 10c9c57da..872802d6f 100644
--- a/docs/docs/services/dynamodb.rst
+++ b/docs/docs/services/dynamodb.rst
@@ -35,7 +35,7 @@ dynamodb
 - [ ] describe_export
 - [ ] describe_global_table
 - [ ] describe_global_table_settings
-- [ ] describe_import
+- [X] describe_import
 - [ ] describe_kinesis_streaming_destination
 - [ ] describe_limits
 - [X] describe_table
@@ -58,7 +58,13 @@ dynamodb
 - [ ] export_table_to_point_in_time
 - [X] get_item
 - [ ] get_resource_policy
-- [ ] import_table
+- [X] import_table
+  
+        Only InputFormat=DYNAMODB_JSON is supported so far.
+        InputCompressionType=ZSTD is not supported.
+        Other parameters that are not supported: InputFormatOptions, CloudWatchLogGroupArn
+        
+
 - [X] list_backups
 - [ ] list_contributor_insights
 - [ ] list_exports
diff --git a/moto/dynamodb/models/__init__.py b/moto/dynamodb/models/__init__.py
index 4363e12f0..574152f60 100644
--- a/moto/dynamodb/models/__init__.py
+++ b/moto/dynamodb/models/__init__.py
@@ -33,6 +33,7 @@ from moto.dynamodb.models.table import (
     RestoredTable,
     Table,
 )
+from moto.dynamodb.models.table_import import TableImport
 from moto.dynamodb.parsing import partiql
 from moto.dynamodb.parsing.executors import UpdateExpressionExecutor
 from moto.dynamodb.parsing.expressions import UpdateExpressionParser  # type: ignore
@@ -44,6 +45,7 @@ class DynamoDBBackend(BaseBackend):
         super().__init__(region_name, account_id)
         self.tables: Dict[str, Table] = OrderedDict()
         self.backups: Dict[str, Backup] = OrderedDict()
+        self.table_imports: Dict[str, TableImport] = {}
 
     @staticmethod
     def default_vpc_endpoint_service(
@@ -934,5 +936,41 @@ class DynamoDBBackend(BaseBackend):
                 responses[idx] = {"Error": {"Code": e.name, "Message": e.message}}  # type: ignore
         return responses
 
+    def import_table(
+        self,
+        s3_source: Dict[str, str],
+        input_format: Optional[str],
+        compression_type: Optional[str],
+        table_name: str,
+        billing_mode: str,
+        throughput: Optional[Dict[str, int]],
+        key_schema: List[Dict[str, str]],
+        global_indexes: Optional[List[Dict[str, Any]]],
+        attrs: List[Dict[str, str]],
+    ) -> TableImport:
+        """
+        Only InputFormat=DYNAMODB_JSON is supported so far.
+        InputCompressionType=ZSTD is not supported.
+        Other parameters that are not supported: InputFormatOptions, CloudWatchLogGroupArn
+        """
+        table_import = TableImport(
+            account_id=self.account_id,
+            s3_source=s3_source,
+            region_name=self.region_name,
+            table_name=table_name,
+            billing_mode=billing_mode,
+            throughput=throughput,
+            key_schema=key_schema,
+            global_indexes=global_indexes,
+            attrs=attrs,
+            compression_type=compression_type,
+        )
+        self.table_imports[table_import.arn] = table_import
+        table_import.start()
+        return table_import
+
+    def describe_import(self, import_arn: str) -> TableImport:
+        return self.table_imports[import_arn]
+
 
 dynamodb_backends = BackendDict(DynamoDBBackend, "dynamodb")
diff --git a/moto/dynamodb/models/table_import.py b/moto/dynamodb/models/table_import.py
new file mode 100644
index 000000000..22e720312
--- /dev/null
+++ b/moto/dynamodb/models/table_import.py
@@ -0,0 +1,138 @@
+from threading import Thread
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from uuid import uuid4
+
+from moto.core.utils import gzip_decompress
+
+if TYPE_CHECKING:
+    from moto.dynamodb.models import DynamoDBBackend
+    from moto.dynamodb.models.table import Table
+    from moto.s3.models import FakeBucket, S3Backend
+
+
+class TableImport(Thread):
+    def __init__(
+        self,
+        account_id: str,
+        s3_source: Dict[str, str],
+        region_name: str,
+        table_name: str,
+        billing_mode: str,
+        throughput: Optional[Dict[str, int]],
+        key_schema: List[Dict[str, str]],
+        global_indexes: Optional[List[Dict[str, Any]]],
+        attrs: List[Dict[str, str]],
+        compression_type: Optional[str],
+    ):
+        super().__init__()
+        self.arn = f"arn:aws:dynamodb:{region_name}:{account_id}:table/{table_name}/import/{str(uuid4()).replace('-', '')}"
+        self.status = "IN_PROGRESS"
+        self.account_id = account_id
+        self.s3_source = s3_source
+        self.region_name = region_name
+
+        self.table_name = table_name
+        self.billing_mode = billing_mode
+        self.throughput = throughput
+        self.key_schema = key_schema
+        self.global_indexes = global_indexes
+        self.attrs = attrs
+        self.compression_type = compression_type
+
+        self.failure_code: Optional[str] = None
+        self.failure_message: Optional[str] = None
+        self.table: Optional["Table"] = None
+        self.table_arn = (
+            f"arn:aws:dynamodb:{self.region_name}:{self.account_id}:table/{table_name}"
+        )
+
+        self.processed_count = 0
+        self.processed_bytes = 0
+        self.error_count = 0
+        self.imported_count = 0
+
+    def run(self) -> None:
+        s3_bucket_name = self.s3_source["S3Bucket"]
+
+        try:
+            from moto.s3.models import s3_backends
+
+            s3_backend = s3_backends[self.account_id]["global"]
+            bucket = s3_backend.buckets[s3_bucket_name]
+        except KeyError:
+            self.status = "FAILED"
+            self.failure_code = "S3NoSuchBucket"
+            self.failure_message = "The specified bucket does not exist"
+            return
+
+        try:
+            self._process_s3_files(s3_backend, bucket)
+        except Exception as e:
+            self.status = "FAILED"
+            self.failure_code = "UNKNOWN"
+            self.failure_message = str(e)
+
+    def _process_s3_files(self, s3_backend: "S3Backend", bucket: "FakeBucket") -> None:
+        # CREATE TABLE
+        from moto.dynamodb.models import dynamodb_backends
+
+        dynamo: DynamoDBBackend = dynamodb_backends[self.account_id][self.region_name]
+        self.table = dynamo.create_table(
+            name=self.table_name,
+            billing_mode=self.billing_mode,
+            throughput=self.throughput,
+            schema=self.key_schema,
+            global_indexes=self.global_indexes,
+            indexes=None,
+            attr=self.attrs,
+            sse_specification=None,
+            streams=None,
+            tags=[],
+            deletion_protection_enabled=False,
+        )
+
+        # Load data from S3
+        keys, _, _, _ = s3_backend.list_objects(
+            bucket,
+            prefix=self.s3_source.get("S3KeyPrefix"),
+            delimiter=None,
+            marker=None,
+            max_keys=None,
+        )
+
+        from py_partiql_parser._internal.json_parser import JsonParser
+
+        parser = JsonParser()
+
+        for key in keys:
+            if self.compression_type == "GZIP":
+                content = gzip_decompress(key.value).decode("utf-8")
+            else:
+                content = key.value.decode("utf-8")
+            result = parser.parse(original=content)
+            if not isinstance(result, list):
+                result = [result]
+            for json_object in result:
+                try:
+                    self.processed_count += 1
+                    self.processed_bytes += len(json_object)
+                    self.table.put_item(item_attrs=json_object["Item"])
+                    self.imported_count += 1
+                except Exception as e:
+                    self.failure_message = str(e)
+                    self.error_count += 1
+
+        self.status = "COMPLETED" if self.error_count == 0 else "FAILED"
+
+    def response(self) -> Dict[str, Any]:
+        return {
+            "ImportArn": self.arn,
+            "ImportStatus": self.status,
+            "TableArn": self.table_arn,
+            "FailureCode": self.failure_code,
+            "FailureMessage": self.failure_message,
+            "ProcessedItemCount": self.processed_count,
+            "ProcessedSizeBytes": self.processed_bytes,
+            "ErrorCount": self.error_count,
+            "ImportedItemCount": self.imported_count,
+        }
diff --git a/moto/dynamodb/responses.py b/moto/dynamodb/responses.py
index a0d705413..fb14ba753 100644
--- a/moto/dynamodb/responses.py
+++ b/moto/dynamodb/responses.py
@@ -1182,3 +1182,43 @@ class DynamoHandler(BaseResponse):
         stmts = self.body.get("Statements", [])
         items = self.dynamodb_backend.batch_execute_statement(stmts)
         return dynamo_json_dump({"Responses": items})
+
+    def import_table(self) -> str:
+        params = self.body
+        s3_source = params.get("S3BucketSource")
+        input_format = params.get("InputFormat") or "DYNAMODB_JSON"
+        compression_type = params.get("InputCompressionType") or "NONE"
+        table_parameters = params.get("TableCreationParameters")
+        table_name = table_parameters["TableName"]
+        table_attrs = table_parameters["AttributeDefinitions"]
+        table_schema = table_parameters["KeySchema"]
+        table_billing = table_parameters.get("BillingMode")
+        table_throughput = table_parameters.get("ProvisionedThroughput")
+        global_indexes = table_parameters.get("GlobalSecondaryIndexes")
+
+        self._validate_table_creation(
+            billing_mode=table_billing,
+            throughput=table_throughput,
+            key_schema=table_schema,
+            global_indexes=global_indexes,
+            local_secondary_indexes=None,
+            attr=table_attrs,
+        )
+
+        import_table = self.dynamodb_backend.import_table(
+            s3_source=s3_source,
+            input_format=input_format,
+            compression_type=compression_type,
+            table_name=table_name,
+            billing_mode=table_billing or "PROVISIONED",
+            throughput=table_throughput,
+            key_schema=table_schema,
+            global_indexes=global_indexes,
+            attrs=table_attrs,
+        )
+        return json.dumps({"ImportTableDescription": import_table.response()})
+
+    def describe_import(self) -> str:
+        import_arn = self.body["ImportArn"]
+        import_table = self.dynamodb_backend.describe_import(import_arn)
+        return json.dumps({"ImportTableDescription": import_table.response()})


Test patch:
diff --git a/tests/test_dynamodb/test_dynamodb_import_table.py b/tests/test_dynamodb/test_dynamodb_import_table.py
new file mode 100644
index 000000000..69f159f3e
--- /dev/null
+++ b/tests/test_dynamodb/test_dynamodb_import_table.py
@@ -0,0 +1,456 @@
+import gzip
+import json
+from time import sleep
+from uuid import uuid4
+
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+
+from . import dynamodb_aws_verified
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_from_missing_s3_table(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-1")
+
+    table_name = "t" + str(uuid4())[0:6]
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": f"{uuid4()}"},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "FAILED"
+    assert import_details["TableArn"].endswith(f":table/{table_name}")
+    assert import_details["FailureCode"] == "S3NoSuchBucket"
+    assert "The specified bucket does not exist" in import_details["FailureMessage"]
+
+    assert table_name not in client.list_tables()["TableNames"]
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_has_regular_validation(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-1")
+
+    table_name = "t" + str(uuid4())[0:6]
+
+    with pytest.raises(ClientError) as exc:
+        client.import_table(
+            S3BucketSource={"S3Bucket": f"{uuid4()}"},
+            InputFormat="DYNAMODB_JSON",
+            InputCompressionType="NONE",
+            TableCreationParameters={
+                "TableName": table_name,
+                "AttributeDefinitions": [
+                    {"AttributeName": "pk", "AttributeType": "S"},
+                ],
+                "KeySchema": [
+                    {"AttributeName": "pk", "KeyType": "HASH"},
+                ],
+                "BillingMode": "PROVISIONED",
+            },
+        )
+
+    err = exc.value.response["Error"]
+    assert (
+        err["Message"]
+        == "One or more parameter values were invalid: ReadCapacityUnits and WriteCapacityUnits must both be specified when BillingMode is PROVISIONED"
+    )
+
+    assert table_name not in client.list_tables()["TableNames"]
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_from_empty_s3_bucket(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-1")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 0
+    assert import_details["ImportedItemCount"] == 0
+    assert import_details["ProcessedSizeBytes"] == 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert client.scan(TableName=table_name)["Items"] == []
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_table_single_file_with_multiple_items():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    data = ""
+    for i in range(5):
+        data += (
+            json.dumps({"Item": {"pk": {"S": f"msg{i}"}, "data": {"S": f"{uuid4()}"}}})
+            + "\n"
+        )
+    for i in range(10, 15):
+        data += json.dumps(
+            {"Item": {"pk": {"S": f"msg{i}"}, "data": {"S": f"{uuid4()}"}}}
+        )
+    filename1 = "data.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=data,
+        Key=filename1,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 10
+    assert import_details["ImportedItemCount"] == 10
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+    expected = [
+        "msg0",
+        "msg1",
+        "msg10",
+        "msg11",
+        "msg12",
+        "msg13",
+        "msg14",
+        "msg2",
+        "msg3",
+        "msg4",
+    ]
+    assert (
+        sorted([i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]])
+        == expected
+    )
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_table_multiple_files():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    items_file1 = {"Item": {"pk": {"S": "msg1"}, "data": {"S": f"{uuid4()}"}}}
+    filename1 = "data.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file1),
+        Key=filename1,
+    )
+
+    items_file2 = {"Item": {"pk": {"S": "msg2"}, "data": {"S": f"{uuid4()}"}}}
+    filename2 = "completely_random_filename_without_extension"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file2),
+        Key=filename2,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 2
+    assert import_details["ImportedItemCount"] == 2
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert [i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]] == [
+        "msg1",
+        "msg2",
+    ]
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename2)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_some_successfull_files_and_some_with_unknown_data():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    items_file1 = {"Item": {"pk": {"S": "msg1"}, "data": {"S": f"{uuid4()}"}}}
+    filename1 = "data.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file1),
+        Key=filename1,
+    )
+
+    items_file2 = {"pk": {"S": "msg2"}, "data": {"S": f"{uuid4()}"}}
+    filename2 = "invaliddata"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file2),
+        Key=filename2,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "FAILED"
+    assert import_details["ErrorCount"] == 1
+    assert import_details["ProcessedItemCount"] == 2
+    assert import_details["ImportedItemCount"] == 1
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert [i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]] == [
+        "msg1"
+    ]
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename2)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_only_process_file_with_prefix():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    items_file1 = {"Item": {"pk": {"S": "msg1"}, "data": {"S": f"{uuid4()}"}}}
+    filename1 = "yesdata.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file1),
+        Key=filename1,
+    )
+
+    items_file2 = {"Item": {"pk": {"S": "msg2"}, "data": {"S": f"{uuid4()}"}}}
+    filename2 = "nodata.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file2),
+        Key=filename2,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name, "S3KeyPrefix": "yes"},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 1
+    assert import_details["ImportedItemCount"] == 1
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert [i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]] == [
+        "msg1"
+    ]
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename2)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_process_gzipped_file():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    items_file1 = {"Item": {"pk": {"S": "msg1"}, "data": {"S": f"{uuid4()}"}}}
+    filename1 = "data.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=gzip.compress(json.dumps(items_file1).encode("utf-8")),
+        Key=filename1,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="GZIP",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 1
+    assert import_details["ImportedItemCount"] == 1
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert [i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]] == [
+        "msg1"
+    ]
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+def wait_for_import(client, import_description):
+    import_details = client.describe_import(ImportArn=import_description["ImportArn"])
+    status = import_details["ImportTableDescription"]["ImportStatus"]
+    while status == "IN_PROGRESS":
+        sleep(0.1)
+        import_details = client.describe_import(
+            ImportArn=import_description["ImportArn"]
+        )
+        status = import_details["ImportTableDescription"]["ImportStatus"]
+    return import_details["ImportTableDescription"]