We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[Docs] Explain DynamoDB online_write_batch uses a batch_writer
## Expected Behavior 

`DynamoDBOnlineStore` method `online_write_batch` uses a [BatchWriter](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/dynamodb.html#batch-writing) under the hood. 

User documentation should explain the advantages of using it, including 

> automatically handle buffering and sending items in batches. In addition, the batch writer will also automatically handle any unprocessed items and resend them as needed. 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/feast/infra/online_stores/dynamodb.py`
Content:
```
1 # Copyright 2021 The Feast Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     https://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 import logging
15 from datetime import datetime
16 from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
17 
18 from pydantic import StrictStr
19 from pydantic.typing import Literal
20 
21 from feast import Entity, FeatureView, utils
22 from feast.infra.infra_object import DYNAMODB_INFRA_OBJECT_CLASS_TYPE, InfraObject
23 from feast.infra.online_stores.helpers import compute_entity_id
24 from feast.infra.online_stores.online_store import OnlineStore
25 from feast.protos.feast.core.DynamoDBTable_pb2 import (
26     DynamoDBTable as DynamoDBTableProto,
27 )
28 from feast.protos.feast.core.InfraObject_pb2 import InfraObject as InfraObjectProto
29 from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
30 from feast.protos.feast.types.Value_pb2 import Value as ValueProto
31 from feast.repo_config import FeastConfigBaseModel, RepoConfig
32 from feast.usage import log_exceptions_and_usage, tracing_span
33 
34 try:
35     import boto3
36     from botocore.exceptions import ClientError
37 except ImportError as e:
38     from feast.errors import FeastExtrasDependencyImportError
39 
40     raise FeastExtrasDependencyImportError("aws", str(e))
41 
42 
43 logger = logging.getLogger(__name__)
44 
45 
46 class DynamoDBOnlineStoreConfig(FeastConfigBaseModel):
47     """Online store config for DynamoDB store"""
48 
49     type: Literal["dynamodb"] = "dynamodb"
50     """Online store type selector"""
51 
52     region: StrictStr
53     """ AWS Region Name """
54 
55 
56 class DynamoDBOnlineStore(OnlineStore):
57     """
58     Online feature store for AWS DynamoDB.
59     """
60 
61     _dynamodb_client = None
62     _dynamodb_resource = None
63 
64     @log_exceptions_and_usage(online_store="dynamodb")
65     def update(
66         self,
67         config: RepoConfig,
68         tables_to_delete: Sequence[FeatureView],
69         tables_to_keep: Sequence[FeatureView],
70         entities_to_delete: Sequence[Entity],
71         entities_to_keep: Sequence[Entity],
72         partial: bool,
73     ):
74         online_config = config.online_store
75         assert isinstance(online_config, DynamoDBOnlineStoreConfig)
76         dynamodb_client = self._get_dynamodb_client(online_config.region)
77         dynamodb_resource = self._get_dynamodb_resource(online_config.region)
78 
79         for table_instance in tables_to_keep:
80             try:
81                 dynamodb_resource.create_table(
82                     TableName=_get_table_name(config, table_instance),
83                     KeySchema=[{"AttributeName": "entity_id", "KeyType": "HASH"}],
84                     AttributeDefinitions=[
85                         {"AttributeName": "entity_id", "AttributeType": "S"}
86                     ],
87                     BillingMode="PAY_PER_REQUEST",
88                 )
89             except ClientError as ce:
90                 # If the table creation fails with ResourceInUseException,
91                 # it means the table already exists or is being created.
92                 # Otherwise, re-raise the exception
93                 if ce.response["Error"]["Code"] != "ResourceInUseException":
94                     raise
95 
96         for table_instance in tables_to_keep:
97             dynamodb_client.get_waiter("table_exists").wait(
98                 TableName=_get_table_name(config, table_instance)
99             )
100 
101         for table_to_delete in tables_to_delete:
102             _delete_table_idempotent(
103                 dynamodb_resource, _get_table_name(config, table_to_delete)
104             )
105 
106     def teardown(
107         self,
108         config: RepoConfig,
109         tables: Sequence[FeatureView],
110         entities: Sequence[Entity],
111     ):
112         online_config = config.online_store
113         assert isinstance(online_config, DynamoDBOnlineStoreConfig)
114         dynamodb_resource = self._get_dynamodb_resource(online_config.region)
115 
116         for table in tables:
117             _delete_table_idempotent(dynamodb_resource, _get_table_name(config, table))
118 
119     @log_exceptions_and_usage(online_store="dynamodb")
120     def online_write_batch(
121         self,
122         config: RepoConfig,
123         table: FeatureView,
124         data: List[
125             Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]
126         ],
127         progress: Optional[Callable[[int], Any]],
128     ) -> None:
129         online_config = config.online_store
130         assert isinstance(online_config, DynamoDBOnlineStoreConfig)
131         dynamodb_resource = self._get_dynamodb_resource(online_config.region)
132 
133         table_instance = dynamodb_resource.Table(_get_table_name(config, table))
134         with table_instance.batch_writer() as batch:
135             for entity_key, features, timestamp, created_ts in data:
136                 entity_id = compute_entity_id(entity_key)
137                 batch.put_item(
138                     Item={
139                         "entity_id": entity_id,  # PartitionKey
140                         "event_ts": str(utils.make_tzaware(timestamp)),
141                         "values": {
142                             k: v.SerializeToString()
143                             for k, v in features.items()  # Serialized Features
144                         },
145                     }
146                 )
147                 if progress:
148                     progress(1)
149 
150     @log_exceptions_and_usage(online_store="dynamodb")
151     def online_read(
152         self,
153         config: RepoConfig,
154         table: FeatureView,
155         entity_keys: List[EntityKeyProto],
156         requested_features: Optional[List[str]] = None,
157     ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
158         online_config = config.online_store
159         assert isinstance(online_config, DynamoDBOnlineStoreConfig)
160         dynamodb_resource = self._get_dynamodb_resource(online_config.region)
161 
162         result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []
163         for entity_key in entity_keys:
164             table_instance = dynamodb_resource.Table(_get_table_name(config, table))
165             entity_id = compute_entity_id(entity_key)
166             with tracing_span(name="remote_call"):
167                 response = table_instance.get_item(Key={"entity_id": entity_id})
168             value = response.get("Item")
169 
170             if value is not None:
171                 res = {}
172                 for feature_name, value_bin in value["values"].items():
173                     val = ValueProto()
174                     val.ParseFromString(value_bin.value)
175                     res[feature_name] = val
176                 result.append((datetime.fromisoformat(value["event_ts"]), res))
177             else:
178                 result.append((None, None))
179         return result
180 
181     def _get_dynamodb_client(self, region: str):
182         if self._dynamodb_client is None:
183             self._dynamodb_client = _initialize_dynamodb_client(region)
184         return self._dynamodb_client
185 
186     def _get_dynamodb_resource(self, region: str):
187         if self._dynamodb_resource is None:
188             self._dynamodb_resource = _initialize_dynamodb_resource(region)
189         return self._dynamodb_resource
190 
191 
192 def _initialize_dynamodb_client(region: str):
193     return boto3.client("dynamodb", region_name=region)
194 
195 
196 def _initialize_dynamodb_resource(region: str):
197     return boto3.resource("dynamodb", region_name=region)
198 
199 
200 def _get_table_name(config: RepoConfig, table: FeatureView) -> str:
201     return f"{config.project}.{table.name}"
202 
203 
204 def _delete_table_idempotent(
205     dynamodb_resource, table_name: str,
206 ):
207     try:
208         table = dynamodb_resource.Table(table_name)
209         table.delete()
210         logger.info(f"Dynamo table {table_name} was deleted")
211     except ClientError as ce:
212         # If the table deletion fails with ResourceNotFoundException,
213         # it means the table has already been deleted.
214         # Otherwise, re-raise the exception
215         if ce.response["Error"]["Code"] != "ResourceNotFoundException":
216             raise
217         else:
218             logger.warning(f"Trying to delete table that doesn't exist: {table_name}")
219 
220 
221 class DynamoDBTable(InfraObject):
222     """
223     A DynamoDB table managed by Feast.
224 
225     Attributes:
226         name: The name of the table.
227         region: The region of the table.
228     """
229 
230     region: str
231 
232     def __init__(self, name: str, region: str):
233         super().__init__(name)
234         self.region = region
235 
236     def to_infra_object_proto(self) -> InfraObjectProto:
237         dynamodb_table_proto = self.to_proto()
238         return InfraObjectProto(
239             infra_object_class_type=DYNAMODB_INFRA_OBJECT_CLASS_TYPE,
240             dynamodb_table=dynamodb_table_proto,
241         )
242 
243     def to_proto(self) -> Any:
244         dynamodb_table_proto = DynamoDBTableProto()
245         dynamodb_table_proto.name = self.name
246         dynamodb_table_proto.region = self.region
247         return dynamodb_table_proto
248 
249     @staticmethod
250     def from_infra_object_proto(infra_object_proto: InfraObjectProto) -> Any:
251         return DynamoDBTable(
252             name=infra_object_proto.dynamodb_table.name,
253             region=infra_object_proto.dynamodb_table.region,
254         )
255 
256     @staticmethod
257     def from_proto(dynamodb_table_proto: DynamoDBTableProto) -> Any:
258         return DynamoDBTable(
259             name=dynamodb_table_proto.name, region=dynamodb_table_proto.region,
260         )
261 
262     def update(self):
263         dynamodb_client = _initialize_dynamodb_client(region=self.region)
264         dynamodb_resource = _initialize_dynamodb_resource(region=self.region)
265 
266         try:
267             dynamodb_resource.create_table(
268                 TableName=f"{self.name}",
269                 KeySchema=[{"AttributeName": "entity_id", "KeyType": "HASH"}],
270                 AttributeDefinitions=[
271                     {"AttributeName": "entity_id", "AttributeType": "S"}
272                 ],
273                 BillingMode="PAY_PER_REQUEST",
274             )
275         except ClientError as ce:
276             # If the table creation fails with ResourceInUseException,
277             # it means the table already exists or is being created.
278             # Otherwise, re-raise the exception
279             if ce.response["Error"]["Code"] != "ResourceInUseException":
280                 raise
281 
282         dynamodb_client.get_waiter("table_exists").wait(TableName=f"{self.name}")
283 
284     def teardown(self):
285         dynamodb_resource = _initialize_dynamodb_resource(region=self.region)
286         _delete_table_idempotent(dynamodb_resource, self.name)
287
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/feast/infra/online_stores/dynamodb.py b/sdk/python/feast/infra/online_stores/dynamodb.py
--- a/sdk/python/feast/infra/online_stores/dynamodb.py
+++ b/sdk/python/feast/infra/online_stores/dynamodb.py
@@ -56,6 +56,10 @@
 class DynamoDBOnlineStore(OnlineStore):
     """
     Online feature store for AWS DynamoDB.
+
+    Attributes:
+        _dynamodb_client: Boto3 DynamoDB client.
+        _dynamodb_resource: Boto3 DynamoDB resource.
     """
 
     _dynamodb_client = None
@@ -71,6 +75,14 @@
         entities_to_keep: Sequence[Entity],
         partial: bool,
     ):
+        """
+        Update tables from the DynamoDB Online Store.
+
+        Args:
+            config: The RepoConfig for the current FeatureStore.
+            tables_to_delete: Tables to delete from the DynamoDB Online Store.
+            tables_to_keep: Tables to keep in the DynamoDB Online Store.
+        """
         online_config = config.online_store
         assert isinstance(online_config, DynamoDBOnlineStoreConfig)
         dynamodb_client = self._get_dynamodb_client(online_config.region)
@@ -109,6 +121,13 @@
         tables: Sequence[FeatureView],
         entities: Sequence[Entity],
     ):
+        """
+        Delete tables from the DynamoDB Online Store.
+
+        Args:
+            config: The RepoConfig for the current FeatureStore.
+            tables: Tables to delete from the feature repo.
+        """
         online_config = config.online_store
         assert isinstance(online_config, DynamoDBOnlineStoreConfig)
         dynamodb_resource = self._get_dynamodb_resource(online_config.region)
@@ -126,6 +145,21 @@
         ],
         progress: Optional[Callable[[int], Any]],
     ) -> None:
+        """
+        Write a batch of feature rows to online DynamoDB store.
+
+        Note: This method applies a ``batch_writer`` to automatically handle any unprocessed items
+        and resend them as needed, this is useful if you're loading a lot of data at a time.
+
+        Args:
+            config: The RepoConfig for the current FeatureStore.
+            table: Feast FeatureView.
+            data: a list of quadruplets containing Feature data. Each quadruplet contains an Entity Key,
+            a dict containing feature values, an event timestamp for the row, and
+            the created timestamp for the row if it exists.
+            progress: Optional function to be called once every mini-batch of rows is written to
+            the online store. Can be used to display progress.
+        """
         online_config = config.online_store
         assert isinstance(online_config, DynamoDBOnlineStoreConfig)
         dynamodb_resource = self._get_dynamodb_resource(online_config.region)
@@ -155,6 +189,17 @@
         entity_keys: List[EntityKeyProto],
         requested_features: Optional[List[str]] = None,
     ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
+        """
+        Retrieve feature values from the online DynamoDB store.
+
+        Note: This method is currently not optimized to retrieve a lot of data at a time
+        as it does sequential gets from the DynamoDB table.
+
+        Args:
+            config: The RepoConfig for the current FeatureStore.
+            table: Feast FeatureView.
+            entity_keys: a list of entity keys that should be read from the FeatureStore.
+        """
         online_config = config.online_store
         assert isinstance(online_config, DynamoDBOnlineStoreConfig)
         dynamodb_resource = self._get_dynamodb_resource(online_config.region)

{"golden_diff": "diff --git a/sdk/python/feast/infra/online_stores/dynamodb.py b/sdk/python/feast/infra/online_stores/dynamodb.py\n--- a/sdk/python/feast/infra/online_stores/dynamodb.py\n+++ b/sdk/python/feast/infra/online_stores/dynamodb.py\n@@ -56,6 +56,10 @@\n class DynamoDBOnlineStore(OnlineStore):\n     \"\"\"\n     Online feature store for AWS DynamoDB.\n+\n+    Attributes:\n+        _dynamodb_client: Boto3 DynamoDB client.\n+        _dynamodb_resource: Boto3 DynamoDB resource.\n     \"\"\"\n \n     _dynamodb_client = None\n@@ -71,6 +75,14 @@\n         entities_to_keep: Sequence[Entity],\n         partial: bool,\n     ):\n+        \"\"\"\n+        Update tables from the DynamoDB Online Store.\n+\n+        Args:\n+            config: The RepoConfig for the current FeatureStore.\n+            tables_to_delete: Tables to delete from the DynamoDB Online Store.\n+            tables_to_keep: Tables to keep in the DynamoDB Online Store.\n+        \"\"\"\n         online_config = config.online_store\n         assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n         dynamodb_client = self._get_dynamodb_client(online_config.region)\n@@ -109,6 +121,13 @@\n         tables: Sequence[FeatureView],\n         entities: Sequence[Entity],\n     ):\n+        \"\"\"\n+        Delete tables from the DynamoDB Online Store.\n+\n+        Args:\n+            config: The RepoConfig for the current FeatureStore.\n+            tables: Tables to delete from the feature repo.\n+        \"\"\"\n         online_config = config.online_store\n         assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n         dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n@@ -126,6 +145,21 @@\n         ],\n         progress: Optional[Callable[[int], Any]],\n     ) -> None:\n+        \"\"\"\n+        Write a batch of feature rows to online DynamoDB store.\n+\n+        Note: This method applies a ``batch_writer`` to automatically handle any unprocessed items\n+        and resend them as needed, this is useful if you're loading a lot of data at a time.\n+\n+        Args:\n+            config: The RepoConfig for the current FeatureStore.\n+            table: Feast FeatureView.\n+            data: a list of quadruplets containing Feature data. Each quadruplet contains an Entity Key,\n+            a dict containing feature values, an event timestamp for the row, and\n+            the created timestamp for the row if it exists.\n+            progress: Optional function to be called once every mini-batch of rows is written to\n+            the online store. Can be used to display progress.\n+        \"\"\"\n         online_config = config.online_store\n         assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n         dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n@@ -155,6 +189,17 @@\n         entity_keys: List[EntityKeyProto],\n         requested_features: Optional[List[str]] = None,\n     ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:\n+        \"\"\"\n+        Retrieve feature values from the online DynamoDB store.\n+\n+        Note: This method is currently not optimized to retrieve a lot of data at a time\n+        as it does sequential gets from the DynamoDB table.\n+\n+        Args:\n+            config: The RepoConfig for the current FeatureStore.\n+            table: Feast FeatureView.\n+            entity_keys: a list of entity keys that should be read from the FeatureStore.\n+        \"\"\"\n         online_config = config.online_store\n         assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n         dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n", "issue": "[Docs] Explain DynamoDB online_write_batch uses a batch_writer\n## Expected Behavior \r\n\r\n`DynamoDBOnlineStore` method `online_write_batch` uses a [BatchWriter](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/dynamodb.html#batch-writing) under the hood. \r\n\r\nUser documentation should explain the advantages of using it, including \r\n\r\n> automatically handle buffering and sending items in batches. In addition, the batch writer will also automatically handle any unprocessed items and resend them as needed. \n", "before_files": [{"content": "# Copyright 2021 The Feast Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     https://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport logging\nfrom datetime import datetime\nfrom typing import Any, Callable, Dict, List, Optional, Sequence, Tuple\n\nfrom pydantic import StrictStr\nfrom pydantic.typing import Literal\n\nfrom feast import Entity, FeatureView, utils\nfrom feast.infra.infra_object import DYNAMODB_INFRA_OBJECT_CLASS_TYPE, InfraObject\nfrom feast.infra.online_stores.helpers import compute_entity_id\nfrom feast.infra.online_stores.online_store import OnlineStore\nfrom feast.protos.feast.core.DynamoDBTable_pb2 import (\n    DynamoDBTable as DynamoDBTableProto,\n)\nfrom feast.protos.feast.core.InfraObject_pb2 import InfraObject as InfraObjectProto\nfrom feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto\nfrom feast.protos.feast.types.Value_pb2 import Value as ValueProto\nfrom feast.repo_config import FeastConfigBaseModel, RepoConfig\nfrom feast.usage import log_exceptions_and_usage, tracing_span\n\ntry:\n    import boto3\n    from botocore.exceptions import ClientError\nexcept ImportError as e:\n    from feast.errors import FeastExtrasDependencyImportError\n\n    raise FeastExtrasDependencyImportError(\"aws\", str(e))\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass DynamoDBOnlineStoreConfig(FeastConfigBaseModel):\n    \"\"\"Online store config for DynamoDB store\"\"\"\n\n    type: Literal[\"dynamodb\"] = \"dynamodb\"\n    \"\"\"Online store type selector\"\"\"\n\n    region: StrictStr\n    \"\"\" AWS Region Name \"\"\"\n\n\nclass DynamoDBOnlineStore(OnlineStore):\n    \"\"\"\n    Online feature store for AWS DynamoDB.\n    \"\"\"\n\n    _dynamodb_client = None\n    _dynamodb_resource = None\n\n    @log_exceptions_and_usage(online_store=\"dynamodb\")\n    def update(\n        self,\n        config: RepoConfig,\n        tables_to_delete: Sequence[FeatureView],\n        tables_to_keep: Sequence[FeatureView],\n        entities_to_delete: Sequence[Entity],\n        entities_to_keep: Sequence[Entity],\n        partial: bool,\n    ):\n        online_config = config.online_store\n        assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n        dynamodb_client = self._get_dynamodb_client(online_config.region)\n        dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n\n        for table_instance in tables_to_keep:\n            try:\n                dynamodb_resource.create_table(\n                    TableName=_get_table_name(config, table_instance),\n                    KeySchema=[{\"AttributeName\": \"entity_id\", \"KeyType\": \"HASH\"}],\n                    AttributeDefinitions=[\n                        {\"AttributeName\": \"entity_id\", \"AttributeType\": \"S\"}\n                    ],\n                    BillingMode=\"PAY_PER_REQUEST\",\n                )\n            except ClientError as ce:\n                # If the table creation fails with ResourceInUseException,\n                # it means the table already exists or is being created.\n                # Otherwise, re-raise the exception\n                if ce.response[\"Error\"][\"Code\"] != \"ResourceInUseException\":\n                    raise\n\n        for table_instance in tables_to_keep:\n            dynamodb_client.get_waiter(\"table_exists\").wait(\n                TableName=_get_table_name(config, table_instance)\n            )\n\n        for table_to_delete in tables_to_delete:\n            _delete_table_idempotent(\n                dynamodb_resource, _get_table_name(config, table_to_delete)\n            )\n\n    def teardown(\n        self,\n        config: RepoConfig,\n        tables: Sequence[FeatureView],\n        entities: Sequence[Entity],\n    ):\n        online_config = config.online_store\n        assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n        dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n\n        for table in tables:\n            _delete_table_idempotent(dynamodb_resource, _get_table_name(config, table))\n\n    @log_exceptions_and_usage(online_store=\"dynamodb\")\n    def online_write_batch(\n        self,\n        config: RepoConfig,\n        table: FeatureView,\n        data: List[\n            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n        ],\n        progress: Optional[Callable[[int], Any]],\n    ) -> None:\n        online_config = config.online_store\n        assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n        dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n\n        table_instance = dynamodb_resource.Table(_get_table_name(config, table))\n        with table_instance.batch_writer() as batch:\n            for entity_key, features, timestamp, created_ts in data:\n                entity_id = compute_entity_id(entity_key)\n                batch.put_item(\n                    Item={\n                        \"entity_id\": entity_id,  # PartitionKey\n                        \"event_ts\": str(utils.make_tzaware(timestamp)),\n                        \"values\": {\n                            k: v.SerializeToString()\n                            for k, v in features.items()  # Serialized Features\n                        },\n                    }\n                )\n                if progress:\n                    progress(1)\n\n    @log_exceptions_and_usage(online_store=\"dynamodb\")\n    def online_read(\n        self,\n        config: RepoConfig,\n        table: FeatureView,\n        entity_keys: List[EntityKeyProto],\n        requested_features: Optional[List[str]] = None,\n    ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:\n        online_config = config.online_store\n        assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n        dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n\n        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n        for entity_key in entity_keys:\n            table_instance = dynamodb_resource.Table(_get_table_name(config, table))\n            entity_id = compute_entity_id(entity_key)\n            with tracing_span(name=\"remote_call\"):\n                response = table_instance.get_item(Key={\"entity_id\": entity_id})\n            value = response.get(\"Item\")\n\n            if value is not None:\n                res = {}\n                for feature_name, value_bin in value[\"values\"].items():\n                    val = ValueProto()\n                    val.ParseFromString(value_bin.value)\n                    res[feature_name] = val\n                result.append((datetime.fromisoformat(value[\"event_ts\"]), res))\n            else:\n                result.append((None, None))\n        return result\n\n    def _get_dynamodb_client(self, region: str):\n        if self._dynamodb_client is None:\n            self._dynamodb_client = _initialize_dynamodb_client(region)\n        return self._dynamodb_client\n\n    def _get_dynamodb_resource(self, region: str):\n        if self._dynamodb_resource is None:\n            self._dynamodb_resource = _initialize_dynamodb_resource(region)\n        return self._dynamodb_resource\n\n\ndef _initialize_dynamodb_client(region: str):\n    return boto3.client(\"dynamodb\", region_name=region)\n\n\ndef _initialize_dynamodb_resource(region: str):\n    return boto3.resource(\"dynamodb\", region_name=region)\n\n\ndef _get_table_name(config: RepoConfig, table: FeatureView) -> str:\n    return f\"{config.project}.{table.name}\"\n\n\ndef _delete_table_idempotent(\n    dynamodb_resource, table_name: str,\n):\n    try:\n        table = dynamodb_resource.Table(table_name)\n        table.delete()\n        logger.info(f\"Dynamo table {table_name} was deleted\")\n    except ClientError as ce:\n        # If the table deletion fails with ResourceNotFoundException,\n        # it means the table has already been deleted.\n        # Otherwise, re-raise the exception\n        if ce.response[\"Error\"][\"Code\"] != \"ResourceNotFoundException\":\n            raise\n        else:\n            logger.warning(f\"Trying to delete table that doesn't exist: {table_name}\")\n\n\nclass DynamoDBTable(InfraObject):\n    \"\"\"\n    A DynamoDB table managed by Feast.\n\n    Attributes:\n        name: The name of the table.\n        region: The region of the table.\n    \"\"\"\n\n    region: str\n\n    def __init__(self, name: str, region: str):\n        super().__init__(name)\n        self.region = region\n\n    def to_infra_object_proto(self) -> InfraObjectProto:\n        dynamodb_table_proto = self.to_proto()\n        return InfraObjectProto(\n            infra_object_class_type=DYNAMODB_INFRA_OBJECT_CLASS_TYPE,\n            dynamodb_table=dynamodb_table_proto,\n        )\n\n    def to_proto(self) -> Any:\n        dynamodb_table_proto = DynamoDBTableProto()\n        dynamodb_table_proto.name = self.name\n        dynamodb_table_proto.region = self.region\n        return dynamodb_table_proto\n\n    @staticmethod\n    def from_infra_object_proto(infra_object_proto: InfraObjectProto) -> Any:\n        return DynamoDBTable(\n            name=infra_object_proto.dynamodb_table.name,\n            region=infra_object_proto.dynamodb_table.region,\n        )\n\n    @staticmethod\n    def from_proto(dynamodb_table_proto: DynamoDBTableProto) -> Any:\n        return DynamoDBTable(\n            name=dynamodb_table_proto.name, region=dynamodb_table_proto.region,\n        )\n\n    def update(self):\n        dynamodb_client = _initialize_dynamodb_client(region=self.region)\n        dynamodb_resource = _initialize_dynamodb_resource(region=self.region)\n\n        try:\n            dynamodb_resource.create_table(\n                TableName=f\"{self.name}\",\n                KeySchema=[{\"AttributeName\": \"entity_id\", \"KeyType\": \"HASH\"}],\n                AttributeDefinitions=[\n                    {\"AttributeName\": \"entity_id\", \"AttributeType\": \"S\"}\n                ],\n                BillingMode=\"PAY_PER_REQUEST\",\n            )\n        except ClientError as ce:\n            # If the table creation fails with ResourceInUseException,\n            # it means the table already exists or is being created.\n            # Otherwise, re-raise the exception\n            if ce.response[\"Error\"][\"Code\"] != \"ResourceInUseException\":\n                raise\n\n        dynamodb_client.get_waiter(\"table_exists\").wait(TableName=f\"{self.name}\")\n\n    def teardown(self):\n        dynamodb_resource = _initialize_dynamodb_resource(region=self.region)\n        _delete_table_idempotent(dynamodb_resource, self.name)\n", "path": "sdk/python/feast/infra/online_stores/dynamodb.py"}], "after_files": [{"content": "# Copyright 2021 The Feast Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     https://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport logging\nfrom datetime import datetime\nfrom typing import Any, Callable, Dict, List, Optional, Sequence, Tuple\n\nfrom pydantic import StrictStr\nfrom pydantic.typing import Literal\n\nfrom feast import Entity, FeatureView, utils\nfrom feast.infra.infra_object import DYNAMODB_INFRA_OBJECT_CLASS_TYPE, InfraObject\nfrom feast.infra.online_stores.helpers import compute_entity_id\nfrom feast.infra.online_stores.online_store import OnlineStore\nfrom feast.protos.feast.core.DynamoDBTable_pb2 import (\n    DynamoDBTable as DynamoDBTableProto,\n)\nfrom feast.protos.feast.core.InfraObject_pb2 import InfraObject as InfraObjectProto\nfrom feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto\nfrom feast.protos.feast.types.Value_pb2 import Value as ValueProto\nfrom feast.repo_config import FeastConfigBaseModel, RepoConfig\nfrom feast.usage import log_exceptions_and_usage, tracing_span\n\ntry:\n    import boto3\n    from botocore.exceptions import ClientError\nexcept ImportError as e:\n    from feast.errors import FeastExtrasDependencyImportError\n\n    raise FeastExtrasDependencyImportError(\"aws\", str(e))\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass DynamoDBOnlineStoreConfig(FeastConfigBaseModel):\n    \"\"\"Online store config for DynamoDB store\"\"\"\n\n    type: Literal[\"dynamodb\"] = \"dynamodb\"\n    \"\"\"Online store type selector\"\"\"\n\n    region: StrictStr\n    \"\"\" AWS Region Name \"\"\"\n\n\nclass DynamoDBOnlineStore(OnlineStore):\n    \"\"\"\n    Online feature store for AWS DynamoDB.\n\n    Attributes:\n        _dynamodb_client: Boto3 DynamoDB client.\n        _dynamodb_resource: Boto3 DynamoDB resource.\n    \"\"\"\n\n    _dynamodb_client = None\n    _dynamodb_resource = None\n\n    @log_exceptions_and_usage(online_store=\"dynamodb\")\n    def update(\n        self,\n        config: RepoConfig,\n        tables_to_delete: Sequence[FeatureView],\n        tables_to_keep: Sequence[FeatureView],\n        entities_to_delete: Sequence[Entity],\n        entities_to_keep: Sequence[Entity],\n        partial: bool,\n    ):\n        \"\"\"\n        Update tables from the DynamoDB Online Store.\n\n        Args:\n            config: The RepoConfig for the current FeatureStore.\n            tables_to_delete: Tables to delete from the DynamoDB Online Store.\n            tables_to_keep: Tables to keep in the DynamoDB Online Store.\n        \"\"\"\n        online_config = config.online_store\n        assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n        dynamodb_client = self._get_dynamodb_client(online_config.region)\n        dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n\n        for table_instance in tables_to_keep:\n            try:\n                dynamodb_resource.create_table(\n                    TableName=_get_table_name(config, table_instance),\n                    KeySchema=[{\"AttributeName\": \"entity_id\", \"KeyType\": \"HASH\"}],\n                    AttributeDefinitions=[\n                        {\"AttributeName\": \"entity_id\", \"AttributeType\": \"S\"}\n                    ],\n                    BillingMode=\"PAY_PER_REQUEST\",\n                )\n            except ClientError as ce:\n                # If the table creation fails with ResourceInUseException,\n                # it means the table already exists or is being created.\n                # Otherwise, re-raise the exception\n                if ce.response[\"Error\"][\"Code\"] != \"ResourceInUseException\":\n                    raise\n\n        for table_instance in tables_to_keep:\n            dynamodb_client.get_waiter(\"table_exists\").wait(\n                TableName=_get_table_name(config, table_instance)\n            )\n\n        for table_to_delete in tables_to_delete:\n            _delete_table_idempotent(\n                dynamodb_resource, _get_table_name(config, table_to_delete)\n            )\n\n    def teardown(\n        self,\n        config: RepoConfig,\n        tables: Sequence[FeatureView],\n        entities: Sequence[Entity],\n    ):\n        \"\"\"\n        Delete tables from the DynamoDB Online Store.\n\n        Args:\n            config: The RepoConfig for the current FeatureStore.\n            tables: Tables to delete from the feature repo.\n        \"\"\"\n        online_config = config.online_store\n        assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n        dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n\n        for table in tables:\n            _delete_table_idempotent(dynamodb_resource, _get_table_name(config, table))\n\n    @log_exceptions_and_usage(online_store=\"dynamodb\")\n    def online_write_batch(\n        self,\n        config: RepoConfig,\n        table: FeatureView,\n        data: List[\n            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n        ],\n        progress: Optional[Callable[[int], Any]],\n    ) -> None:\n        \"\"\"\n        Write a batch of feature rows to online DynamoDB store.\n\n        Note: This method applies a ``batch_writer`` to automatically handle any unprocessed items\n        and resend them as needed, this is useful if you're loading a lot of data at a time.\n\n        Args:\n            config: The RepoConfig for the current FeatureStore.\n            table: Feast FeatureView.\n            data: a list of quadruplets containing Feature data. Each quadruplet contains an Entity Key,\n            a dict containing feature values, an event timestamp for the row, and\n            the created timestamp for the row if it exists.\n            progress: Optional function to be called once every mini-batch of rows is written to\n            the online store. Can be used to display progress.\n        \"\"\"\n        online_config = config.online_store\n        assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n        dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n\n        table_instance = dynamodb_resource.Table(_get_table_name(config, table))\n        with table_instance.batch_writer() as batch:\n            for entity_key, features, timestamp, created_ts in data:\n                entity_id = compute_entity_id(entity_key)\n                batch.put_item(\n                    Item={\n                        \"entity_id\": entity_id,  # PartitionKey\n                        \"event_ts\": str(utils.make_tzaware(timestamp)),\n                        \"values\": {\n                            k: v.SerializeToString()\n                            for k, v in features.items()  # Serialized Features\n                        },\n                    }\n                )\n                if progress:\n                    progress(1)\n\n    @log_exceptions_and_usage(online_store=\"dynamodb\")\n    def online_read(\n        self,\n        config: RepoConfig,\n        table: FeatureView,\n        entity_keys: List[EntityKeyProto],\n        requested_features: Optional[List[str]] = None,\n    ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:\n        \"\"\"\n        Retrieve feature values from the online DynamoDB store.\n\n        Note: This method is currently not optimized to retrieve a lot of data at a time\n        as it does sequential gets from the DynamoDB table.\n\n        Args:\n            config: The RepoConfig for the current FeatureStore.\n            table: Feast FeatureView.\n            entity_keys: a list of entity keys that should be read from the FeatureStore.\n        \"\"\"\n        online_config = config.online_store\n        assert isinstance(online_config, DynamoDBOnlineStoreConfig)\n        dynamodb_resource = self._get_dynamodb_resource(online_config.region)\n\n        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n        for entity_key in entity_keys:\n            table_instance = dynamodb_resource.Table(_get_table_name(config, table))\n            entity_id = compute_entity_id(entity_key)\n            with tracing_span(name=\"remote_call\"):\n                response = table_instance.get_item(Key={\"entity_id\": entity_id})\n            value = response.get(\"Item\")\n\n            if value is not None:\n                res = {}\n                for feature_name, value_bin in value[\"values\"].items():\n                    val = ValueProto()\n                    val.ParseFromString(value_bin.value)\n                    res[feature_name] = val\n                result.append((datetime.fromisoformat(value[\"event_ts\"]), res))\n            else:\n                result.append((None, None))\n        return result\n\n    def _get_dynamodb_client(self, region: str):\n        if self._dynamodb_client is None:\n            self._dynamodb_client = _initialize_dynamodb_client(region)\n        return self._dynamodb_client\n\n    def _get_dynamodb_resource(self, region: str):\n        if self._dynamodb_resource is None:\n            self._dynamodb_resource = _initialize_dynamodb_resource(region)\n        return self._dynamodb_resource\n\n\ndef _initialize_dynamodb_client(region: str):\n    return boto3.client(\"dynamodb\", region_name=region)\n\n\ndef _initialize_dynamodb_resource(region: str):\n    return boto3.resource(\"dynamodb\", region_name=region)\n\n\ndef _get_table_name(config: RepoConfig, table: FeatureView) -> str:\n    return f\"{config.project}.{table.name}\"\n\n\ndef _delete_table_idempotent(\n    dynamodb_resource, table_name: str,\n):\n    try:\n        table = dynamodb_resource.Table(table_name)\n        table.delete()\n        logger.info(f\"Dynamo table {table_name} was deleted\")\n    except ClientError as ce:\n        # If the table deletion fails with ResourceNotFoundException,\n        # it means the table has already been deleted.\n        # Otherwise, re-raise the exception\n        if ce.response[\"Error\"][\"Code\"] != \"ResourceNotFoundException\":\n            raise\n        else:\n            logger.warning(f\"Trying to delete table that doesn't exist: {table_name}\")\n\n\nclass DynamoDBTable(InfraObject):\n    \"\"\"\n    A DynamoDB table managed by Feast.\n\n    Attributes:\n        name: The name of the table.\n        region: The region of the table.\n    \"\"\"\n\n    region: str\n\n    def __init__(self, name: str, region: str):\n        super().__init__(name)\n        self.region = region\n\n    def to_infra_object_proto(self) -> InfraObjectProto:\n        dynamodb_table_proto = self.to_proto()\n        return InfraObjectProto(\n            infra_object_class_type=DYNAMODB_INFRA_OBJECT_CLASS_TYPE,\n            dynamodb_table=dynamodb_table_proto,\n        )\n\n    def to_proto(self) -> Any:\n        dynamodb_table_proto = DynamoDBTableProto()\n        dynamodb_table_proto.name = self.name\n        dynamodb_table_proto.region = self.region\n        return dynamodb_table_proto\n\n    @staticmethod\n    def from_infra_object_proto(infra_object_proto: InfraObjectProto) -> Any:\n        return DynamoDBTable(\n            name=infra_object_proto.dynamodb_table.name,\n            region=infra_object_proto.dynamodb_table.region,\n        )\n\n    @staticmethod\n    def from_proto(dynamodb_table_proto: DynamoDBTableProto) -> Any:\n        return DynamoDBTable(\n            name=dynamodb_table_proto.name, region=dynamodb_table_proto.region,\n        )\n\n    def update(self):\n        dynamodb_client = _initialize_dynamodb_client(region=self.region)\n        dynamodb_resource = _initialize_dynamodb_resource(region=self.region)\n\n        try:\n            dynamodb_resource.create_table(\n                TableName=f\"{self.name}\",\n                KeySchema=[{\"AttributeName\": \"entity_id\", \"KeyType\": \"HASH\"}],\n                AttributeDefinitions=[\n                    {\"AttributeName\": \"entity_id\", \"AttributeType\": \"S\"}\n                ],\n                BillingMode=\"PAY_PER_REQUEST\",\n            )\n        except ClientError as ce:\n            # If the table creation fails with ResourceInUseException,\n            # it means the table already exists or is being created.\n            # Otherwise, re-raise the exception\n            if ce.response[\"Error\"][\"Code\"] != \"ResourceInUseException\":\n                raise\n\n        dynamodb_client.get_waiter(\"table_exists\").wait(TableName=f\"{self.name}\")\n\n    def teardown(self):\n        dynamodb_resource = _initialize_dynamodb_resource(region=self.region)\n        _delete_table_idempotent(dynamodb_resource, self.name)\n", "path": "sdk/python/feast/infra/online_stores/dynamodb.py"}]}