This module provides a function for sending a Slack notification with error details.

get_slack_webhook_url()

Returns the Slack webhook URL.

Returns:
  • str

    The Slack webhook URL.

Source code in safe_init/slack.py
def get_slack_webhook_url() -> str:
    """
    Returns the Slack webhook URL.

    Returns:
        The Slack webhook URL.
    """
    ctx_slack_webhook_url = get_contextvar_named("safe_init_slack_webhook_url")
    if ctx_slack_webhook_url:
        return cast(str, ctx_slack_webhook_url)
    slack_webhook_url = os.environ.get("SAFE_INIT_SLACK_WEBHOOK_URL")
    if not slack_webhook_url:
        msg = "SLACK_WEBHOOK_URL environment variable nor global variable are not set"
        raise ValueError(msg)
    return slack_webhook_url

slack_notify(context_message, e, *, handler_name=None, lambda_context=None, sentry_capture_result=None, message_title=None, additional_context=None)

Sends a Slack notification with error details.

Parameters:
  • context_message (str) –

    A message describing the context of the error.

  • e (Exception) –

    The exception that was raised.

  • handler_name (str | None, default: None ) –

    The name of the handler function that raised the exception.

  • lambda_context (object | None, default: None ) –

    The Lambda context object, if running in a Lambda environment.

  • sentry_capture_result (bool | None, default: None ) –

    The result of attempting to capture the exception with Sentry.

  • message_title (str | None, default: None ) –

    The title of the Slack message.

  • additional_context (str | None, default: None ) –

    Additional context to include in the Slack message.

Returns:
  • None

    None.

Source code in safe_init/slack.py
def slack_notify(
    context_message: str,
    e: Exception,
    *,
    handler_name: str | None = None,
    lambda_context: object | None = None,
    sentry_capture_result: bool | None = None,
    message_title: str | None = None,
    additional_context: str | None = None,
) -> None:
    """
    Sends a Slack notification with error details.

    Args:
        context_message: A message describing the context of the error.
        e: The exception that was raised.
        handler_name: The name of the handler function that raised the exception.
        lambda_context: The Lambda context object, if running in a Lambda environment.
        sentry_capture_result: The result of attempting to capture the exception with Sentry.
        message_title: The title of the Slack message.
        additional_context: Additional context to include in the Slack message.

    Returns:
        None.
    """
    try:
        slack_webhook_url = get_slack_webhook_url()
    except ValueError:
        from safe_init.safe_logging import log_warning

        log_warning("Slack webhook URL is not set, skipping Slack notification")
        return

    import requests

    from safe_init.safe_logging import log_error, log_exception

    _lambda_context = None
    if lambda_context:
        if not is_lambda_context(lambda_context):
            log_error("lambda_context is not an instance of LambdaContext")
        else:
            from awslambdaric.lambda_context import LambdaContext

            _lambda_context = cast(LambdaContext, lambda_context)

    env = os.environ.get("SAFE_INIT_ENV", "unknown").upper()

    main_context = []

    if handler_name:
        main_context.append(
            {
                "type": "mrkdwn",
                "text": f":point_right: *Handler:* {handler_name}",
            },
        )
    if _lambda_context:
        main_context.extend(
            [
                {
                    "type": "mrkdwn",
                    "text": f":point_right: *AWS Request ID:* {_lambda_context.aws_request_id}",
                },
                {
                    "type": "mrkdwn",
                    "text": f":point_right: *Lambda function name:* {_lambda_context.function_name}",
                },
            ],
        )

    if dd_wrapped_handler := os.environ.get("DD_LAMBDA_HANDLER"):
        main_context.append(
            {
                "type": "mrkdwn",
                "text": f":point_right: *ddtrace-wrapped:* {dd_wrapped_handler}",
            },
        )

    if (lambda_name := os.environ.get("AWS_LAMBDA_FUNCTION_NAME")) and not _lambda_context:
        main_context.append(
            {
                "type": "mrkdwn",
                "text": f":point_right: *Lambda name:* {lambda_name}",
            },
        )

    if not message_title:
        message_title = f"{'Application' if not _lambda_context else 'Lambda'} execution failed"

    main_context_blocks = []
    if main_context:
        main_context_blocks.append(
            {
                "type": "context",
                "elements": main_context,
            },
        )

    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": f"[{env}] {message_title} :hear_no_evil:",
            },
        },
        {
            "type": "context",
            "elements": [
                {
                    "type": "plain_text",
                    "text": context_message,
                },
            ],
        },
        {
            "type": "divider",
        },
        {
            "type": "section",
            "text": {
                "type": "plain_text",
                "text": f":face_palm: {e}",
            },
        },
        *main_context_blocks,
    ]

    if additional_context:
        blocks.append(
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": additional_context,
                },
            },
        )

    if sentry_capture_result is not None:
        blocks.append(
            {
                "type": "section",
                "text": {
                    "type": "plain_text",
                    "text": (
                        ":rage: There also was an error sending the event to Sentry."
                        if not sentry_capture_result
                        else ":ok_hand: The error has been sent to Sentry."
                    ),
                },
            },
        )

    slack_message = {
        "text": f"[{env}] Safe Init — {message_title} :pleading_face:",
        "attachments": [
            {"color": "#e12424", "blocks": blocks},
        ],
    }

    try:
        response = requests.post(slack_webhook_url, json=slack_message, timeout=15)

        if response.status_code != 200:  # noqa: PLR2004
            log_error(f"Failed to send Slack message: {response.text}", message=slack_message)
    except Exception:
        log_exception("Slack message sending exception", message=slack_message)