def _process_batch_write_response()

in src/dynamodb_encryption_sdk/internal/utils.py [0:0]

• 23 lines of code
• 10 McCabe index (conditional complexity)

def _process_batch_write_response(request, response, table_crypto_config):
    # type: (Dict, Dict, Dict[Text, CryptoConfig]) -> Dict
    """Handle unprocessed items in the response from a transparently encrypted write.

    :param dict request: The DynamoDB plaintext request dictionary
    :param dict response: The DynamoDB response from the batch operation
    :param Dict[Text, CryptoConfig] table_crypto_config: table level CryptoConfig used in encrypting the request items
    :return: DynamoDB response, with any unprocessed items reverted back to the original plaintext values
    :rtype: dict
    """
    try:
        unprocessed_items = response["UnprocessedItems"]
    except KeyError:
        return response

    # Unprocessed items need to be returned in their original state
    for table_name, unprocessed in unprocessed_items.items():
        original_items = request[table_name]
        crypto_config = table_crypto_config[table_name]

        if crypto_config.encryption_context.partition_key_name:
            items_match = partial(_item_keys_match, crypto_config)
        else:
            items_match = partial(_item_attributes_match, crypto_config)

        for pos, operation in enumerate(unprocessed):
            for request_type, item in operation.items():
                if request_type != "PutRequest":
                    continue

                for plaintext_item in original_items:
                    if plaintext_item.get(request_type) and items_match(
                        plaintext_item[request_type]["Item"], item["Item"]
                    ):
                        unprocessed[pos] = plaintext_item.copy()
                        break

    return response