Add a new Amazon Simple Email Service Notifier

[ferruzzi]

What it says in the title.

Manually tested using the following Dags (you will need an AWS account and use the SES console to verify the required email addresses)

from datetime import datetime, timedelta

from airflow.providers.amazon.aws.hooks.ses import SesHook
from airflow.providers.amazon.aws.notifications.ses import SesNotifier
from airflow.sdk import task, DAG
from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference, AsyncCallback

PAST_DATE = datetime(1980, 8, 10, 2)
MSG_PREFIX = "SES Testing Rnd 3:"

DEFAULTS = {
    "mail_from": << REDACTED >>,
    "to": << REDACTED >>,
    "html_content": "message body goes here"
}


def _ses_callback(context):
    SesHook().send_email(**DEFAULTS, subject=f"{MSG_PREFIX} Hook as Callback -- {context['dag_run'].dag_id}")


@task
def send_ses_email(subject):
    SesHook().send_email(**DEFAULTS, subject=subject)


@task.bash(task_id='sleep_task')
def sleep_n_secs(seconds):
    return f'sleep {seconds}'


with DAG(
    "ses_hook_via_taskflow",
    tags=["ses"]
):
    send_ses_email(subject=f"{MSG_PREFIX} Taskflow -- {{{{ dag_run.dag_id }}}} ")

with DAG(
    "ses_callback_on_success",
    on_success_callback=_ses_callback,
    tags=["ses"],
):
    sleep_n_secs(1)

with DAG(
    "ses_notifier_as_callback",
    on_success_callback=SesNotifier(**DEFAULTS, subject=f"{MSG_PREFIX} Sync Notifier as callback -- {{{{ dag_run.dag_id }}}} "),
    tags=["ses"],
):
    sleep_n_secs(1)


with DAG(
    "ses_notifier_as_deadline",
    deadline=DeadlineAlert(
        reference=DeadlineReference.FIXED_DATETIME(PAST_DATE),
        interval=timedelta(0),
        callback=AsyncCallback(
            SesNotifier,
            kwargs={**DEFAULTS, "subject": f"{MSG_PREFIX} Async Notifier as Deadline -- {{{{ dag_run.dag_id }}}} "}
        ),
    ),
    tags=["ses"],
):
    sleep_n_secs(1)

and proof that the emails are delivered and templates work:

cc; @ramitkataria @seanghaeli

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[vincbeck]

Just one comment but overall, looks good 👍

[zachliu]

silly question, why aren't we _build_email_content as in the SmtpNotifier

airflow/providers/smtp/src/airflow/providers/smtp/notifications/smtp.py

Line 118 in 1c98f57

def _build_email_content(self, smtp: SmtpHook, context: Context):

so that they can be drop-in replacement for each other?

[ferruzzi]

Silly answer, 🤷

No particular reason. Didn't think of it?

[zachliu]

lol cool, i'll submit a pr 😜