Source code for lambdas.utils.dynamodb_util

import os
import logging
from pymorri.exceptions import MoriException
from pymorri.util.aws_utils import dynamodb_get_rows
from pymorri.util.aws_utils import dynamodb_update_item
from app.lambdas.utils.constants import STOREPICK_PRODUCT_EVENT_HANDLER
from pymorri.util.log_utils import setup_json_formatter_handler, setup_message_format

logger = logging.getLogger(__name__)
logger.setLevel(level=logging.INFO)

setup_json_formatter_handler(
    STOREPICK_PRODUCT_EVENT_HANDLER,
    logger,
    type="DYNAMODB",
    event_type="product")


def last_event_id(config_details: dict):
    """
        This function is used to get last_event_id(watermark) from dynamo_db_table to process further

        :param config_details: config details

        Returns: get last event id
    """
    try:
        last_event_id, time_stamp = None, None

        items = dynamodb_get_rows(config_details.get(
                'EVENTS_WATERMARKS_TABLE'), "event", 'product')

        for item in items:
            last_event_id = item.get('last_event_id')
            break
        return last_event_id
    except Exception as ex:
        logger.error(
            setup_message_format(
                message=f"Failed to fetch last event id {last_event_id} {ex}"))
        raise MoriException(detail=ex)


def update_last_event(
        config_details,
        last_event_id: str,
        updated_timestamp: str):
    """
        This function is used to update the
        watermark dynamodb table with last event id from product v1 API call.

        :param config_details: config details
        :param last_event_id: last event id to update in dynamoDB table
        :param updated_timestamp: time stamp to update

    """
    try:
        db_keys_dict = {"event": "product"}

        expression = f"SET last_event_id=:last_event_id, update_timestamp=:update_timestamp,event_source=:event_source"
        dynamodb_update_item(
            config_details.get('EVENTS_WATERMARKS_TABLE'), db_keys_dict, expression, {
                ':last_event_id': last_event_id, ":update_timestamp": updated_timestamp,":event_source": config_details['GETPRODUCTEVENT_URL']})
    except Exception as ex:
        logger.error(
            setup_message_format(
                message=f"Failed to update last event id {last_event_id} {ex}"))
        raise MoriException(detail=ex)