Error handling and logging guidelines

handle_error.py

""" logging sample
"""

import json
import os
import logging
import traceback

def lambda_handler(_event, _context):
    """Sample pure Lambda function
    """

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

    try:
        result = _business_logic(_event)

        return {
            "statusCode": 200,
            "body": result,
        }

    except KnownException as ex:
        logger.info(type(ex).__name__ + ": " + str(ex))
        return _format_error(ex)

    except BaseException as ex:
        logger.error(type(ex).__name__ + ": " + str(ex))
        logger.debug(traceback.format_exc())
        return _format_error(ex)

def _business_logic(_):

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

    # Process or object lifecycle events are logged as INFO level as needed.
    logger.info("Business logic begins.")

    # Errors that prevent normal termination of processes such as the following
    # are escalated to the calling function.
    # It is the responsibility of the controller (or the View module called from the controller)
    # to report these errors to the user or system.
    #
    if False:
        raise NotFoundException("Resource is not found.")

    if False:
        raise ValidationException("Arguments is not valid.")

    if False:
        raise ForbiddenException("Resource is not available.")

    if False:
        raise BaseException("Unknown Error.")

    # Debugging logs such as the following are logged as DEBUG level.
    logger.debug("Log for debugging.")

    # Though it is not frequent, a business logic reports critical logs.
    # It doesn't prevent the process from terminating normally,
    # Because it is recovered within the responsibility of business logic.
    # However, it should be reported as WARN level logs to operators and developers
    # as unexpected condition.
    #
    # For example:
    # 1. Batch jobs. If an error occurs in each element, the process must complete
    #   successfully without interrupting the process.
    # 2. Errors in the another services that is the backend system of this. When
    #   it can be hidden from the user (for example, by returning a dummy result from the cache).
    # 
    try:
        _some_sub_process()
    except BaseException:
        logger.warn("An unexpected condition has occurred, but the process continues.")

    logger.info("Business logic has finished.")

    return json.dumps({
        "message": "hello world",
    })

def _some_sub_process():
    raise Exception


DEFAULT_LOG_LEVEL = logging.INFO

def _get_loglevel():

    log_level = os.getenv('LOG_LEVEL')

    if logging.ERROR == log_level:
        return logging.ERROR
    elif logging.WARN == log_level:
        return logging.WARN
    elif logging.INFO == log_level:
        return logging.INFO
    elif logging.DEBUG == log_level:
        return logging.DEBUG
    else:
        return DEFAULT_LOG_LEVEL

class KnownException(Exception):
    """ KnownException
    """

    def __init__(self, message, code):
        super().__init__(message)
        self.code = code

class ValidationException(KnownException):
    """ ValidationException
    """

    def __init__(self, message):
        super().__init__(message, 400)

class NotFoundException(KnownException):
    """ NotFoundException
    """

    def __init__(self, message):
        super().__init__(message, 404)

class ForbiddenException(KnownException):
    """ ForbiddenException
    """

    def __init__(self, message):
        super().__init__(message, 403)

def _format_error(ex):
    if isinstance(ex, KnownException):
        return {
            "statusCode": ex.code,
            "body": json.dumps({
                "message": str(ex),
            }),
        }

    return {
        "statusCode": 500,
        "body": json.dumps({
            "message": "Internal Server Error",
        }),
    }