#
# 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.
"""
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
# Stays on httpx (not httpx2): the client below is handed to kiota_http, which builds and
# consumes httpx objects, and the two packages' classes are distinct.
# Migrate once msgraph-core/kiota_http move to httpx2; tracked at https://github.com/apache/airflow/issues/70522
import httpx
from azure.core.credentials_async import AsyncTokenCredential
from azure.identity.aio import CertificateCredential, ClientSecretCredential
from httpx import AsyncHTTPTransport, Response, Timeout
from kiota_abstractions.api_error import APIError
from kiota_abstractions.method import Method
from kiota_abstractions.request_information import RequestInformation
from kiota_abstractions.response_handler import ResponseHandler
from kiota_abstractions.serialization import ParseNodeFactoryRegistry
from kiota_authentication_azure.azure_identity_authentication_provider import (
AzureIdentityAuthenticationProvider,
)
from kiota_http.httpx_request_adapter import HttpxRequestAdapter
from kiota_http.middleware.options import ResponseHandlerOption
from kiota_serialization_json.json_parse_node_factory import JsonParseNodeFactory
from kiota_serialization_text.text_parse_node_factory import TextParseNodeFactory
from msgraph_core import APIVersion, GraphClientFactory
from msgraph_core._enums import NationalClouds
from airflow.exceptions import AirflowBadRequest, AirflowConfigException, AirflowProviderDeprecationWarning
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
from kiota_abstractions.response_handler import NativeResponseType
from kiota_abstractions.serialization import ParsableFactory
from kiota_authentication_azure.azure_identity_access_token_provider import (
AzureIdentityAccessTokenProvider,
)
from airflow.providers.common.compat.sdk import Connection
[docs]
def execute_callable(func: Callable, *args: Any, **kwargs: Any) -> Any:
"""Dynamically call a function by matching its signature to provided args/kwargs."""
sig = inspect.signature(func)
accepts_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values())
if not accepts_kwargs:
# Only pass arguments the function explicitly declares
filtered_kwargs = {k: v for k, v in kwargs.items() if k in sig.parameters}
else:
filtered_kwargs = kwargs
try:
sig.bind(*args, **filtered_kwargs)
except TypeError as err:
raise TypeError(
f"Failed to bind arguments to function {func.__name__}: {err}\n"
f"Expected parameters: {list(sig.parameters.keys())}\n"
f"Provided kwargs: {list(kwargs.keys())}"
) from err
return func(*args, **filtered_kwargs)
[docs]
class CachedAsyncTokenCredential(AsyncTokenCredential): # type: ignore[misc]
"""
Wraps an async Azure credential to prevent ``kiota`` from closing it after each token request.
``kiota_authentication_azure`` calls ``await credential.close()`` after every successful
``get_token`` call (see ``AzureIdentityAccessTokenProvider.get_authorization_token``). That
tears down the underlying ``AioHttpTransport`` session so the next request fails with
"HTTP transport has already been closed". Suppressing ``close()`` keeps the session alive
for the lifetime of the cached ``RequestAdapter``.
"""
def __init__(self, credential: ClientSecretCredential | CertificateCredential):
self._credential = credential
@property
def _transport(self) -> RequestsTransport:
return self._credential._client._pipeline._transport
@property
[docs]
def closed(self) -> bool:
# _closed is set to True by AioHttpTransport.close(); check it first as it
# is authoritative even when _has_been_opened is still False.
if getattr(self._transport, "_closed", False):
return True
if not self._transport._has_been_opened and self._transport.session is None:
return False
if self._transport.session is not None:
return self._transport.session.closed
return True
[docs]
async def __aenter__(self) -> AsyncTokenCredential:
return self
[docs]
async def __aexit__(
self,
exc_type: type[BaseException] | None = None,
exc_value: BaseException | None = None,
traceback: TracebackType | None = None,
) -> None:
await self.close()
[docs]
async def get_token(self, *args: Any, **kwargs: Any) -> Any:
return await self._credential.get_token(*args, **kwargs)
[docs]
async def get_token_info(self, *args: Any, **kwargs: Any) -> Any:
return await self._credential.get_token_info(*args, **kwargs) # type: ignore[union-attr]
[docs]
async def close(self) -> None:
"""Intentionally a no-op — the credential session is closed when the adapter is evicted."""
[docs]
class DefaultResponseHandler(ResponseHandler):
"""DefaultResponseHandler returns JSON payload or content in bytes or response headers."""
@staticmethod
[docs]
def get_value(response: Response) -> Any:
with suppress(JSONDecodeError, UnicodeDecodeError):
return response.json()
content = response.content
if not content:
return {key: value for key, value in response.headers.items()}
return content
[docs]
async def handle_response_async(
self, response: NativeResponseType, error_map: dict[str, ParsableFactory] | None
) -> Any:
"""
Invoke this callback method when a response is received.
param response: The type of the native response object.
param error_map: The error dict to use in case of a failed request.
"""
resp: Response = cast("Response", response)
value = self.get_value(resp)
if resp.status_code not in {200, 201, 202, 204, 302}:
message = value or resp.reason_phrase
status_code = HTTPStatus(resp.status_code)
if status_code == HTTPStatus.BAD_REQUEST:
raise AirflowBadRequest(message)
if status_code == HTTPStatus.UNAUTHORIZED:
raise PermissionError(message)
if status_code == HTTPStatus.NOT_FOUND:
raise AirflowNotFoundException(message)
raise AirflowException(message)
return value
[docs]
class KiotaRequestAdapterHook(BaseHook):
"""
A Microsoft Graph API interaction hook, a Wrapper around KiotaRequestAdapter.
https://github.com/microsoftgraph/msgraph-sdk-python-core
:param conn_id: The HTTP Connection ID to run the trigger against.
: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).
You can pass an enum named APIVersion which has 2 possible members v1 and beta,
or you can pass a string as "v1.0" or "beta".
"""
[docs]
DEFAULT_SCOPE = "https://graph.microsoft.com/.default"
[docs]
cached_request_adapters: dict[str, tuple[str, RequestAdapter]] = {}
[docs]
conn_type: str = "msgraph"
[docs]
conn_name_attr: str = "conn_id"
[docs]
default_conn_name: str = "msgraph_default"
[docs]
hook_name: str = "Microsoft Graph API"
def __init__(
self,
conn_id: str = default_conn_name,
timeout: float | None = None,
proxies: dict | None = None,
host: str | None = None,
scopes: str | list[str] | None = None,
api_version: APIVersion | str | None = None,
):
super().__init__()
[docs]
self.host = self._ensure_protocol(host)
if isinstance(scopes, str):
self.scopes = [scopes]
else:
self.scopes = scopes or [self.DEFAULT_SCOPE]
[docs]
self.api_version = self.resolve_api_version_from_value(api_version)
[docs]
self.allowed_netloc: str | None = None
def _ensure_protocol(self, host: str | None, schema: str = "https") -> str | None:
"""Ensure URL has http:// or https:// protocol prefix."""
if not host:
return None
if host.startswith(("http://", "https://")):
return host
self.log.warning(
"URL '%s' is missing protocol prefix. Automatically adding '%s://'. "
"Please update your connection configuration to include the full URL with protocol.",
host,
schema,
)
return f"{schema}://{host}"
@classmethod
@classmethod
[docs]
def get_ui_field_behaviour(cls) -> dict[str, Any]:
"""Return custom field behaviour."""
return {
"hidden_fields": ["extra"],
"relabeling": {
"login": "Client ID",
"password": "Client Secret",
},
"default_values": {
"schema": "https",
"host": NationalClouds.Global.value,
"port": 443,
},
}
@staticmethod
[docs]
def resolve_api_version_from_value(
api_version: APIVersion | str, default: str | None = None
) -> str | None:
if isinstance(api_version, APIVersion):
return api_version.value
return api_version or default
[docs]
def get_api_version(self, config: dict) -> str:
return self.api_version or self.resolve_api_version_from_value(
config.get("api_version"), APIVersion.v1.value
) # type: ignore
[docs]
def get_host(self, connection: Connection) -> str:
if not self.host:
if connection.schema and connection.host:
return f"{connection.schema}://{connection.host}"
return NationalClouds.Global.value
schema = connection.schema or "https"
return cast("str", self._ensure_protocol(self.host, schema))
[docs]
def get_base_url(self, host: str, api_version: str, config: dict) -> str:
base_url = config.get("base_url", urljoin(host, api_version)).strip()
if not base_url.endswith("/"):
return f"{base_url}/"
return base_url
@staticmethod
@classmethod
[docs]
def to_httpx_proxies(cls, proxies: dict | None) -> dict | None:
if proxies:
proxies = proxies.copy()
if proxies.get("http"):
proxies["http://"] = AsyncHTTPTransport(proxy=proxies.pop("http"))
if proxies.get("https"):
proxies["https://"] = AsyncHTTPTransport(proxy=proxies.pop("https"))
if proxies.get("no"):
for url in proxies.pop("no", "").split(","):
proxies[cls.format_no_proxy_url(url.strip())] = None
return proxies
return None
[docs]
def to_msal_proxies(self, authority: str | None, proxies: dict | None) -> dict | None:
self.log.debug("authority: %s", authority)
if authority and proxies:
no_proxies = proxies.get("no")
self.log.debug("no_proxies: %s", no_proxies)
if no_proxies:
for url in no_proxies.split(","):
self.log.info("url: %s", url)
domain_name = urlparse(url).path.replace("*", "")
self.log.debug("domain_name: %s", domain_name)
if authority.endswith(domain_name):
return None
return proxies
if proxies:
return proxies
return None
@staticmethod
[docs]
def get_allowed_hosts(authority: str | None, config: dict) -> list[str]:
allowed_hosts = config.get("allowed_hosts", authority)
if not allowed_hosts:
return []
return [host for host in allowed_hosts.split(",") if host]
def _build_request_adapter(self, connection) -> tuple[str, RequestAdapter]:
client_id = connection.login
client_secret = connection.password
# TODO (#54350): do not use connection.extra_dejson until it's fixed in Airflow otherwise expect:
# RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop.
config = json.loads(connection.extra) if connection.extra else {}
api_version = self.get_api_version(config)
host = self.get_host(connection) # type: ignore[arg-type]
base_url = self.get_base_url(host, api_version, config)
authority = config.get("authority")
proxies = self.get_proxies(config)
httpx_proxies = self.to_httpx_proxies(proxies=proxies)
scopes = config.get("scopes", self.scopes)
if isinstance(scopes, str):
scopes = scopes.split(",")
verify = config.get("verify", True)
trust_env = config.get("trust_env", False)
allowed_hosts = self.get_allowed_hosts(authority, config)
self.log.info(
"Creating Microsoft Graph SDK client %s for conn_id: %s",
api_version,
self.conn_id,
)
self.log.info("Host: %s", host)
self.log.info("Base URL: %s", base_url)
self.log.info("Client id: %s", client_id)
self.log.info("Client secret: %s", redact(client_secret, name="client_secret"))
self.log.info("API version: %s", api_version)
self.log.info("Scope: %s", scopes)
self.log.info("Verify: %s", verify)
self.log.info("Timeout: %s", self.timeout)
self.log.info("Trust env: %s", trust_env)
self.log.info("Authority: %s", authority)
self.log.info("Allowed hosts: %s", allowed_hosts)
self.log.info("Proxies: %s", redact(proxies, name="proxies"))
self.log.info("HTTPX Proxies: %s", redact(httpx_proxies, name="proxies"))
credentials = self.get_credentials(
login=connection.login,
password=connection.password,
config=config,
authority=authority,
verify=verify,
proxies=proxies,
)
http_client = GraphClientFactory.create_with_default_middleware(
api_version=api_version,
client=httpx.AsyncClient(
mounts=httpx_proxies,
timeout=Timeout(timeout=self.timeout),
verify=verify,
trust_env=trust_env,
base_url=base_url,
),
host=host,
)
auth_provider = AzureIdentityAuthenticationProvider(
credentials=credentials,
scopes=scopes,
allowed_hosts=allowed_hosts,
)
parse_node_factory = ParseNodeFactoryRegistry()
parse_node_factory.CONTENT_TYPE_ASSOCIATED_FACTORIES["text/plain"] = TextParseNodeFactory()
parse_node_factory.CONTENT_TYPE_ASSOCIATED_FACTORIES["application/json"] = JsonParseNodeFactory()
request_adapter = HttpxRequestAdapter(
authentication_provider=auth_provider,
parse_node_factory=parse_node_factory,
http_client=http_client,
base_url=base_url,
)
return api_version, request_adapter
[docs]
def get_conn(self) -> RequestAdapter:
"""
Initiate a new RequestAdapter connection.
.. warning::
This method is deprecated. Use :meth:`get_async_conn` instead.
"""
if not self.conn_id:
raise AirflowException("Failed to create the KiotaRequestAdapterHook. No conn_id provided!")
warnings.warn(
"get_conn is deprecated, please use the async get_async_conn method!",
category=AirflowProviderDeprecationWarning,
stacklevel=2,
)
api_version, request_adapter = self.cached_request_adapters.get(self.conn_id, (None, None))
if not request_adapter:
connection = self.get_connection(conn_id=self.conn_id)
api_version, request_adapter = self._build_request_adapter(connection)
self.cached_request_adapters[self.conn_id] = (api_version, request_adapter)
self.api_version = api_version
return request_adapter
@staticmethod
def _is_http_client_closed(request_adapter: RequestAdapter) -> bool:
"""Return True when the underlying httpx AsyncClient has been closed."""
adapter = cast("HttpxRequestAdapter", request_adapter)
if adapter._http_client.is_closed:
return True
provider = cast("BaseBearerTokenAuthenticationProvider", adapter._authentication_provider)
access_token_provider = cast("AzureIdentityAccessTokenProvider", provider.access_token_provider)
credential = cast("CachedAsyncTokenCredential", access_token_provider._credentials)
return credential.closed
[docs]
async def get_async_conn(self) -> RequestAdapter:
"""Initiate a new RequestAdapter connection asynchronously."""
if not self.conn_id:
raise AirflowException("Failed to create the KiotaRequestAdapterHook. No conn_id provided!")
api_version, request_adapter = self.cached_request_adapters.get(self.conn_id, (None, None))
if request_adapter and self._is_http_client_closed(request_adapter):
self.log.warning(
"Cached request adapter for conn_id '%s' has a closed HTTP client. Rebuilding.",
self.conn_id,
)
self.cached_request_adapters.pop(self.conn_id, None)
request_adapter = None
if not request_adapter:
connection = await get_async_connection(conn_id=self.conn_id)
api_version, request_adapter = self._build_request_adapter(connection)
self.cached_request_adapters[self.conn_id] = (api_version, request_adapter)
self.api_version = api_version
self.allowed_netloc = urlparse(request_adapter.base_url).netloc.lower()
return request_adapter
[docs]
def get_proxies(self, config: dict) -> dict | None:
proxies = self.proxies if self.proxies is not None else config.get("proxies", {})
if proxies:
if isinstance(proxies, str):
# TODO: Once provider depends on Airflow 2.10 or higher code below won't be needed anymore as
# we could then use the get_extra_dejson method on the connection which deserializes
# nested json. Make sure to use connection.get_extra_dejson(nested=True) instead of
# connection.extra_dejson.
with suppress(JSONDecodeError):
proxies = json.loads(proxies)
with suppress(Exception):
proxies = literal_eval(proxies)
if not isinstance(proxies, dict):
raise AirflowConfigException(
f"Proxies must be of type dict, got {type(proxies).__name__} instead!"
)
return proxies
return None
[docs]
def get_credentials(
self,
login: str | None,
password: str | None,
config,
authority: str | None,
verify: bool,
proxies: dict | None,
) -> AsyncTokenCredential:
tenant_id = config.get("tenant_id") or config.get("tenantId")
certificate_path = config.get("certificate_path")
certificate_data = config.get("certificate_data")
disable_instance_discovery = config.get("disable_instance_discovery", False)
msal_proxies = self.to_msal_proxies(authority=authority, proxies=proxies)
self.log.info("Tenant id: %s", tenant_id)
self.log.info("Certificate path: %s", certificate_path)
self.log.info("Certificate data: %s", certificate_data is not None)
self.log.info("Authority: %s", authority)
self.log.info("Disable instance discovery: %s", disable_instance_discovery)
self.log.info("MSAL Proxies: %s", redact(msal_proxies, name="proxies"))
if certificate_path or certificate_data:
return CachedAsyncTokenCredential(
CertificateCredential(
tenant_id=tenant_id,
client_id=login, # type: ignore
password=password,
certificate_path=certificate_path,
certificate_data=certificate_data.encode() if certificate_data else None,
authority=authority,
proxies=msal_proxies,
disable_instance_discovery=disable_instance_discovery,
connection_verify=verify,
)
)
return CachedAsyncTokenCredential(
ClientSecretCredential(
tenant_id=tenant_id,
client_id=login, # type: ignore
client_secret=password, # type: ignore
authority=authority,
proxies=msal_proxies,
disable_instance_discovery=disable_instance_discovery,
connection_verify=verify,
)
)
[docs]
def test_connection(self):
"""Test HTTP Connection."""
try:
asyncio.run(self.run())
return True, "Connection successfully tested"
except Exception as e:
return False, str(e)
@staticmethod
[docs]
async def run(
self,
url: str = "",
response_type: str | None = None,
path_parameters: dict[str, Any] | None = None,
method: str = "GET",
query_parameters: dict[str, Any] | None = None,
headers: dict[str, str] | None = None,
data: dict[str, Any] | str | BytesIO | None = None,
):
response = await self.send_request(
request_info=self.request_information(
url=url,
response_type=response_type,
path_parameters=path_parameters,
method=method,
query_parameters=query_parameters,
headers=headers,
data=data,
),
response_type=response_type,
)
self.log.debug("response: %s", response)
return response
[docs]
async def get_allowed_netlocs(self) -> set[str]:
"""Return the endpoint's host and the connection's allowed hosts, for checking pagination links."""
request_adapter = await self.get_async_conn()
adapter = cast("HttpxRequestAdapter", request_adapter)
provider = cast("BaseBearerTokenAuthenticationProvider", adapter._authentication_provider)
access_token_provider = cast("AzureIdentityAccessTokenProvider", provider.access_token_provider)
allowed_hosts = access_token_provider.get_allowed_hosts_validator().get_allowed_hosts()
netlocs = {host.lower() for host in allowed_hosts}
if self.allowed_netloc:
netlocs.add(self.allowed_netloc)
return netlocs
[docs]
async def assert_allowed_host(self, url: str | None) -> None:
"""
Refuse an absolute ``url`` whose host the connection does not allow.
A pagination link (e.g. ``@odata.nextLink``) is echoed from the API response and is re-fetched
with the connection's bearer token attached. That token is withheld only from hosts outside
``allowed_hosts``, which defaults to empty (any host) unless configured, so a tampered response
could send it to an arbitrary host (CWE-918).
"""
if not url or not url.startswith("http"):
return
allowed_netlocs = await self.get_allowed_netlocs()
if urlparse(url).netloc.lower() not in allowed_netlocs:
raise ValueError(
f"Refusing to follow pagination link {url!r}: its host is not among the allowed "
f"Microsoft Graph endpoints {sorted(allowed_netlocs)}."
)
[docs]
async def paginated_run(
self,
url: str = "",
response_type: str | None = None,
path_parameters: dict[str, Any] | None = None,
method: str = "GET",
query_parameters: dict[str, Any] | None = None,
headers: dict[str, str] | None = None,
data: dict[str, Any] | str | BytesIO | None = None,
pagination_function: PaginationCallable | None = None,
):
if pagination_function is None:
pagination_function = self.default_pagination
responses: list[dict] = []
async def run(
url: str = "",
query_parameters: dict[str, Any] | None = None,
):
while url:
response = await self.run(
url=url,
response_type=response_type,
path_parameters=path_parameters,
method=method,
query_parameters=query_parameters,
headers=headers,
data=data,
)
if response:
responses.append(response)
if pagination_function:
next_url, query_parameters = execute_callable(
pagination_function,
response=response,
url=url,
response_type=response_type,
path_parameters=path_parameters,
method=method,
query_parameters=query_parameters,
headers=headers,
data=data,
responses=lambda: responses,
)
await self.assert_allowed_host(next_url)
url = next_url
else:
break
await run(url=url, query_parameters=query_parameters)
return responses
[docs]
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
[docs]
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("CachedAsyncTokenCredential", access_token_provider._credentials)
await credential._credential.close()
@staticmethod
[docs]
def normalize_url(url: str) -> str | None:
if url.startswith("/"):
return url.replace("/", "", 1)
return url
@staticmethod
[docs]
def encoded_query_parameters(query_parameters) -> dict:
if query_parameters:
return {quote(key): value for key, value in query_parameters.items()}
return {}
@staticmethod
[docs]
def error_mapping() -> dict[str, type[ParsableFactory]]:
return {
"4XX": APIError, # type: ignore
"5XX": APIError, # type: ignore
}
[docs]
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.
[docs]
MAX_ATTACHMENTS_SIZE = 3 * 1024 * 1024
@staticmethod
@staticmethod
@classmethod
[docs]
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
[docs]
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
[docs]
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
[docs]
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},
)
[docs]
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())
[docs]
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,
)