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_RANGE_SPACE_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_RANGE_SPACE_HANDLER,
    logger,
    type="DYNAMODB",
    event_type="range")


def last_event_id(config_details: dict, event_type: str):
    """

        This function is used to get last_event_id(watermark) from dynamo_db_table to process further

        :param event_type: This argument is used for reding data from dynamoDB table value is
        :param config_details: config details

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

        if "space" == event_type:
            items = dynamodb_get_rows(config_details.get(
                'EVENTS_WATERMARKS_TABLE'), "event", 'space')
        else:
            items = dynamodb_get_rows(config_details.get(
                'EVENTS_WATERMARKS_TABLE'), "event", 'range')

        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,
        event_entity: str,
        last_event_id: str,
        updated_timestamp: str):
    """

        This function is used to update the
        watermark dynamodb table with last event id from range v1 API call.

        :param config_details: config details
        :param event_entity: 'range' or 'space' events
        :param last_event_id: last event id to update in dynamoDb table
        :param updated_timestamp: time stamp to update

    """
    try:
        if "space" == event_entity:
            db_keys_dict = {"event": "space"}
        else:
            db_keys_dict = {"event": "range"}

        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['GETRANGEANDSPACEEVENT_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)