m

Author: Lucas McDonald

DynamoDbEncryption/runtimes/python/src/aws_dbesdk_dynamodb/internaldafny/extern/InternalLegacyOverride.py

@@ -1,5 +1,6 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
 from aws_dbesdk_dynamodb.internaldafny.generated.AwsCryptographyDbEncryptionSdkDynamoDbItemEncryptorTypes import (
-    DynamoDbItemEncryptorConfig,
     DynamoDbItemEncryptorConfig_DynamoDbItemEncryptorConfig,
 )
 import smithy_dafny_standard_library.internaldafny.generated.Wrappers as Wrappers
@@ -47,7 +48,10 @@ def Build(config: DynamoDbItemEncryptorConfig_DynamoDbItemEncryptorConfig):
         maybe_actions = InternalLegacyOverride.legacyActions(legacy_override.attributeActionsOnEncrypt)
         if maybe_actions.is_Failure:
             return InternalLegacyOverride.CreateBuildFailure(maybe_actions.error())
+        
+        # TODO: Implement this
 
+    @staticmethod
     def legacyEncryptionContext(config: DynamoDbItemEncryptorConfig_DynamoDbItemEncryptorConfig):
         try:
             encryption_context_kwargs = {
@@ -60,17 +64,25 @@ def legacyEncryptionContext(config: DynamoDbItemEncryptorConfig_DynamoDbItemEncr
             return InternalLegacyOverride.CreateBuildSuccess(encryption_context)
         except Exception as e:
             return InternalLegacyOverride.CreateBuildFailure(InternalLegacyOverride.CreateError(str(e)))
+            
+    @staticmethod
+    def legacyActions(attribute_actions_on_encrypt):
+        # TODO: Implement this
+        pass
 
     @staticmethod
     def EncryptItem(input):
+        # TODO: Implement this
         return Wrappers.Result_Failure("TODO-legacy-Encryptitem")
 
     @staticmethod
     def DecryptItem(input):
+        # TODO: Implement this
         return Wrappers.Result_Failure("TODO-legacy-Decryptitem")
 
     @staticmethod
     def IsLegacyinput(input):
+        # TODO: Implement this
         return False
 
 

DynamoDbEncryption/runtimes/python/src/aws_dbesdk_dynamodb/internaldafny/extern/__init__.py

@@ -1,3 +1,5 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
 from . import (
     InternalLegacyOverride,
 )

DynamoDbEncryption/runtimes/python/src/aws_dbesdk_dynamodb/smithygenerated/aws_cryptography_dbencryptionsdk_dynamodb/__init__.py

@@ -1,3 +1,4 @@
 # Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 # Do not modify this file. This file is machine generated, and any changes to it will be overwritten.
+

DynamoDbEncryption/runtimes/python/src/aws_dbesdk_dynamodb/smithygenerated/aws_cryptography_dbencryptionsdk_dynamodb/aws_sdk_to_dafny.py

@@ -20,89 +20,21 @@
 
 def com_amazonaws_dynamodb_AttributeValue(native_input):
     if "S" in native_input.keys():
-        AttributeValue_union_value = AttributeValue_S(
-            Seq(
-                "".join(
-                    [
-                        chr(int.from_bytes(pair, "big"))
-                        for pair in zip(*[iter(native_input["S"].encode("utf-16-be"))] * 2)
-                    ]
-                )
-            )
-        )
+        AttributeValue_union_value = AttributeValue_S(Seq(''.join([chr(int.from_bytes(pair, 'big')) for pair in zip(*[iter(native_input["S"].encode('utf-16-be'))]*2)])))
     elif "N" in native_input.keys():
-        AttributeValue_union_value = AttributeValue_N(
-            Seq(
-                "".join(
-                    [
-                        chr(int.from_bytes(pair, "big"))
-                        for pair in zip(*[iter(native_input["N"].encode("utf-16-be"))] * 2)
-                    ]
-                )
-            )
-        )
+        AttributeValue_union_value = AttributeValue_N(Seq(''.join([chr(int.from_bytes(pair, 'big')) for pair in zip(*[iter(native_input["N"].encode('utf-16-be'))]*2)])))
     elif "B" in native_input.keys():
         AttributeValue_union_value = AttributeValue_B(Seq(native_input["B"]))
     elif "SS" in native_input.keys():
-        AttributeValue_union_value = AttributeValue_SS(
-            Seq(
-                [
-                    Seq(
-                        "".join(
-                            [
-                                chr(int.from_bytes(pair, "big"))
-                                for pair in zip(*[iter(list_element.encode("utf-16-be"))] * 2)
-                            ]
-                        )
-                    )
-                    for list_element in native_input["SS"]
-                ]
-            )
-        )
+        AttributeValue_union_value = AttributeValue_SS(Seq([Seq(''.join([chr(int.from_bytes(pair, 'big')) for pair in zip(*[iter(list_element.encode('utf-16-be'))]*2)])) for list_element in native_input["SS"]]))
     elif "NS" in native_input.keys():
-        AttributeValue_union_value = AttributeValue_NS(
-            Seq(
-                [
-                    Seq(
-                        "".join(
-                            [
-                                chr(int.from_bytes(pair, "big"))
-                                for pair in zip(*[iter(list_element.encode("utf-16-be"))] * 2)
-                            ]
-                        )
-                    )
-                    for list_element in native_input["NS"]
-                ]
-            )
-        )
+        AttributeValue_union_value = AttributeValue_NS(Seq([Seq(''.join([chr(int.from_bytes(pair, 'big')) for pair in zip(*[iter(list_element.encode('utf-16-be'))]*2)])) for list_element in native_input["NS"]]))
     elif "BS" in native_input.keys():
         AttributeValue_union_value = AttributeValue_BS(Seq([Seq(list_element) for list_element in native_input["BS"]]))
     elif "M" in native_input.keys():
-        AttributeValue_union_value = AttributeValue_M(
-            Map(
-                {
-                    Seq(
-                        "".join(
-                            [chr(int.from_bytes(pair, "big")) for pair in zip(*[iter(key.encode("utf-16-be"))] * 2)]
-                        )
-                    ): aws_cryptography_internal_dynamodb.smithygenerated.com_amazonaws_dynamodb.aws_sdk_to_dafny.com_amazonaws_dynamodb_AttributeValue(
-                        value
-                    )
-                    for (key, value) in native_input["M"].items()
-                }
-            )
-        )
+        AttributeValue_union_value = AttributeValue_M(Map({Seq(''.join([chr(int.from_bytes(pair, 'big')) for pair in zip(*[iter(key.encode('utf-16-be'))]*2)])): aws_cryptography_internal_dynamodb.smithygenerated.com_amazonaws_dynamodb.aws_sdk_to_dafny.com_amazonaws_dynamodb_AttributeValue(value) for (key, value) in native_input["M"].items() }))
     elif "L" in native_input.keys():
-        AttributeValue_union_value = AttributeValue_L(
-            Seq(
-                [
-                    aws_cryptography_internal_dynamodb.smithygenerated.com_amazonaws_dynamodb.aws_sdk_to_dafny.com_amazonaws_dynamodb_AttributeValue(
-                        list_element
-                    )
-                    for list_element in native_input["L"]
-                ]
-            )
-        )
+        AttributeValue_union_value = AttributeValue_L(Seq([aws_cryptography_internal_dynamodb.smithygenerated.com_amazonaws_dynamodb.aws_sdk_to_dafny.com_amazonaws_dynamodb_AttributeValue(list_element) for list_element in native_input["L"]]))
     elif "NULL" in native_input.keys():
         AttributeValue_union_value = AttributeValue_NULL(native_input["NULL"])
     elif "BOOL" in native_input.keys():

DynamoDbEncryption/runtimes/python/src/aws_dbesdk_dynamodb/smithygenerated/aws_cryptography_dbencryptionsdk_dynamodb/client.py

@@ -35,17 +35,15 @@
 Input = TypeVar("Input")
 Output = TypeVar("Output")
 
-
 class DynamoDbEncryption:
-    """Client for DynamoDbEncryption.
+    """Client for DynamoDbEncryption
 
     :param config: Configuration for the client.
     """
-
     def __init__(
         self,
         config: DynamoDbEncryptionConfig | None = None,
-        dafny_client: IDynamoDbEncryptionClient | None = None,
+        dafny_client: IDynamoDbEncryptionClient | None = None
     ):
         if config is None:
             self._config = Config()
@@ -62,15 +60,10 @@ def __init__(
         if dafny_client is not None:
             self._config.dafnyImplInterface.impl = dafny_client
 
-    def create_dynamo_db_encryption_branch_key_id_supplier(
-        self, input: CreateDynamoDbEncryptionBranchKeyIdSupplierInput
-    ) -> CreateDynamoDbEncryptionBranchKeyIdSupplierOutput:
-        """Create a Branch Key Supplier for use with the Hierarchical Keyring
-        that decides what Branch Key to use based on the primary key of the
-        DynamoDB item being read or written.
+    def create_dynamo_db_encryption_branch_key_id_supplier(self, input: CreateDynamoDbEncryptionBranchKeyIdSupplierInput) -> CreateDynamoDbEncryptionBranchKeyIdSupplierOutput:
+        """Create a Branch Key Supplier for use with the Hierarchical Keyring that decides what Branch Key to use based on the primary key of the DynamoDB item being read or written.
 
-        :param input: Inputs for creating a Branch Key Supplier from a
-            DynamoDB Key Branch Key Id Supplier
+        :param input: Inputs for creating a Branch Key Supplier from a DynamoDB Key Branch Key Id Supplier
         """
         return self._execute_operation(
             input=input,
@@ -81,9 +74,7 @@ def create_dynamo_db_encryption_branch_key_id_supplier(
             operation_name="CreateDynamoDbEncryptionBranchKeyIdSupplier",
         )
 
-    def get_encrypted_data_key_description(
-        self, input: GetEncryptedDataKeyDescriptionInput
-    ) -> GetEncryptedDataKeyDescriptionOutput:
+    def get_encrypted_data_key_description(self, input: GetEncryptedDataKeyDescriptionInput) -> GetEncryptedDataKeyDescriptionOutput:
         """Returns encrypted data key description.
 
         :param input: Input for getting encrypted data key description.
@@ -107,7 +98,9 @@ def _execute_operation(
         operation_name: str,
     ) -> Output:
         try:
-            return self._handle_execution(input, plugins, serialize, deserialize, config, operation_name)
+            return self._handle_execution(
+                input, plugins, serialize, deserialize, config, operation_name
+            )
         except Exception as e:
             # Make sure every exception that we throw is an instance of ServiceError so
             # customers can reliably catch everything we throw.
@@ -131,13 +124,12 @@ def _handle_execution(
             transport_response=None,
         )
         try:
-            _client_interceptors = config.interceptors
+          _client_interceptors = config.interceptors
         except AttributeError:
-            config.interceptors = []
-            _client_interceptors = config.interceptors
+          config.interceptors = []
+          _client_interceptors = config.interceptors
         client_interceptors = cast(
-            list[Interceptor[Input, Output, DafnyRequest, DafnyResponse]],
-            _client_interceptors,
+            list[Interceptor[Input, Output, DafnyRequest, DafnyResponse]], _client_interceptors
         )
         interceptors = client_interceptors
 
@@ -171,7 +163,9 @@ def _handle_execution(
                 interceptor.read_before_serialization(context)
 
             # Step 4: Serialize the request
-            context_with_transport_request = cast(InterceptorContext[Input, None, DafnyRequest, None], context)
+            context_with_transport_request = cast(
+                InterceptorContext[Input, None, DafnyRequest, None], context
+            )
             context_with_transport_request._transport_request = serialize(
                 context_with_transport_request.request, config
             )
@@ -182,8 +176,8 @@ def _handle_execution(
 
             # Step 6: Invoke modify_before_retry_loop
             for interceptor in interceptors:
-                context_with_transport_request._transport_request = interceptor.modify_before_retry_loop(
-                    context_with_transport_request
+                context_with_transport_request._transport_request = (
+                    interceptor.modify_before_retry_loop(context_with_transport_request)
                 )
 
             # Step 7: Acquire the retry token.
@@ -218,7 +212,7 @@ def _handle_execution(
                             error_info=RetryErrorInfo(
                                 # TODO: Determine the error type.
                                 error_type=RetryErrorType.CLIENT_ERROR,
-                            ),
+                            )
                         )
                     except SmithyRetryException:
                         raise context_with_response.response
@@ -233,8 +227,7 @@ def _handle_execution(
         # The response will be set either with the modeled output or an exception. The
         # transport_request and transport_response may be set or None.
         execution_context = cast(
-            InterceptorContext[Input, Output, DafnyRequest | None, DafnyResponse | None],
-            context,
+            InterceptorContext[Input, Output, DafnyRequest | None, DafnyResponse | None], context
         )
         return self._finalize_execution(interceptors, execution_context)
 
@@ -255,7 +248,9 @@ def _handle_attempt(
             if config.dafnyImplInterface.impl is None:
                 raise Exception("No impl found on the operation config.")
 
-            context_with_response = cast(InterceptorContext[Input, None, DafnyRequest, DafnyResponse], context)
+            context_with_response = cast(
+                InterceptorContext[Input, None, DafnyRequest, DafnyResponse], context
+            )
 
             context_with_response._transport_response = config.dafnyImplInterface.handle_request(
                 input=context_with_response.transport_request
@@ -267,8 +262,8 @@ def _handle_attempt(
 
             # Step 7o: Invoke modify_before_deserialization
             for interceptor in interceptors:
-                context_with_response._transport_response = interceptor.modify_before_deserialization(
-                    context_with_response
+                context_with_response._transport_response = (
+                    interceptor.modify_before_deserialization(context_with_response)
                 )
 
             # Step 7p: Invoke read_before_deserialization
@@ -280,7 +275,9 @@ def _handle_attempt(
                 InterceptorContext[Input, Output, DafnyRequest, DafnyResponse],
                 context_with_response,
             )
-            context_with_output._response = deserialize(context_with_output._transport_response, config)
+            context_with_output._response = deserialize(
+                context_with_output._transport_response, config
+            )
 
             # Step 7r: Invoke read_after_deserialization
             for interceptor in interceptors:
@@ -293,8 +290,7 @@ def _handle_attempt(
         # None. This will also be true after _finalize_attempt because there is no opportunity
         # there to set the transport_response.
         attempt_context = cast(
-            InterceptorContext[Input, Output, DafnyRequest, DafnyResponse | None],
-            context,
+            InterceptorContext[Input, Output, DafnyRequest, DafnyResponse | None], context
         )
         return self._finalize_attempt(interceptors, attempt_context)
 
@@ -306,7 +302,9 @@ def _finalize_attempt(
         # Step 7s: Invoke modify_before_attempt_completion
         try:
             for interceptor in interceptors:
-                context._response = interceptor.modify_before_attempt_completion(context)
+                context._response = interceptor.modify_before_attempt_completion(
+                    context
+                )
         except Exception as e:
             context._response = e