dabla commented on code in PR #71565:
URL: https://github.com/apache/airflow/pull/71565#discussion_r3956861072
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py:
##########
@@ -789,3 +804,276 @@ 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]
+
+ @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},
+ )
+
+ 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()
Review Comment:
This method should be renamed as close and should move to the parent hook
`KiotaRequestAdapterHook`. Also we know the `KiotaRequestAdapterHook` uses the
`CachedAsyncTokenCredential` which is a decorator holding a reference to an
`AsyncTokenCredential` instance (e.g. `CertificateCredential` or
`ClientSecretCredential`) and call the close there like you already did.
So we could write the async `close` method like this in
`KiotaRequestAdapterHook`:
```
async def close(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
try:
adapter = cast("HttpxRequestAdapter", request_adapter)
await adapter._http_client.aclose()
finally:
provider = cast("BaseBearerTokenAuthenticationProvider",
adapter._authentication_provider)
access_token_provider = cast("AzureIdentityAccessTokenProvider",
provider.access_token_provider)
credential = cast("AsyncTokenCredential",
access_token_provider._credentials)
await credential._credential.close()
```
Then in the `send_request` method of `KiotaRequestAdapterHook`, we could
write it like this:
```
async def send_request(self, request_info: RequestInformation,
response_type: str | None = None):
conn = await self.get_async_conn()
try:
self.log.info("Executing url '%s' as '%s'", request_info.url,
request_info.http_method)
if response_type:
return await conn.send_primitive_async(
request_info=request_info,
response_type=response_type,
error_map=self.error_mapping(),
)
return await conn.send_no_response_content_async(
request_info=request_info,
error_map=self.error_mapping(),
)
except (PermissionError, RuntimeError, ValueError) as e:
self.log.warning(
"Request failed for conn_id '%s': %s. Invalidating cached
request adapter.",
self.conn_id,
e,
)
await self.close()
raise
```
--
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]
