SafeInitDummyHandler

A dummy handler that's returned when the execution context has a DLQ configured and the safe init handler fails during the init phase. This handler is then returned to extract the input arguments and send them to the DLQ, then re-raise the original exception.

Source code in safe_init/dlq.py 
class SafeInitDummyHandler:
    """
    A dummy handler that's returned when the execution context has a DLQ configured and the safe init
    handler fails during the init phase. This handler is then returned to extract the input arguments
    and send them to the DLQ, then re-raise the original exception.
    """

    def __init__(self, exc: Exception) -> None:
        self._exc = exc
        if not isinstance(exc, Exception):
            log_error(
                "Dummy handler initialized with an invalid exception. Will use a custom one instead",
                passed_exc=exc,
            )
            self._exc = Exception(f"Dummy handler initialized with an invalid exception: {exc}")

    def __call__(self, *args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
        if not context_has_dlq():
            log_error(
                (
                    "Dummy handler called without a DLQ configured. This should never happen and is a bug in the safe"
                    " init code. Please report this to the infrastructure team"
                ),
                args=args,
                kwargs=kwargs,
            )
            self._raise()

        log_warning(
            (
                "Application failed during the import phase. Using a dummy handler to fetch the event and send to a "
                "DLQ. The original exception will be re-raised"
            ),
            args=args,
            kwargs=kwargs,
        )
        push_event_to_dlq(*args, **kwargs)
        self._raise()

    def _raise(self) -> None:
        raise self._exc

context_has_dlq()

Returns True if the current execution context has a dead-letter queue configured.

Source code in safe_init/dlq.py 
def context_has_dlq() -> bool:
    """
    Returns True if the current execution context has a dead-letter queue configured.
    """
    return os.environ.get("SAFE_INIT_DLQ") is not None

preload_sqs_client()

Preloads the SQS client to avoid the first invocation penalty.

Source code in safe_init/dlq.py 
def preload_sqs_client() -> None:
    """
    Preloads the SQS client to avoid the first invocation penalty.
    """
    _get_sqs_client()

push_event_to_dlq(*args, **kwargs)

Pushes the given event to the dead-letter queue configured for the current execution context.

Source code in safe_init/dlq.py 
def push_event_to_dlq(*args: Any, **kwargs: Any) -> None:  # noqa: ANN401
    """
    Pushes the given event to the dead-letter queue configured for the current execution context.
    """
    try:
        dlq_url = _get_dlq_url()
        sqs_client = _get_sqs_client()
        sqs_client.send_message(QueueUrl=dlq_url, **_prepare_sqs_message(*args, **kwargs))
    except Exception:
        log_exception("There was an error pushing the event to the DLQ", args=args, kwargs=kwargs)