def handle_geofence_enter()

in src/aws-lambda/location-geofence-event/location-geofence-event.py [0:0]

• 47 lines of code
• 17 McCabe index (conditional complexity)

def handle_geofence_enter(event):
    """
    Somebody has entered the geofence. We handle it.
    Args:
        event: See the lambda_handler docstring for the kind of structure we might expect. Examples are there.

    Returns:
        None.
    """
    location_device_id = event['detail']['DeviceId']
    # If there are multiple geofences, it is possible to learn which geofence by inspecting the event
    # at event['detail']['GeofenceID']
    # and event['detail']['GeofenceCollectionARN']
    # We currently assume that there is only one Geofence Collection associated with the tracker.

    # We can get the geofence collection description by using the GeofenceCollectionID to hit the
    # geofence service.

    # Here we trust the sending party to not be sending us the wrong user ID.
    cognito_user_id = location_device_id

    user_pool_id = os.environ['UserPoolId']  # Cognito user pool for retail demo store.

    response = cognito_idp.admin_get_user(
        UserPoolId=user_pool_id,
        Username=cognito_user_id
    )

    # Grab the user attributes as a dictionary and pull relevant ones.
    cognito_user_attributes = {att['Name']: att['Value'] for att in response['UserAttributes']}

    try:
        shopper_user_id = cognito_user_attributes['custom:profile_user_id']
    except KeyError:
        logger.error(f'No profile (shopper) ID for {cognito_user_id} - quitting event handling.')
        return

    try:
        demo_journey = cognito_user_attributes['custom:demo_journey']
        logger.info(f'User has demo_journey attribute: {demo_journey}')
    except KeyError:
        demo_journey = 'PURCHASE'
        logger.warning(f'No demo journey configured for user {cognito_user_id} - defaulting to {demo_journey}')

    # Grab the service URLs.
    orders_service = get_orders_service()
    products_service = get_products_service()
    users_service = get_users_service()

    if demo_journey == 'PURCHASE':

        # If there are any Pinpoint campaigns waiting for Location events of type parametrised
        # by GEOFENCE_PINPOINT_EVENTTYPE (see defined at top)
        # They should get triggered for all endpoints for this user:
        pinpoint_fire_location_approached_event(shopper_user_id, event['detail']['SampleTime'])
        # The implementation for browser push is custom (though Apple, iOS and Android push can be also done
        # with Pinpoint campaigns).
        send_purchase_browser_notification(cognito_user_id)

    elif demo_journey == 'COLLECTION':

        to_email = cognito_user_attributes['email']

        try:
            first_name = cognito_user_attributes['custom:profile_first_name']
        except KeyError:
            first_name = 'Madam/Sir'
            logger.warning(f'No first name for user {cognito_user_id} - defaulting to "{first_name}"')

        try:
            last_name = cognito_user_attributes['custom:profile_last_name']
        except KeyError:
            last_name = ''
            logger.warning(f'No last name for user {cognito_user_id} - defaulting to "{last_name}"')

        # Get a list of waiting orders.
        username = userid_to_username(shopper_user_id, users_service)
        orders = get_orders_with_details(username, orders_service, products_service)

        if len(orders) > 0:
            logger.info(f'User {shopper_user_id}/{username} (name {first_name} {last_name}) has waiting orders')
            send_pickup_email(to_email, orders)
            send_pickup_sms(orders)  # phone number is recorded against the order
            send_pickup_browser_notification(cognito_user_id, orders)
    else:
        logger.error(f'Not a known journey type {demo_journey} for user {shopper_user_id}')