dabla commented on code in PR #71565:
URL: https://github.com/apache/airflow/pull/71565#discussion_r3948526360
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py:
##########
@@ -789,3 +807,264 @@ def error_mapping() -> dict[str, type[ParsableFactory]]:
"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]
+
+ @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,
+ ) -> 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.
+ """
+ # 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}."
+ )
+
+ message = self.build_message(
+ to=to,
+ subject=subject,
+ html_content=html_content,
+ files=files,
+ cc=cc,
+ bcc=bcc,
+ custom_headers=custom_headers,
+ )
+
+ await self.run(
+ url="users/{user_id}/sendMail",
+ path_parameters={"user_id": sender},
+ method="POST",
+ data={"message": message, "saveToSentItems": save_to_sent_items},
+ )
+
+ async def close_async_conn(self) -> None:
+ """Close the request adapter cached for this connection and evict it
from the cache."""
+ _, request_adapter = self.cached_request_adapters.pop(self.conn_id,
(None, None))
+
+ if not request_adapter:
+ return
+
+ adapter = cast("HttpxRequestAdapter", request_adapter)
+ await adapter._http_client.aclose()
+ provider = cast("BaseBearerTokenAuthenticationProvider",
adapter._authentication_provider)
+ access_token_provider = cast("AzureIdentityAccessTokenProvider",
provider.access_token_provider)
+ credential = cast("CachedAsyncTokenCredential",
access_token_provider._credentials)
+ await credential._credential.close()
+
+ 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,
+ ) -> 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.
+ """
+ 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,
+ )
+ 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_async_conn()
+
+ 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."
+ )
+
+ if dryrun:
+ log.info("Dry run, not sending email with subject %r to %s", subject,
to)
Review Comment:
Maybe the dry run could call the `build_message` method, so at least it does
something except sending the actual mail?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
