remove bson dependency and use pymongo (#7022)

calvernaz · web-flow · commit 420ddee09a85 · 2022-10-14T12:02:13.000+02:00
diff --git a/localstack/services/dynamodbstreams/dynamodbstreams_api.py b/localstack/services/dynamodbstreams/dynamodbstreams_api.py
@@ -1,9 +1,10 @@
+import contextlib
 import logging
 import threading
 import time
 from typing import Dict
 
-import bson
+from bson.json_util import dumps
 
 from localstack.aws.accounts import get_aws_account_id
 from localstack.aws.api.dynamodbstreams import StreamStatus, StreamViewType
@@ -67,29 +68,25 @@ def forward_events(records: Dict) -> None:
     kinesis = aws_stack.connect_to_service("kinesis")
     for record in records:
         table_arn = record.pop("eventSourceARN", "")
-        stream = get_stream_for_table(table_arn)
-        if stream:
+        if stream := get_stream_for_table(table_arn):
             table_name = table_name_from_stream_arn(stream["StreamArn"])
             stream_name = get_kinesis_stream_name(table_name)
             kinesis.put_record(
                 StreamName=stream_name,
-                Data=bson.dumps(record),
+                Data=dumps(record),
                 PartitionKey="TODO",
             )
 
 
 def delete_streams(table_arn: str) -> None:
     store = get_dynamodbstreams_store()
     table_name = table_name_from_table_arn(table_arn)
-    stream = store.ddb_streams.pop(table_name, None)
-    if stream:
+    if store.ddb_streams.pop(table_name, None):
         stream_name = get_kinesis_stream_name(table_name)
-        try:
+        with contextlib.suppress(Exception):
             aws_stack.connect_to_service("kinesis").delete_stream(StreamName=stream_name)
             # sleep a bit, as stream deletion can take some time ...
             time.sleep(1)
-        except Exception:
-            pass  # ignore "stream not found" errors
 
 
 def get_kinesis_stream_name(table_name: str) -> str:
diff --git a/localstack/services/dynamodbstreams/provider.py b/localstack/services/dynamodbstreams/provider.py
@@ -1,7 +1,7 @@
 import copy
 import logging
 
-import bson
+from bson.json_util import loads
 
 from localstack.aws.api import RequestContext, handler
 from localstack.aws.api.dynamodbstreams import (
@@ -103,7 +103,7 @@ def get_records(self, context: RequestContext, payload: GetRecordsInput) -> GetR
             "NextShardIterator": kinesis_records.get("NextShardIterator"),
         }
         for record in kinesis_records["Records"]:
-            record_data = bson.loads(record["Data"])
+            record_data = loads(record["Data"])
             record_data["dynamodb"]["SequenceNumber"] = record["SequenceNumber"]
             result["Records"].append(record_data)
         return GetRecordsOutput(**result)
diff --git a/setup.cfg b/setup.cfg
@@ -92,7 +92,7 @@ runtime =
     xmltodict>=0.11.0
     awscrt>=0.13.14
     vosk==0.3.43
-    bson==0.5.10
+    pymongo>=4.2.0
 
 # @deprecated - use extra 'runtime' instead.
 full =
