"""Exceptions module for lambdas.""" from typing import Any, NoReturn from src import status from src.constants import errors class LoggedException(Exception): """Exception to avoid propagation of general notification messages.""" class TranscodingRequestError(LoggedException): """Raised when a transcoding API request fails.""" class OwsAssetsError(LoggedException): """Raised when an OWS Assets API request fails.""" def notify( *, function: str, error_status: str, error_code: str, filename: str, bucket: str, error_params: dict[str, Any] | None = None, input_params: dict[str, Any] | None = None, ) -> str: """Send status notification. Args: function (str): Lambda function name that causes an error. error_status (str): Error status for lambda function. error_code (str): Error code. filename (str): Current processed asset filename. bucket (str): Current processed asset source bucket name. error_params (dict): Additional error params. input_params (dict): Lambda additional input params for reporting. Returns: str: Formatted error message. """ if not error_params: error_params = {} error_message = errors.ERROR_CODE_TO_MESSAGE.get(error_code, "Unknown Error").format(**error_params) status.send_general_status( function=function, status=error_status, filename=filename, error_code=error_code, description=error_message, bucket=bucket, input_params=input_params, ) return error_message def notify_and_raise( *, function: str, error_status: str, error_code: str, filename: str, bucket: str, error_params: dict[str, Any] | None = None, input_params: dict[str, Any] | None = None, ) -> NoReturn: """Send status notification and raise exception. Args: function (str): Lambda function name that causes an error. error_status (str): Error status for lambda function. error_code (str): Error code. filename (str): Current processed asset filename. bucket (str): Current processed asset source bucket name. error_params (dict): Additional error params. input_params (dict): Lambda additional input params for reporting. Raises: LoggedException: Always raised with formatted message. """ error_message = notify( function=function, error_status=error_status, error_code=error_code, filename=filename, bucket=bucket, error_params=error_params, input_params=input_params, ) raise LoggedException(error_message)