m

Author: Lucas McDonald

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

@@ -48,7 +48,7 @@ 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
@@ -64,7 +64,7 @@ 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

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

@@ -1,4 +1,3 @@
 # 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,21 +20,89 @@
 
 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,15 +35,15 @@
 Input = TypeVar("Input")
 Output = TypeVar("Output")
 
+
 class DynamoDbEncryption:
     """Client for DynamoDbEncryption
 
     :param config: Configuration for the client.
     """
+
     def __init__(
-        self,
-        config: DynamoDbEncryptionConfig | None = None,
-        dafny_client: IDynamoDbEncryptionClient | None = None
+        self, config: DynamoDbEncryptionConfig | None = None, dafny_client: IDynamoDbEncryptionClient | None = None
     ):
         if config is None:
             self._config = Config()
@@ -60,7 +60,9 @@ 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:
+    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
@@ -74,7 +76,9 @@ def create_dynamo_db_encryption_branch_key_id_supplier(self, input: CreateDynamo
             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.
@@ -98,9 +102,7 @@ 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.
@@ -124,13 +126,11 @@ def _handle_execution(
             transport_response=None,
         )
         try:
-          _client_interceptors = config.interceptors
+            _client_interceptors = config.interceptors
         except AttributeError:
-          config.interceptors = []
-          _client_interceptors = config.interceptors
-        client_interceptors = cast(
-            list[Interceptor[Input, Output, DafnyRequest, DafnyResponse]], _client_interceptors
-        )
+            config.interceptors = []
+            _client_interceptors = config.interceptors
+        client_interceptors = cast(list[Interceptor[Input, Output, DafnyRequest, DafnyResponse]], _client_interceptors)
         interceptors = client_interceptors
 
         try:
@@ -163,9 +163,7 @@ 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
             )
@@ -176,8 +174,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.
@@ -212,7 +210,7 @@ def _handle_execution(
                             error_info=RetryErrorInfo(
                                 # TODO: Determine the error type.
                                 error_type=RetryErrorType.CLIENT_ERROR,
-                            )
+                            ),
                         )
                     except SmithyRetryException:
                         raise context_with_response.response
@@ -226,9 +224,7 @@ def _handle_execution(
         # At this point, the context's request will have been definitively set, and
         # 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
-        )
+        execution_context = cast(InterceptorContext[Input, Output, DafnyRequest | None, DafnyResponse | None], context)
         return self._finalize_execution(interceptors, execution_context)
 
     def _handle_attempt(
@@ -248,9 +244,7 @@ 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
@@ -262,8 +256,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
@@ -275,9 +269,7 @@ 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:
@@ -289,9 +281,7 @@ def _handle_attempt(
         # the response is either set or an exception, and the transport_resposne is either set or
         # 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
-        )
+        attempt_context = cast(InterceptorContext[Input, Output, DafnyRequest, DafnyResponse | None], context)
         return self._finalize_attempt(interceptors, attempt_context)
 
     def _finalize_attempt(
@@ -302,9 +292,7 @@ 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
 

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

@@ -17,6 +17,8 @@
 
 
 _ServiceInterceptor = Any
+
+
 @dataclass(init=False)
 class Config:
     """Configuration for DynamoDbEncryption."""
@@ -46,28 +48,25 @@ def __init__(
         self.retry_strategy = retry_strategy or SimpleRetryStrategy()
         self.dafnyImplInterface = dafnyImplInterface
 
+
 # A callable that allows customizing the config object on each request.
 Plugin: TypeAlias = Callable[[Config], None]
 
+
 class DynamoDbEncryptionConfig(Config):
     def __init__(
         self,
     ):
-        """Constructor for DynamoDbEncryptionConfig
-        """
+        """Constructor for DynamoDbEncryptionConfig"""
         super().__init__()
 
     def as_dict(self) -> Dict[str, Any]:
-        """Converts the DynamoDbEncryptionConfig to a dictionary.
-
-        """
+        """Converts the DynamoDbEncryptionConfig to a dictionary."""
         return {}
 
     @staticmethod
     def from_dict(d: Dict[str, Any]) -> "DynamoDbEncryptionConfig":
-        """Creates a DynamoDbEncryptionConfig from a dictionary.
-
-        """
+        """Creates a DynamoDbEncryptionConfig from a dictionary."""
         return DynamoDbEncryptionConfig()
 
     def __repr__(self) -> str:
@@ -78,16 +77,22 @@ def __repr__(self) -> str:
     def __eq__(self, other: Any) -> bool:
         return isinstance(other, DynamoDbEncryptionConfig)
 
+
 def dafny_config_to_smithy_config(dafny_config) -> DynamoDbEncryptionConfig:
     """
     Converts the provided Dafny shape for this localService's config
     into the corresponding Smithy-modelled shape.
     """
-    return aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb.dafny_to_smithy.aws_cryptography_dbencryptionsdk_dynamodb_DynamoDbEncryptionConfig(dafny_config)
+    return aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb.dafny_to_smithy.aws_cryptography_dbencryptionsdk_dynamodb_DynamoDbEncryptionConfig(
+        dafny_config
+    )
+
 
 def smithy_config_to_dafny_config(smithy_config) -> DafnyDynamoDbEncryptionConfig:
     """
     Converts the provided Smithy-modelled shape for this localService's config
     into the corresponding Dafny shape.
     """
-    return aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb.smithy_to_dafny.aws_cryptography_dbencryptionsdk_dynamodb_DynamoDbEncryptionConfig(smithy_config)
+    return aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb.smithy_to_dafny.aws_cryptography_dbencryptionsdk_dynamodb_DynamoDbEncryptionConfig(
+        smithy_config
+    )

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

@@ -5,6 +5,7 @@
 from aws_dbesdk_dynamodb.internaldafny.generated.DynamoDbEncryption import DynamoDbEncryptionClient
 from .dafny_protocol import DafnyRequest
 
+
 class DafnyImplInterface:
     impl: DynamoDbEncryptionClient | None = None
 
@@ -22,9 +23,9 @@ def handle_request(self, input: DafnyRequest):
                 "GetEncryptedDataKeyDescription": self.impl.GetEncryptedDataKeyDescription,
             }
 
-         # This logic is where a typical Smithy client would expect the "server" to be.
-         # This code can be thought of as logic our Dafny "server" uses
-         #   to route incoming client requests to the correct request handler code.
+        # This logic is where a typical Smithy client would expect the "server" to be.
+        # This code can be thought of as logic our Dafny "server" uses
+        #   to route incoming client requests to the correct request handler code.
         if input.dafny_operation_input is None:
             return self.operation_map[input.operation_name]()
         else:

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

@@ -12,6 +12,7 @@
 import smithy_dafny_standard_library.internaldafny.generated.Wrappers as Wrappers
 from typing import Union
 
+
 class DafnyRequest:
     operation_name: str
 
@@ -26,6 +27,7 @@ def __init__(self, operation_name, dafny_operation_input):
         self.operation_name = operation_name
         self.dafny_operation_input = dafny_operation_input
 
+
 class DafnyResponse(Wrappers.Result):
     def __init__(self):
         super().__init__(self)