This is an automated email from the ASF dual-hosted git repository.
dabla pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 754ba31050e Add support for sending emails through Microsoft Graph
(#71565)
754ba31050e is described below
commit 754ba31050ea1253c8d2f72ffa7ec5d7f442da16
Author: rjgoyln <[email protected]>
AuthorDate: Wed Sep 9 14:27:32 2026 +0800
Add support for sending emails through Microsoft Graph (#71565)
Sending mail from an Office 365 or Outlook cloud mailbox is expected to go
through Microsoft Graph, and with EWS gone the SMTP relay into those
mailboxes
is the last legacy path standing. Airflow could only reach them over SMTP,
so a
deployment on a Microsoft tenant had no supported way to alert on task
failures
from its own mailbox.
---
providers/microsoft/azure/docs/email-backend.rst | 72 +++++
providers/microsoft/azure/docs/index.rst | 2 +
.../microsoft/azure/docs/notifications/index.rst | 27 ++
.../microsoft/azure/docs/notifications/msgraph.rst | 65 +++++
providers/microsoft/azure/provider.yaml | 6 +
.../providers/microsoft/azure/get_provider_info.py | 2 +
.../providers/microsoft/azure/hooks/msgraph.py | 274 ++++++++++++++++++
.../microsoft/azure/notifications/__init__.py | 16 ++
.../microsoft/azure/notifications/msgraph.py | 131 +++++++++
.../unit/microsoft/azure/hooks/test_msgraph.py | 309 +++++++++++++++++++++
.../unit/microsoft/azure/notifications/__init__.py | 16 ++
.../microsoft/azure/notifications/test_msgraph.py | 113 ++++++++
12 files changed, 1033 insertions(+)
diff --git a/providers/microsoft/azure/docs/email-backend.rst
b/providers/microsoft/azure/docs/email-backend.rst
new file mode 100644
index 00000000000..bb39c266db2
--- /dev/null
+++ b/providers/microsoft/azure/docs/email-backend.rst
@@ -0,0 +1,72 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _email-configuration-msgraph:
+
+Send email using Microsoft Graph
+================================
+
+Airflow can be configured to send e-mail from an Office 365 or Outlook cloud
mailbox using
+`Microsoft Graph
<https://learn.microsoft.com/en-us/graph/api/user-sendmail>`__ as the
+``email_backend``, so that task callback emails (success, failure, retry) go
out through Graph instead
+of SMTP. See :doc:`apache-airflow:howto/email-config` for the general Airflow
email configuration this
+page builds on.
+
+.. note::
+
+ If you instead want to send an email from a task callback or a deadline
alert, use the
+ :ref:`MSGraphNotifier <howto/notifier:MSGraphNotifier>` documented in
:doc:`notifications/msgraph`.
+
+Follow the steps below to enable it:
+
+1. Install the ``microsoft-azure`` provider as part of your Airflow
installation:
+
+ .. code-block:: bash
+
+ pip install 'apache-airflow[microsoft-azure]'
+
+2. Update the ``[email]`` section in ``airflow.cfg``:
+
+ .. code-block:: ini
+
+ [email]
+ email_backend =
airflow.providers.microsoft.azure.hooks.msgraph.send_email
+ email_conn_id = msgraph_default
+ from_email = From email <[email protected]>
+
+ Equivalent environment variables look like:
+
+ .. code-block:: sh
+
+
AIRFLOW__EMAIL__EMAIL_BACKEND=airflow.providers.microsoft.azure.hooks.msgraph.send_email
+ AIRFLOW__EMAIL__EMAIL_CONN_ID=msgraph_default
+ [email protected]
+
+ ``from_email`` is required and selects the mailbox the message is sent from.
+
+3. Create a connection called ``msgraph_default``, or choose a custom
connection name and set it in
+ ``email_conn_id``, of type ``Microsoft Graph API``. See
+ :ref:`Microsoft Graph API Connection <howto/connection:msgraph>` for how to
configure it.
+
+The app registration behind the connection needs the ``Mail.Send`` permission.
When application
+permissions are used, an administrator has to grant the application access to
the sending mailbox, for
+example by scoping it with an
+`application access policy
<https://learn.microsoft.com/en-us/graph/auth-limit-mailbox-access>`__.
+
+Attachments are sent inline in the request body, which Microsoft Graph limits
to 4 MB in total.
+Attachments adding up to more than 3 MB are rejected with a ``ValueError``,
and have to be uploaded to
+a draft message with an upload session instead.
diff --git a/providers/microsoft/azure/docs/index.rst
b/providers/microsoft/azure/docs/index.rst
index c3134d9b294..8b87081166b 100644
--- a/providers/microsoft/azure/docs/index.rst
+++ b/providers/microsoft/azure/docs/index.rst
@@ -35,7 +35,9 @@
:caption: Guides
Connection types <connections/index>
+ Email backend <email-backend>
Message queues <message-queues/index>
+ Notifications <notifications/index>
Operators <operators/index>
Transfers <transfer/index>
Filesystems <filesystems/index>
diff --git a/providers/microsoft/azure/docs/notifications/index.rst
b/providers/microsoft/azure/docs/notifications/index.rst
new file mode 100644
index 00000000000..f884155bdfa
--- /dev/null
+++ b/providers/microsoft/azure/docs/notifications/index.rst
@@ -0,0 +1,27 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+
+
+Microsoft Azure Notifications
+=============================
+
+.. toctree::
+ :maxdepth: 1
+ :glob:
+
+ *
diff --git a/providers/microsoft/azure/docs/notifications/msgraph.rst
b/providers/microsoft/azure/docs/notifications/msgraph.rst
new file mode 100644
index 00000000000..b597b434288
--- /dev/null
+++ b/providers/microsoft/azure/docs/notifications/msgraph.rst
@@ -0,0 +1,65 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/notifier:MSGraphNotifier:
+
+Microsoft Graph Notifier
+========================
+
+`Microsoft Graph <https://learn.microsoft.com/en-us/graph/overview>`__ is the
recommended way to send mail
+from an Office 365 or Outlook cloud mailbox programmatically.
+:class:`~airflow.providers.microsoft.azure.notifications.msgraph.MSGraphNotifier`
sends an email through the
+`sendMail <https://learn.microsoft.com/en-us/graph/api/user-sendmail>`__
endpoint over a
+:ref:`Microsoft Graph API connection <howto/connection:msgraph>`.
+
+The permissions the app registration needs and the limit on attachments passed
through ``files`` are the
+same as for the email backend, and are described in :doc:`../email-backend`.
That page also covers routing
+the ``email_on_failure`` and ``email_on_retry`` emails through Microsoft Graph.
+
+Sending a notification
+----------------------
+
+.. code-block:: python
+
+ from airflow.providers.microsoft.azure.notifications.msgraph import
MSGraphNotifier
+
+ dag_failure_notification = MSGraphNotifier(
+ from_email="[email protected]",
+ to="[email protected]",
+ subject="Dag {{ dag.dag_id }} failed",
+ html_content="Task {{ ti.task_id }} failed on {{ ds }}",
+ )
+
+ with DAG(
+ dag_id="mydag",
+ schedule="@daily",
+ on_failure_callback=[dag_failure_notification],
+ ):
+ BashOperator(
+ task_id="mytask",
+ bash_command="fail",
+ on_failure_callback=[
+ MSGraphNotifier(
+ from_email="[email protected]",
+ to="[email protected]",
+ subject="Task {{ ti.task_id }} failed",
+ html_content="Have a look at the logs of {{ ti.log_url }}",
+ )
+ ],
+ )
+
+When ``from_email`` is omitted, the ``[email] from_email`` configuration
option is used instead.
diff --git a/providers/microsoft/azure/provider.yaml
b/providers/microsoft/azure/provider.yaml
index bb5c770251f..954082aa228 100644
--- a/providers/microsoft/azure/provider.yaml
+++ b/providers/microsoft/azure/provider.yaml
@@ -406,6 +406,9 @@ transfers:
how-to-guide:
/docs/apache-airflow-providers-microsoft-azure/transfer/gcs_to_wasb.rst
python-module: airflow.providers.microsoft.azure.transfers.gcs_to_wasb
+notifications:
+ - airflow.providers.microsoft.azure.notifications.msgraph.MSGraphNotifier
+
connection-types:
- hook-class-name:
airflow.providers.microsoft.azure.hooks.analysis_services.AzureAnalysisServicesHook
hook-name: "Azure Analysis Services"
@@ -1022,6 +1025,9 @@ connection-types:
secrets-backends:
- airflow.providers.microsoft.azure.secrets.key_vault.AzureKeyVaultBackend
+email-backends:
+ - airflow.providers.microsoft.azure.hooks.msgraph.send_email
+
logging:
- airflow.providers.microsoft.azure.log.wasb_task_handler.WasbTaskHandler
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py
index f45dd5af9a6..5a22d603e63 100644
---
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py
@@ -403,6 +403,7 @@ def get_provider_info():
"python-module":
"airflow.providers.microsoft.azure.transfers.gcs_to_wasb",
},
],
+ "notifications":
["airflow.providers.microsoft.azure.notifications.msgraph.MSGraphNotifier"],
"connection-types": [
{
"hook-class-name":
"airflow.providers.microsoft.azure.hooks.analysis_services.AzureAnalysisServicesHook",
@@ -997,6 +998,7 @@ def get_provider_info():
},
],
"secrets-backends":
["airflow.providers.microsoft.azure.secrets.key_vault.AzureKeyVaultBackend"],
+ "email-backends":
["airflow.providers.microsoft.azure.hooks.msgraph.send_email"],
"logging":
["airflow.providers.microsoft.azure.log.wasb_task_handler.WasbTaskHandler"],
"remote-logging": [
{
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py
index 0a9cca43d0d..a389c8725be 100644
---
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py
@@ -15,18 +15,31 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+"""
+This module contains a Microsoft Graph API hook and the email backend built on
it.
+
+.. spelling:word-list::
+
+ dryrun
+"""
+
from __future__ import annotations
import asyncio
import inspect
import json
+import mimetypes
+import re
import warnings
from ast import literal_eval
+from base64 import b64encode
from collections.abc import Callable
from contextlib import suppress
+from email.utils import parseaddr
from http import HTTPStatus
from io import BytesIO
from json import JSONDecodeError
+from pathlib import Path
from types import TracebackType
from typing import TYPE_CHECKING, Any, cast
from urllib.parse import quote, urljoin, urlparse
@@ -55,6 +68,8 @@ from airflow.providers.common.compat.connection import
get_async_connection
from airflow.providers.common.compat.sdk import AirflowException,
AirflowNotFoundException, BaseHook, redact
if TYPE_CHECKING:
+ from collections.abc import Iterable
+
from azure.core.pipeline.transport._requests_basic import RequestsTransport
from kiota_abstractions.authentication import
BaseBearerTokenAuthenticationProvider
from kiota_abstractions.request_adapter import RequestAdapter
@@ -812,3 +827,262 @@ class KiotaRequestAdapterHook(BaseHook):
"4XX": APIError, # type: ignore
"5XX": APIError, # type: ignore
}
+
+
+class MSGraphMailHook(KiotaRequestAdapterHook):
+ """
+ Send mail from an Office 365 mailbox through the Microsoft Graph
``sendMail`` endpoint.
+
+ The application registration behind the connection needs the ``Mail.Send``
permission, and with
+ application permissions an administrator has to grant it access to the
sending mailbox.
+
+ https://learn.microsoft.com/en-us/graph/api/user-sendmail
+
+ :param conn_id: The :ref:`Microsoft Graph API connection id
<howto/connection:msgraph>`.
+ :param timeout: The HTTP timeout being used by the KiotaRequestAdapter
(default is None).
+ When no timeout is specified or set to None then no HTTP timeout is
applied on each request.
+ :param proxies: A Dict defining the HTTP proxies to be used (default is
None).
+ :param host: The host to be used (default is
"https://graph.microsoft.com";).
+ :param scopes: The scopes to be used (default is
["https://graph.microsoft.com/.default";]).
+ :param api_version: The API version of the Microsoft Graph API to be used
(default is v1).
+ """
+
+ # Microsoft Graph documents 3 MB as the largest content that can ride
inline on a message.
+ # Anything bigger has to go through an upload session on a draft message
instead.
+ MAX_ATTACHMENTS_SIZE = 3 * 1024 * 1024
+
+ @staticmethod
+ def extract_email_addresses(addresses: str | Iterable[str] | None) ->
list[str]:
+ """Split a comma or semicolon separated string, or an iterable, into a
list of addresses."""
+ if not addresses:
+ return []
+ if isinstance(addresses, str):
+ addresses = re.split(r"\s*[,;]\s*", addresses)
+ return [address for address in addresses if address]
+
+ @staticmethod
+ def extract_sender(from_email: str | None) -> str:
+ """Extract the bare address of the mailbox to send from."""
+ # The mailbox is addressed in the request path, so the bare address is
needed even though
+ # ``[email] from_email`` is conventionally configured as "Display name
<[email protected]>".
+ sender = parseaddr(from_email or "")[1]
+ if not sender:
+ raise ValueError(
+ f"A `from_email` holding the mailbox to send from is required,
got {from_email!r}."
+ )
+ return sender
+
+ @classmethod
+ def build_recipients(cls, addresses: str | Iterable[str] | None) ->
list[dict[str, Any]]:
+ """Build the Microsoft Graph recipient representation for the given
addresses."""
+ return [{"emailAddress": {"address": address}} for address in
cls.extract_email_addresses(addresses)]
+
+ @classmethod
+ def build_attachments(cls, files: Iterable[str] | None) -> list[dict[str,
Any]]:
+ """Read each file from disk and build its Microsoft Graph
``fileAttachment`` representation."""
+ attachments = []
+ total_size = 0
+ for file in files or []:
+ path = Path(file)
+ content = path.read_bytes()
+ total_size += len(content)
+ if total_size > cls.MAX_ATTACHMENTS_SIZE:
+ raise ValueError(
+ f"The attachments add up to at least {total_size} bytes,
which is more than the "
+ f"{cls.MAX_ATTACHMENTS_SIZE} bytes Microsoft Graph accepts
on a sendMail request. "
+ f"Upload larger files to a draft message with an upload
session instead."
+ )
+ attachments.append(
+ {
+ "@odata.type": "#microsoft.graph.fileAttachment",
+ "name": path.name,
+ "contentType": mimetypes.guess_type(path.name)[0] or
"application/octet-stream",
+ "contentBytes": b64encode(content).decode("ascii"),
+ }
+ )
+ return attachments
+
+ @classmethod
+ def build_message(
+ cls,
+ to: str | Iterable[str],
+ subject: str,
+ html_content: str,
+ files: Iterable[str] | None = None,
+ cc: str | Iterable[str] | None = None,
+ bcc: str | Iterable[str] | None = None,
+ custom_headers: dict[str, Any] | None = None,
+ ) -> dict[str, Any]:
+ """Build the Microsoft Graph message body for the given email
fields."""
+ recipients = cls.build_recipients(to)
+ if not recipients:
+ raise ValueError("No recipients were resolved from the `to`
argument.")
+
+ message: dict[str, Any] = {
+ "subject": subject,
+ "body": {"contentType": "HTML", "content": html_content},
+ "toRecipients": recipients,
+ }
+ if cc:
+ message["ccRecipients"] = cls.build_recipients(cc)
+ if bcc:
+ message["bccRecipients"] = cls.build_recipients(bcc)
+ if files:
+ message["attachments"] = cls.build_attachments(files)
+ if custom_headers:
+ # Microsoft Graph rejects custom header names that are not
prefixed with "x-".
+ message["internetMessageHeaders"] = [
+ {"name": name, "value": str(value)} for name, value in
custom_headers.items()
+ ]
+ return message
+
+ async def asend_email(
+ self,
+ from_email: str,
+ to: str | Iterable[str],
+ subject: str,
+ html_content: str,
+ files: Iterable[str] | None = None,
+ cc: str | Iterable[str] | None = None,
+ bcc: str | Iterable[str] | None = None,
+ custom_headers: dict[str, Any] | None = None,
+ save_to_sent_items: bool = True,
+ dryrun: bool = False,
+ ) -> None:
+ """
+ Send an email from the ``from_email`` mailbox (async).
+
+ :param from_email: The mailbox the message is sent from.
+ :param to: Recipient email address or list of addresses.
+ :param subject: Email subject.
+ :param html_content: Email body in HTML format.
+ :param files: List of file paths to attach to the email.
+ :param cc: Carbon copy recipient email address or list of addresses.
+ :param bcc: Blind carbon copy recipient email address or list of
addresses.
+ :param custom_headers: Custom internet message headers, whose names
have to start with "x-".
+ :param save_to_sent_items: Whether the message is saved in the
mailbox's Sent Items folder.
+ :param dryrun: If True, the message is prepared but not sent.
+ """
+ sender = self.extract_sender(from_email)
+
+ message = self.build_message(
+ to=to,
+ subject=subject,
+ html_content=html_content,
+ files=files,
+ cc=cc,
+ bcc=bcc,
+ custom_headers=custom_headers,
+ )
+
+ if dryrun:
+ self.log.info("Dry run, not sending email with subject %r to %s",
subject, to)
+ return
+
+ await self.run(
+ url="users/{user_id}/sendMail",
+ path_parameters={"user_id": sender},
+ method="POST",
+ data={"message": message, "saveToSentItems": save_to_sent_items},
+ )
+
+ def send_email(
+ self,
+ from_email: str,
+ to: str | Iterable[str],
+ subject: str,
+ html_content: str,
+ files: Iterable[str] | None = None,
+ cc: str | Iterable[str] | None = None,
+ bcc: str | Iterable[str] | None = None,
+ custom_headers: dict[str, Any] | None = None,
+ save_to_sent_items: bool = True,
+ dryrun: bool = False,
+ ) -> None:
+ """
+ Send an email from the ``from_email`` mailbox.
+
+ :param from_email: The mailbox the message is sent from.
+ :param to: Recipient email address or list of addresses.
+ :param subject: Email subject.
+ :param html_content: Email body in HTML format.
+ :param files: List of file paths to attach to the email.
+ :param cc: Carbon copy recipient email address or list of addresses.
+ :param bcc: Blind carbon copy recipient email address or list of
addresses.
+ :param custom_headers: Custom internet message headers, whose names
have to start with "x-".
+ :param save_to_sent_items: Whether the message is saved in the
mailbox's Sent Items folder.
+ :param dryrun: If True, the message is prepared but not sent.
+ """
+ try:
+ asyncio.get_running_loop()
+ except RuntimeError:
+ pass
+ else:
+ raise RuntimeError(
+ "send_email cannot be called from a running event loop, await
asend_email instead."
+ )
+
+ async def send_and_close() -> None:
+ try:
+ await self.asend_email(
+ from_email=from_email,
+ to=to,
+ subject=subject,
+ html_content=html_content,
+ files=files,
+ cc=cc,
+ bcc=bcc,
+ custom_headers=custom_headers,
+ save_to_sent_items=save_to_sent_items,
+ dryrun=dryrun,
+ )
+ finally:
+ # asyncio.run tears down the event loop that the request
adapter is bound to, so the
+ # cached adapter is unusable by the time the next email is
sent. Close it here,
+ # while the loop that owns its sockets is still running.
+ await self.close()
+
+ asyncio.run(send_and_close())
+
+
+def send_email(
+ to: str | Iterable[str],
+ subject: str,
+ html_content: str,
+ files: list[str] | None = None,
+ dryrun: bool = False,
+ cc: str | Iterable[str] | None = None,
+ bcc: str | Iterable[str] | None = None,
+ mime_subtype: str = "mixed",
+ mime_charset: str = "utf-8",
+ conn_id: str | None = None,
+ from_email: str | None = None,
+ custom_headers: dict[str, Any] | None = None,
+ **kwargs,
+) -> None:
+ """
+ Email backend for Microsoft Graph.
+
+ .. note::
+ For more information, see :ref:`email-configuration-msgraph`
+ """
+ if not from_email:
+ raise ValueError(
+ "The `from_email` configuration has to be set for the Microsoft
Graph emailer, as it "
+ "determines which mailbox the message is sent from."
+ )
+
+ # ``mime_subtype`` and ``mime_charset`` are part of the email backend
contract but have no
+ # counterpart here: Microsoft Graph composes the MIME message itself from
the JSON payload.
+ hook = MSGraphMailHook(conn_id=conn_id or
MSGraphMailHook.default_conn_name)
+ hook.send_email(
+ from_email=from_email,
+ to=to,
+ subject=subject,
+ html_content=html_content,
+ files=files,
+ cc=cc,
+ bcc=bcc,
+ custom_headers=custom_headers,
+ dryrun=dryrun,
+ )
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/notifications/__init__.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/notifications/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/notifications/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/notifications/msgraph.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/notifications/msgraph.py
new file mode 100644
index 00000000000..6058e548707
--- /dev/null
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/notifications/msgraph.py
@@ -0,0 +1,131 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from functools import cached_property
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.compat.notifier import BaseNotifier
+from airflow.providers.common.compat.sdk import conf
+from airflow.providers.microsoft.azure.hooks.msgraph import MSGraphMailHook
+from airflow.providers.microsoft.azure.version_compat import AIRFLOW_V_3_1_PLUS
+
+if TYPE_CHECKING:
+ from collections.abc import Iterable
+
+ from airflow.providers.common.compat.sdk import Context
+
+
+class MSGraphNotifier(BaseNotifier):
+ """
+ Send an email from an Office 365 mailbox through the Microsoft Graph
``sendMail`` endpoint.
+
+ .. code-block:: python
+
+ EmptyOperator(
+ task_id="task",
+ on_failure_callback=MSGraphNotifier(
+ from_email="[email protected]",
+ to="[email protected]",
+ subject="Task {{ ti.task_id }} failed",
+ html_content="Dag {{ ti.dag_id }} failed on {{ ds }}",
+ ),
+ )
+
+ :param to: Recipient email address or list of addresses.
+ :param subject: Email subject.
+ :param html_content: Email body in HTML format.
+ :param from_email: The mailbox the message is sent from. Falls back to the
``[email] from_email``
+ configuration option.
+ :param files: List of file paths to attach to the email.
+ :param cc: Carbon copy recipient email address or list of addresses.
+ :param bcc: Blind carbon copy recipient email address or list of addresses.
+ :param custom_headers: Custom internet message headers, whose names have
to start with "x-".
+ :param conn_id: The :ref:`Microsoft Graph API connection id
<howto/connection:msgraph>`.
+ :param save_to_sent_items: Whether the message is saved in the mailbox's
Sent Items folder.
+ """
+
+ template_fields = (
+ "from_email",
+ "to",
+ "subject",
+ "html_content",
+ "files",
+ "cc",
+ "bcc",
+ "custom_headers",
+ )
+
+ def __init__(
+ self,
+ to: str | Iterable[str],
+ subject: str,
+ html_content: str,
+ from_email: str | None = None,
+ files: list[str] | None = None,
+ cc: str | Iterable[str] | None = None,
+ bcc: str | Iterable[str] | None = None,
+ custom_headers: dict[str, Any] | None = None,
+ conn_id: str = MSGraphMailHook.default_conn_name,
+ save_to_sent_items: bool = True,
+ **kwargs,
+ ):
+ if AIRFLOW_V_3_1_PLUS:
+ # Support for passing context was added in 3.1.0
+ super().__init__(**kwargs)
+ else:
+ super().__init__()
+ self.to = to
+ self.subject = subject
+ self.html_content = html_content
+ self.from_email = from_email or conf.get("email", "from_email",
fallback=None)
+ self.files = files
+ self.cc = cc
+ self.bcc = bcc
+ self.custom_headers = custom_headers
+ self.conn_id = conn_id
+ self.save_to_sent_items = save_to_sent_items
+
+ @cached_property
+ def hook(self) -> MSGraphMailHook:
+ """Microsoft Graph mail hook."""
+ return MSGraphMailHook(conn_id=self.conn_id)
+
+ def notify(self, context: Context) -> None:
+ """Send an email through Microsoft Graph."""
+ self.hook.send_email(**self._build_email_arguments())
+
+ async def async_notify(self, context: Context) -> None:
+ """Send an email through Microsoft Graph (async)."""
+ await self.hook.asend_email(**self._build_email_arguments())
+
+ def _build_email_arguments(self) -> dict[str, Any]:
+ return {
+ "from_email": self.from_email,
+ "to": self.to,
+ "subject": self.subject,
+ "html_content": self.html_content,
+ "files": self.files,
+ "cc": self.cc,
+ "bcc": self.bcc,
+ "custom_headers": self.custom_headers,
+ "save_to_sent_items": self.save_to_sent_items,
+ }
+
+
+send_msgraph_notification = MSGraphNotifier
diff --git
a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py
index c2fe69ff9f3..1d0fa23b697 100644
--- a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py
+++ b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py
@@ -18,6 +18,8 @@ from __future__ import annotations
import asyncio
import inspect
+import json
+from base64 import b64encode
from contextlib import AbstractAsyncContextManager
from json import JSONDecodeError
from os.path import dirname
@@ -27,6 +29,7 @@ from unittest.mock import AsyncMock, Mock, patch
import pytest
from httpx import AsyncClient, Response
from httpx._utils import URLPattern
+from kiota_abstractions.method import Method
from kiota_abstractions.request_information import RequestInformation
from kiota_http.httpx_request_adapter import HttpxRequestAdapter
from kiota_serialization_json.json_parse_node import JsonParseNode
@@ -40,7 +43,9 @@ from airflow.providers.microsoft.azure.hooks.msgraph import (
CachedAsyncTokenCredential,
DefaultResponseHandler,
KiotaRequestAdapterHook,
+ MSGraphMailHook,
execute_callable,
+ send_email,
)
from tests_common.test_utils.file_loading import load_file_from_resources,
load_json_from_resources
@@ -978,3 +983,307 @@ class TestResponseHandler:
raise AirflowProviderDeprecationWarning(
f"Check TODO's to remove obsolete code in get_proxies
method:\n\r\n\r\t\t\t{method_source}"
)
+
+
+class TestMSGraphMailHook:
+ FROM_EMAIL = "[email protected]"
+
+ @staticmethod
+ def get_request_information(mock_get_http_response) -> RequestInformation:
+ return mock_get_http_response.call_args.args[0]
+
+ @pytest.mark.parametrize(
+ ("addresses", "expected"),
+ (
+ pytest.param(None, [], id="none"),
+ pytest.param("", [], id="empty-string"),
+ pytest.param("[email protected]", ["[email protected]"],
id="single-address"),
+ pytest.param(
+ "[email protected],[email protected]",
+ ["[email protected]", "[email protected]"],
+ id="comma-separated",
+ ),
+ pytest.param(
+ "[email protected] ; [email protected]",
+ ["[email protected]", "[email protected]"],
+ id="semicolon-separated",
+ ),
+ pytest.param(
+ ["[email protected]", "[email protected]"],
+ ["[email protected]", "[email protected]"],
+ id="list",
+ ),
+ ),
+ )
+ def test_extract_email_addresses(self, addresses, expected):
+ assert MSGraphMailHook.extract_email_addresses(addresses) == expected
+
+ @pytest.mark.parametrize(
+ ("from_email", "expected"),
+ (
+ pytest.param(FROM_EMAIL, FROM_EMAIL, id="bare-address"),
+ pytest.param(f"Airflow alerts <{FROM_EMAIL}>", FROM_EMAIL,
id="with-display-name"),
+ ),
+ )
+ def test_extract_sender(self, from_email, expected):
+ assert MSGraphMailHook.extract_sender(from_email) == expected
+
+ @pytest.mark.parametrize(
+ "from_email",
+ (
+ pytest.param(None, id="none"),
+ pytest.param("", id="empty"),
+ pytest.param("Airflow alerts <>",
id="display-name-without-an-address"),
+ ),
+ )
+ def test_extract_sender_without_a_mailbox(self, from_email):
+ with pytest.raises(ValueError, match="mailbox to send from is
required"):
+ MSGraphMailHook.extract_sender(from_email)
+
+ def test_build_message(self, tmp_path):
+ attachment = tmp_path / "report.csv"
+ attachment.write_bytes(b"a,b\n1,2\n")
+
+ actual = MSGraphMailHook.build_message(
+ to="[email protected],[email protected]",
+ subject="Airflow alert",
+ html_content="<b>Something</b> happened",
+ files=[attachment.as_posix()],
+ cc="[email protected]",
+ bcc=["[email protected]"],
+ custom_headers={"x-custom": 1},
+ )
+
+ assert actual == {
+ "subject": "Airflow alert",
+ "body": {"contentType": "HTML", "content": "<b>Something</b>
happened"},
+ "toRecipients": [
+ {"emailAddress": {"address": "[email protected]"}},
+ {"emailAddress": {"address": "[email protected]"}},
+ ],
+ "ccRecipients": [{"emailAddress": {"address":
"[email protected]"}}],
+ "bccRecipients": [{"emailAddress": {"address":
"[email protected]"}}],
+ "attachments": [
+ {
+ "@odata.type": "#microsoft.graph.fileAttachment",
+ "name": "report.csv",
+ "contentType": "text/csv",
+ "contentBytes": b64encode(b"a,b\n1,2\n").decode("ascii"),
+ }
+ ],
+ "internetMessageHeaders": [{"name": "x-custom", "value": "1"}],
+ }
+
+ def test_build_message_without_recipients(self):
+ with pytest.raises(ValueError, match="No recipients"):
+ MSGraphMailHook.build_message(to=[], subject="Airflow alert",
html_content="Something happened")
+
+ def test_build_attachments(self, tmp_path):
+ attachment = tmp_path / "report.csv"
+ attachment.write_bytes(b"a,b\n1,2\n")
+
+ assert MSGraphMailHook.build_attachments([attachment.as_posix()]) == [
+ {
+ "@odata.type": "#microsoft.graph.fileAttachment",
+ "name": "report.csv",
+ "contentType": "text/csv",
+ "contentBytes": b64encode(b"a,b\n1,2\n").decode("ascii"),
+ }
+ ]
+
+ def test_build_attachments_when_the_attachments_are_too_large(self,
tmp_path):
+ first = tmp_path / "first.csv"
+ first.write_bytes(b"x" * MSGraphMailHook.MAX_ATTACHMENTS_SIZE)
+ second = tmp_path / "second.csv"
+ second.write_bytes(b"x")
+
+ with pytest.raises(ValueError, match="add up to at least 3145729
bytes"):
+ MSGraphMailHook.build_attachments([first.as_posix(),
second.as_posix()])
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ "from_email",
+ (
+ pytest.param(FROM_EMAIL, id="bare-address"),
+ pytest.param(f"Airflow alerts <{FROM_EMAIL}>",
id="with-display-name"),
+ ),
+ )
+ async def test_asend_email(self, from_email):
+ with patch_hook_and_request_adapter(mock_json_response(202)) as mocks:
+ mock_get_http_response = mocks[-1]
+ hook = MSGraphMailHook(conn_id="msgraph_api")
+
+ await hook.asend_email(
+ from_email=from_email,
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ save_to_sent_items=False,
+ )
+
+ request_information =
self.get_request_information(mock_get_http_response)
+ assert request_information.http_method == Method.POST
+ request_information.path_parameters["baseurl"] =
"https://graph.microsoft.com/v1.0/";
+ assert (
+ request_information.url
+ ==
"https://graph.microsoft.com/v1.0/users/airflow%40example.com/sendMail";
+ )
+ assert json.loads(request_information.content) == {
+ "message": {
+ "subject": "Airflow alert",
+ "body": {"contentType": "HTML", "content": "Something
happened"},
+ "toRecipients": [{"emailAddress": {"address":
"[email protected]"}}],
+ },
+ "saveToSentItems": False,
+ }
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("from_email", (pytest.param(None, id="none"),
pytest.param("", id="empty")))
+ async def test_asend_email_without_a_sender_mailbox(self, from_email):
+ hook = MSGraphMailHook(conn_id="msgraph_api")
+
+ with pytest.raises(ValueError, match="mailbox to send from is
required"):
+ await hook.asend_email(
+ from_email=from_email,
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ )
+
+ @pytest.mark.asyncio
+ async def test_asend_email_when_dryrun(self):
+ with patch_hook_and_request_adapter(mock_json_response(202)) as mocks:
+ mock_get_http_response = mocks[-1]
+ hook = MSGraphMailHook(conn_id="msgraph_api")
+
+ await hook.asend_email(
+ from_email=self.FROM_EMAIL,
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ dryrun=True,
+ )
+
+ mock_get_http_response.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_asend_email_when_dryrun_still_builds_the_message(self):
+ hook = MSGraphMailHook(conn_id="msgraph_api")
+
+ with pytest.raises(ValueError, match="No recipients"):
+ await hook.asend_email(
+ from_email=self.FROM_EMAIL,
+ to=[],
+ subject="Airflow alert",
+ html_content="Something happened",
+ dryrun=True,
+ )
+
+ def test_send_email(self):
+ with patch_hook_and_request_adapter(mock_json_response(202)) as mocks:
+ mock_get_http_response = mocks[-1]
+ hook = MSGraphMailHook(conn_id="msgraph_api")
+
+ hook.send_email(
+ from_email=self.FROM_EMAIL,
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ )
+
+ request_information =
self.get_request_information(mock_get_http_response)
+ assert json.loads(request_information.content)["saveToSentItems"]
is True
+ # The adapter is bound to the event loop asyncio.run just closed,
so it must not be reused.
+ assert "msgraph_api" not in MSGraphMailHook.cached_request_adapters
+
+ def test_send_email_without_a_sender_mailbox(self):
+ hook = MSGraphMailHook(conn_id="msgraph_api")
+
+ # The connection is never opened, so the cleanup in the sync wrapper
has nothing to close.
+ with pytest.raises(ValueError, match="mailbox to send from is
required"):
+ hook.send_email(
+ from_email=None,
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ )
+
+ @pytest.mark.asyncio
+ async def test_send_email_inside_a_running_event_loop(self):
+ hook = MSGraphMailHook(conn_id="msgraph_api")
+
+ with pytest.raises(RuntimeError, match="await asend_email instead"):
+ hook.send_email(
+ from_email=self.FROM_EMAIL,
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ )
+
+
+class TestSendEmail:
+ @patch("airflow.providers.microsoft.azure.hooks.msgraph.MSGraphMailHook",
autospec=True)
+ def test_send_email(self, mock_hook):
+ send_email(
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ conn_id="msgraph_api",
+ from_email="[email protected]",
+ )
+
+ mock_hook.assert_called_once_with(conn_id="msgraph_api")
+ mock_hook.return_value.send_email.assert_called_once_with(
+ from_email="[email protected]",
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ files=None,
+ cc=None,
+ bcc=None,
+ custom_headers=None,
+ dryrun=False,
+ )
+
+ @patch("airflow.providers.microsoft.azure.hooks.msgraph.MSGraphMailHook",
autospec=True)
+ def test_send_email_without_conn_id(self, mock_hook):
+ mock_hook.default_conn_name = MSGraphMailHook.default_conn_name
+
+ send_email(
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ conn_id=None,
+ from_email="[email protected]",
+ )
+
+ mock_hook.assert_called_once_with(conn_id="msgraph_default")
+
+ @patch("airflow.providers.microsoft.azure.hooks.msgraph.MSGraphMailHook",
autospec=True)
+ def test_send_email_without_from_email(self, mock_hook):
+ with pytest.raises(ValueError, match="`from_email` configuration has
to be set"):
+ send_email(to="[email protected]", subject="Airflow alert",
html_content="Something happened")
+
+ mock_hook.assert_not_called()
+
+ @patch("airflow.providers.microsoft.azure.hooks.msgraph.MSGraphMailHook",
autospec=True)
+ def test_send_email_when_dryrun(self, mock_hook):
+ send_email(
+ to="[email protected]",
+ subject="Airflow alert",
+ html_content="Something happened",
+ dryrun=True,
+ from_email="[email protected]",
+ )
+
+ assert mock_hook.return_value.send_email.call_args.kwargs["dryrun"] is
True
+
+ def test_send_email_when_dryrun_reports_an_invalid_message(self):
+ with pytest.raises(ValueError, match="No recipients"):
+ send_email(
+ to=[],
+ subject="Airflow alert",
+ html_content="Something happened",
+ dryrun=True,
+ from_email="[email protected]",
+ )
diff --git
a/providers/microsoft/azure/tests/unit/microsoft/azure/notifications/__init__.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/notifications/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++
b/providers/microsoft/azure/tests/unit/microsoft/azure/notifications/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git
a/providers/microsoft/azure/tests/unit/microsoft/azure/notifications/test_msgraph.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/notifications/test_msgraph.py
new file mode 100644
index 00000000000..801c252db8d
--- /dev/null
+++
b/providers/microsoft/azure/tests/unit/microsoft/azure/notifications/test_msgraph.py
@@ -0,0 +1,113 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+
+from airflow.providers.microsoft.azure.notifications.msgraph import (
+ MSGraphNotifier,
+ send_msgraph_notification,
+)
+
+from tests_common.test_utils.config import conf_vars
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_1_PLUS
+
+TEST_DAG_ID = "test_dag"
+FROM_EMAIL = "[email protected]"
+TO_EMAIL = "[email protected]"
+
+EXPECTED_EMAIL = {
+ "from_email": FROM_EMAIL,
+ "to": TO_EMAIL,
+ "subject": "Airflow alert",
+ "html_content": "Something happened",
+ "files": None,
+ "cc": None,
+ "bcc": None,
+ "custom_headers": None,
+ "save_to_sent_items": True,
+}
+
+
[email protected]("airflow.providers.microsoft.azure.notifications.msgraph.MSGraphMailHook",
autospec=True)
+class TestMSGraphNotifier:
+ @pytest.mark.parametrize(
+ ("given", "expected_conn_id"),
+ (
+ pytest.param({}, "msgraph_default", id="default-connection"),
+ pytest.param({"conn_id": "msgraph_api"}, "msgraph_api",
id="explicit-connection"),
+ ),
+ )
+ def test_notifier(self, mock_hook, create_dag_without_db, given,
expected_conn_id):
+ notifier = send_msgraph_notification(
+ from_email=FROM_EMAIL,
+ to=TO_EMAIL,
+ subject="Airflow alert",
+ html_content="Something happened",
+ **given,
+ )
+
+ notifier({"dag": create_dag_without_db(TEST_DAG_ID)})
+
+ mock_hook.assert_called_once_with(conn_id=expected_conn_id)
+
mock_hook.return_value.send_email.assert_called_once_with(**EXPECTED_EMAIL)
+
+ def test_notifier_templated(self, mock_hook, create_dag_without_db):
+ notifier = MSGraphNotifier(
+ from_email=FROM_EMAIL,
+ to=TO_EMAIL,
+ subject="Dag {{ dag.dag_id }} failed",
+ html_content="Dag {{ dag.dag_id }} needs attention",
+ )
+
+ notifier({"dag": create_dag_without_db(TEST_DAG_ID)})
+
+ mock_hook.return_value.send_email.assert_called_once_with(
+ **{
+ **EXPECTED_EMAIL,
+ "subject": f"Dag {TEST_DAG_ID} failed",
+ "html_content": f"Dag {TEST_DAG_ID} needs attention",
+ }
+ )
+
+ def test_notifier_falls_back_to_the_configured_from_email(self, mock_hook,
create_dag_without_db):
+ with conf_vars({("email", "from_email"): FROM_EMAIL}):
+ notifier = MSGraphNotifier(
+ to=TO_EMAIL, subject="Airflow alert", html_content="Something
happened"
+ )
+
+ notifier({"dag": create_dag_without_db(TEST_DAG_ID)})
+
+
mock_hook.return_value.send_email.assert_called_once_with(**EXPECTED_EMAIL)
+
+ @pytest.mark.skipif(not AIRFLOW_V_3_1_PLUS, reason="Async support was
added to BaseNotifier in 3.1.0")
+ @pytest.mark.asyncio
+ async def test_async_notifier(self, mock_hook, create_dag_without_db):
+ notifier = MSGraphNotifier(
+ from_email=FROM_EMAIL,
+ to=TO_EMAIL,
+ subject="Airflow alert",
+ html_content="Something happened",
+ context={"dag": create_dag_without_db(TEST_DAG_ID)},
+ )
+
+ await notifier
+
+
mock_hook.return_value.asend_email.assert_awaited_once_with(**EXPECTED_EMAIL)
