Codebase Improvements

use-cases/data-kiosk/code/python/src/process_notification_handler.py

@@ -18,74 +18,109 @@ def lambda_handler(event, context):
     try:
         execution_arn = None
         # Iterate over SQS messages
-        for message in event['Records']:
-
-            body_message = message['body']
+        for message in event["Records"]:
+            body_message = message["body"]
             logger.info(f"Received new notification: {body_message}")
 
             notification = json.loads(body_message)
 
-            notification_type = notification[constants.NOTIFICATION_KEY_NOTIFICATION_TYPE]
+            notification_type = notification[
+                constants.NOTIFICATION_KEY_NOTIFICATION_TYPE
+            ]
 
             # Only process the notification if it is of type 'DATA_KIOSK_QUERY_PROCESSING_FINISHED'
             if notification_type != constants.NOTIFICATION_TYPE_DK:
                 logger.warning(f"Notification type {notification_type} skipped")
                 continue
 
-            dk_notification = notification['payload']
+            dk_notification = notification["payload"]
 
             # Start a Step Functions workflow execution to retrieve query results from Data Kiosk
             execution_arn, message = start_step_functions_execution(dk_notification)
-            logger.info(f"Notification Processing Message: {message}, Execution arn: {execution_arn}")
+            logger.info(
+                f"Notification Processing Message: {message}, Execution arn: {execution_arn}"
+            )
 
         return "Finished processing incoming notifications"
 
     except json.JSONDecodeError as e:
-        logger.error(f"Message body could not be mapped to an SP-API Notification: {e.__dict__}")
+        logger.error(
+            f"Message body could not be mapped to an SP-API Notification: {e.__dict__}"
+        )
 
 
-# This function starts a Step Functions workflow execution.
-def start_step_functions_execution(dk_notification):
+# Extracts necessary data from notification.
+def extract_info_from_notification(dk_notification):
     query_id = dk_notification[constants.NOTIFICATION_KEY_QUERY_ID]
     query = dk_notification[constants.NOTIFICATION_KEY_QUERY]
     processing_status = dk_notification[constants.NOTIFICATION_KEY_PROCESSING_STATUS]
     account_id = dk_notification[constants.NOTIFICATION_KEY_ACCOUNT_ID]
-    state_machine_arn = os.environ.get(constants.STATE_MACHINE_ARN_ENV_VARIABLE)
-    data_document_id = dk_notification.get(constants.NOTIFICATION_KEY_DATA_DOCUMENT_ID, None)
+    return query_id, query, processing_status, account_id
 
+
+# Determines the appropriate document ID based on processing status.
+def determine_document_id(dk_notification, processing_status):
+    notification_key = constants.NOTIFICATION_KEY_DATA_DOCUMENT_ID
     if processing_status == constants.NOTIFICATION_KEY_FATAL_PROCESSING:
-        logger.info(f"Query Processing is {processing_status}. Check the Error Document for more info. "
-                    f"Change the query and submit again.")
-
-        # Setting document_id to parse errorDocumentId instead of dataDocumentId since the processing is not DONE
-        if dk_notification[constants.NOTIFICATION_KEY_ERROR_DOCUMENT_ID]:
-            document_id = dk_notification[constants.NOTIFICATION_KEY_ERROR_DOCUMENT_ID]
-        else:
-            message = "Processing Done: Status - Fatal : No error document"
-            return None, message
-    elif data_document_id:
-        document_id = dk_notification[constants.NOTIFICATION_KEY_DATA_DOCUMENT_ID]
-    else:
-        message = "Processing Done: Status - Done : No data document available (Empty Data)"
-        return None, message
-
-    input_payload = {
+        logger.info(
+            f"Query Processing is {processing_status}. Check the Error Document for more info."
+        )
+        notification_key = constants.NOTIFICATION_KEY_ERROR_DOCUMENT_ID
+    return dk_notification.get(notification_key, None)
+
+
+# Constructs the input payload for the state machine.
+def construct_input_payload(
+    query_id, query, processing_status, account_id, document_id
+):
+    document = (
+        {constants.STATE_MACHINE_KEY_DOCUMENT_ID: document_id} if document_id else {}
+    )
+    return {
         constants.STATE_MACHINE_KEY_QUERY_ID: query_id,
         constants.STATE_MACHINE_KEY_QUERY: query,
         constants.NOTIFICATION_KEY_PROCESSING_STATUS: processing_status,
-        constants.STATE_MACHINE_KEY_DOCUMENT: {
-            constants.STATE_MACHINE_KEY_DOCUMENT_ID: document_id
-        },
-        constants.STATE_MACHINE_KEY_ACCOUNT_ID: account_id
+        constants.STATE_MACHINE_KEY_DOCUMENT: document,
+        constants.STATE_MACHINE_KEY_ACCOUNT_ID: account_id,
     }
 
-    request = {
-        'stateMachineArn': state_machine_arn,
-        'name': f"{account_id}-{query_id}-{str(uuid.uuid4())}",
-        'input': json.dumps(input_payload)
-    }
 
+# Start the Step Functions execution
+def start_execution(state_machine_arn, account_id, query_id, input_payload):
     step_functions = boto3.client(constants.AWS_STEP_FUNCTIONS_CLIENT_NAME)
+    execution_name = f"{account_id}-{query_id}-{str(uuid.uuid4())}"
+    request = {
+        "stateMachineArn": state_machine_arn,
+        "name": execution_name,
+        "input": json.dumps(input_payload),
+    }
     result = step_functions.start_execution(**request)
+    return result[constants.AWS_STEP_FUNCTIONS_EXECUTION_ARN_NAME]
+
 
-    return result[constants.AWS_STEP_FUNCTIONS_EXECUTION_ARN_NAME], "State machine successfully started. "
+# This function starts a Step Functions workflow execution.
+def start_step_functions_execution(dk_notification):
+    state_machine_arn = os.environ.get(constants.STATE_MACHINE_ARN_ENV_VARIABLE)
+    query_id, query, processing_status, account_id = extract_info_from_notification(
+        dk_notification
+    )
+    document_id = determine_document_id(dk_notification, processing_status)
+
+    if (
+        not document_id
+        and processing_status == constants.NOTIFICATION_KEY_FATAL_PROCESSING
+    ):
+        return None, "Processing Done: Status - Fatal : No error document available"
+    elif not document_id:
+        return (
+            None,
+            "Processing Done: Status - Done : No data document available (Empty Data)",
+        )
+
+    input_payload = construct_input_payload(
+        query_id, query, processing_status, account_id, document_id
+    )
+    execution_arn = start_execution(
+        state_machine_arn, account_id, query_id, input_payload
+    )
+    return execution_arn, "State machine successfully started."

use-cases/data-kiosk/code/python/src/store_document_handler.py

@@ -73,8 +73,6 @@ def get_json_from_url(url):
         if response.info().get('Content-Encoding') == 'gzip':
             data = gzip.decompress(data)
 
-        # Decode the content as UTF-8
-        decoded_data = data.decode('utf-8')
         return data
 
 

use-cases/data-kiosk/code/python/src/subscribe_notifications_handler.py

@@ -66,7 +66,6 @@ def create_destination(region_code, refresh_token):
     # Invoke the Notifications API using a grantless notifications scope
     api_utils = ApiUtils(refresh_token, region_code, "notifications", constants.LWA_NOTIFICATIONS_SCOPE)
 
-    destination_id = ""
     try:
         # Create destination
         create_destination_response = api_utils.call_notifications_api(

use-cases/merchant-fulfillment/code/python/src/inventory_check_handler.py

@@ -1,3 +1,4 @@
+from collections import defaultdict
 import os
 import boto3
 import logging
@@ -15,70 +16,82 @@
 logger = logging.getLogger(__name__)
 
 
-def lambda_handler(event: MfnOrder, context):
-    logger.info(f"InventoryCheck Lambda input: {event}")
-
-    package_weight_value = 0
-    package_weight_unit = ""
-
-    package_length = 0
-    package_width = 0
-    package_height = 0
-    package_size_unit = ""
-
-    mfn_order = MfnOrder.from_dict(event)
-    # Iterate over all order items and retrieve stock, size, and weight from the database
-    for order_item in mfn_order.orderItems:
-        # Retrieve the item from DynamoDB by SKU
-        # Update this section to match your product's logic
-        key = {constants.INVENTORY_TABLE_HASH_KEY_NAME: {"S": order_item.sku}}
-
-        dynamodb = boto3.client(constants.AWS_DYNAMO_DB_CLIENT_KEY_NAME)
-        get_item_result = dynamodb.get_item(TableName=os.environ.get(constants.INVENTORY_TABLE_NAME_ENV_VARIABLE),
-                                            Key=key)
-        item = get_item_result.get('Item', {})
-
-        stock = int(item.get(constants.INVENTORY_TABLE_STOCK_ATTRIBUTE_NAME, {"N": "0"})["N"])
-        if stock < order_item.quantity:
-            raise Exception(f"Stock level for SKU {order_item.sku} is not enough to fulfill the requested quantity")
-
-        item_weight_value = int(item.get(constants.INVENTORY_TABLE_WEIGHT_VALUE_ATTRIBUTE_NAME, {"N": "0"})["N"])
-
-        # Valid values for the database records are uppercase: [OZ, G]
-        item_weight_unit = item.get(constants.INVENTORY_TABLE_WEIGHT_UNIT_ATTRIBUTE_NAME, {"S": ""})["S"]
+def get_item_from_dynamodb(sku):
+    dynamodb = boto3.client(constants.AWS_DYNAMO_DB_CLIENT_KEY_NAME)
+    key = {constants.INVENTORY_TABLE_HASH_KEY_NAME: {"S": sku}}
+    response = dynamodb.get_item(
+        TableName=os.environ.get(constants.INVENTORY_TABLE_NAME_ENV_VARIABLE), Key=key
+    )
+    return response.get("Item", {})
 
-        item_length = int(item.get(constants.INVENTORY_TABLE_LENGTH_ATTRIBUTE_NAME, {"N": "0"})["N"])
-        item_width = int(item.get(constants.INVENTORY_TABLE_WIDTH_ATTRIBUTE_NAME, {"N": "0"})["N"])
-        item_height = int(item.get(constants.INVENTORY_TABLE_HEIGHT_ATTRIBUTE_NAME, {"N": "0"})["N"])
 
-        # Valid values for the database records are uppercase: [INCHES, CENTIMETERS]
-        item_size_unit = item.get(constants.INVENTORY_TABLE_SIZE_UNIT_ATTRIBUTE_NAME, {"S": ""})["S"]
+def check_stock_and_update_order(order_item, item):
+    stock = int(
+        item.get(constants.INVENTORY_TABLE_STOCK_ATTRIBUTE_NAME, {"N": "0"})["N"]
+    )
+    if stock < order_item.quantity:
+        raise Exception(
+            f"Stock level for SKU {order_item.sku} is not enough to fulfill the requested quantity"
+        )
+    order_item.itemWeight = parse_weight(item)
 
-        unit_of_weight_enum = UnitOfWeight.G if item_weight_unit == "G" else UnitOfWeight.OZ if item_weight_unit == "OZ" else None
-        item_weight = Weight(unit=unit_of_weight_enum, value=Decimal(str(item_weight_value)))
 
-        order_item.itemWeight = item_weight
+def parse_weight(item):
+    weight_value = int(
+        item.get(constants.INVENTORY_TABLE_WEIGHT_VALUE_ATTRIBUTE_NAME, {"N": "0"})["N"]
+    )
+    weight_unit = item.get(
+        constants.INVENTORY_TABLE_WEIGHT_UNIT_ATTRIBUTE_NAME, {"S": ""}
+    )["S"]
+    unit_enum = (
+        UnitOfWeight.G
+        if weight_unit == "G"
+        else UnitOfWeight.OZ
+        if weight_unit == "OZ"
+        else None
+    )
+    return Weight(unit=unit_enum, value=Decimal(str(weight_value)))
+
+
+def calculate_package_dimensions(order_items):
+    dimensions = {"length": 0, "width": 0, "height": 0}
+    units = defaultdict(int)
+    for item in order_items:
+        dimensions["length"] += int(
+            item.get(constants.INVENTORY_TABLE_LENGTH_ATTRIBUTE_NAME, {"N": "0"})["N"]
+        )
+        dimensions["width"] += int(
+            item.get(constants.INVENTORY_TABLE_WIDTH_ATTRIBUTE_NAME, {"N": "0"})["N"]
+        )
+        dimensions["height"] += int(
+            item.get(constants.INVENTORY_TABLE_HEIGHT_ATTRIBUTE_NAME, {"N": "0"})["N"]
+        )
+        item_size_unit = item.get(
+            constants.INVENTORY_TABLE_SIZE_UNIT_ATTRIBUTE_NAME, {"S": ""}
+        )["S"]
+        units[item_size_unit] += 1
+
+    # Determine the most common unit of measurement
+    unit_of_length_enum = max(units, key=units.get)
+    return PackageDimensions(
+        length=dimensions["length"],
+        width=dimensions["width"],
+        height=dimensions["height"],
+        unit=getattr(UnitOfLength, unit_of_length_enum),
+    )
 
-        # Package weight is calculated by adding the individual weights
-        # Update this section to match your selling partners' logic
-        package_weight_value += item_weight_value
-        package_weight_unit = item_weight_unit
 
-        # Package size is calculated by adding the individual sizes
-        # Update this section to match your selling partners' logic
-        package_length += item_length
-        package_width += item_width
-        package_height += item_height
-        package_size_unit = item_size_unit
+def lambda_handler(event, context):
+    logger.info(f"InventoryCheck Lambda input: {event}")
 
-    event_unit_of_weight_enum = UnitOfWeight.G if package_weight_unit == "G" else UnitOfWeight.OZ if package_weight_unit == "OZ" else None
-    event_unit_of_length_enum = UnitOfLength.INCHES if package_size_unit == "INCHES" else UnitOfLength.CENTIMETERS if package_size_unit == "CENTIMETERS" else None
+    mfn_order = MfnOrder.from_dict(event)
+    for order_item in mfn_order.orderItems:
+        item = get_item_from_dynamodb(order_item.sku)
+        check_stock_and_update_order(order_item, item)
 
-    mfn_order.weight = Weight(unit=event_unit_of_weight_enum, value=Decimal(str(package_weight_value)))
-    mfn_order.packageDimensions = PackageDimensions(
-        length=package_length,
-        width=package_width,
-        height=package_height,
-        unit=event_unit_of_length_enum
+    package_dimensions = calculate_package_dimensions(
+        [item.to_json() for item in mfn_order.orderItems]
     )
+    mfn_order.packageDimensions = package_dimensions
+
     return mfn_order.to_json()