Source code for ecstask.resources.utils

"""
This module contains common utilities.
"""
import copy
import logging
import uuid
from http import HTTPStatus

from ecstask.resources import constants
from pymorri import exceptions


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


def generate_correlation_id() -> str:
    """
    Generate a new correlation id.
    """
    return f"strpick-pick-task-{uuid.uuid4().hex}"

def retry_if_mori_exc_and_not_404(exc: Exception) -> bool:
    """
    Return True if we should retry, False otherwise.
    """
    if not isinstance(exc, exceptions.MoriException):
        return False

    if exc.http_status_code != HTTPStatus.NOT_FOUND:
        return True

    raise exc


def validate_kafka_message(kafka_message: dict,
                           store_validator, min_validator) -> tuple[dict, dict, str]:
    """
    Validate the message that was received from Kafka
    before creating the payload that will be sent to the stock transfer service.

    Args:
        kafka_message: The message that was received from Kafka
        store_validator: The wrapper of location API
        min_validator: The wrapper of the product API

    Returns:
        tuple: The Kafka message stripped of invalid store and MINs,
                the message to be written to DLQ and status message
    """
    dlq_message = {"original_message": kafka_message,
                   "invalid_store_or_mins": {"store": "", "mins": []}}
    if kafka_message == {}:
        logger.warning("Kafka message is empty")
        return {}, dlq_message, constants.FOUND_EMPTY_MESSAGE

    new_message = copy.deepcopy(kafka_message)

    store_id, value = kafka_message.copy().popitem()
    items_dict = new_message.get(store_id, {}).get("items", {})
    if store_id == "" or items_dict == {}:
        logger.warning("Kafka message has no store or no mins")
        return {}, dlq_message, constants.FOUND_PARTIAL_MESSAGE

    if validate_store(store_validator, store_id, dlq_message):
        status = validate_mins(min_validator, items_dict, dlq_message)
    else:
        status = constants.FOUND_INVALID_STORE
        new_message = {}

    if items_dict == {}:
        new_message = {}

    return new_message, dlq_message, status


def validate_store(store_validator, store_id: str, dlq_message: dict) -> bool:
    """Validate the given store."""
    if not store_validator.validate_store(store_id):
        logger.warning("Kafka message has invalid store")
        dlq_message["invalid_store_or_mins"]["store"] = store_id
        return False
    return True


def validate_mins(min_validator, items_dict: dict, dlq_message: dict) -> str:
    """Validate the given MINs."""
    _, invalid_mins = min_validator.return_valid_and_invalid_mins(list(items_dict.keys()))
    if len(invalid_mins) == 0:
        dlq_message.clear()
        return constants.NO_VALIDATION_ERRORS
    logger.warning("Kafka message has invalid mins")
    dlq_message["invalid_store_or_mins"]["mins"] = invalid_mins
    for item in invalid_mins:
        items_dict.pop(item)
    return constants.FOUND_INVALID_MINS