def parse_record()

in warehouse/src/table_update/main.py [0:0]

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

def parse_record(ddb_record: dict) -> Optional[dict]:
    """
    Parse a DynamoDB record into an EventBridge event
    """

    # Discard records that concern removed events, non-metadata items or items that are
    # not in the COMPLETED status.
    if (ddb_record["eventName"].upper() == "REMOVE"
            or ddb_record["dynamodb"]["NewImage"]["productId"]["S"] != METADATA_KEY
            or ddb_record["dynamodb"]["NewImage"]["status"]["S"] != "COMPLETED"):
        return None

    # Gather information
    order_id = ddb_record["dynamodb"]["NewImage"]["orderId"]["S"]
    products = get_products(order_id)

    # Create the detail
    detail_type = "PackagingFailed"
    detail = {
        "orderId": order_id
    }
    # If there are products, we successfully created a package
    if len(products) > 0:
        detail_type = "PackageCreated"
        detail["products"] = products

    metrics.add_metric(name=event_type_to_metric[detail_type], unit=MetricUnit.Count, value=1)

    # Return event
    return {
        "Time": datetime.datetime.now(),
        "Source": "ecommerce.warehouse",
        "Resources": [order_id],
        "EventBusName": EVENT_BUS_NAME,
        "DetailType": detail_type,
        "Detail": json.dumps(detail, cls=Encoder)
    }