How-to Guide for Amazon Simple Queue Service (Amazon SQS) notifications

Introduction

Amazon SQS notifier SqsNotifier allows users to push messages to an Amazon SQS Queue using the various on_*_callbacks at both the DAG level and Task level.

Example Code:

from datetime import datetime, timezone
from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.amazon.aws.notifications.sqs import send_sqs_notification

dag_failure_sqs_notification = send_sqs_notification(
    aws_conn_id="aws_default",
    queue_url="https://sqs.eu-west-1.amazonaws.com/123456789098/MyQueue",
    message_body="The DAG {{ dag.dag_id }} failed",
)
task_failure_sqs_notification = send_sqs_notification(
    aws_conn_id="aws_default",
    region_name="eu-west-1",
    queue_url="https://sqs.eu-west-1.amazonaws.com/123456789098/MyQueue",
    message_body="The task {{ ti.task_id }} failed",
)

with DAG(
    dag_id="mydag",
    schedule="@once",
    start_date=datetime(2023, 1, 1, tzinfo=timezone.utc),
    on_failure_callback=[dag_failure_sqs_notification],
    catchup=False,
):
    BashOperator(task_id="mytask", on_failure_callback=[task_failure_sqs_notification], bash_command="fail")

Was this entry helpful?