graph_email_utils

Utils to read Outlook emails via Microsoft Graph API.

This module provides functions to connect to a Microsoft 365 mailbox via the Graph API (client-credentials flow) and search emails by subject.

Installation

from physical_operations_utils.graph_email_utils import (
    connect,
    get_emails_by_subject,
)

Basic usage

from physical_operations_utils.graph_email_utils import connect, get_emails_by_subject

# Initialise the client (secrets are fetched from Azure Key Vault)
connect(
    tenant_id_secret="AxpoGlobal-TenantId",
    client_id_secret="EmailService-ClientId",
    client_secret_secret="EmailService-ClientSecret",
    user_id="nomination.no@axpo.com",
)

# Search emails by subject
emails = get_emails_by_subject("Daily Report", max_results=10)
for email in emails:
    print(email["subject"], email["from_address"])

Email dict structure

All read functions return dicts with the following keys:

id (str): Graph message ID.
subject (str): Email subject line.
body (str): Email body content (HTML or plain text).
is_html (bool): Whether the body is HTML.
from_address (str): Sender email address.
received_at (datetime): UTC datetime when the email was received.
to_recipients (list[str]): List of recipient email addresses.

Prerequisites

The Entra (Azure AD) app registration must have the Mail.ReadWrite application permission granted with admin consent.

`connect(tenant_id_secret, client_id_secret, client_secret_secret, user_id)`

Initialise the Microsoft Graph client using client-credentials flow.

Fetches the client ID and client secret from Azure Key Vault using the provided secret names, then builds a GraphServiceClient ready for use by the other functions in this module.

Parameters:

Name	Type	Description	Default
`tenant_id_secret`	`str`	Key Vault secret name that stores the Azure AD / Entra tenant ID.	required
`client_id_secret`	`str`	Key Vault secret name that stores the app registration client ID.	required
`client_secret_secret`	`str`	Key Vault secret name that stores the app registration client secret.	required
`user_id`	`str`	Mailbox user principal name or object ID (e.g. `"shared-mailbox@yourtenant.onmicrosoft.com"`).	required

Raises:

Type	Description
`ValueError`	If any argument is empty.

Example

from physical_operations_utils.graph_email_utils import connect

connect(
    tenant_id_secret="AxpoGlobal-TenantId",
    client_id_secret="EmailService-ClientId",
    client_secret_secret="EmailService-ClientSecret",
    user_id="nomination.no@axpo.com",
)

Source code in physical_operations_utils/graph_email_utils.py

def connect(
    tenant_id_secret: str,
    client_id_secret: str,
    client_secret_secret: str,
    user_id: str,
) -> None:
    """Initialise the Microsoft Graph client using client-credentials flow.

    Fetches the client ID and client secret from Azure Key Vault using the
    provided secret names, then builds a ``GraphServiceClient`` ready for
    use by the other functions in this module.

    Args:
        tenant_id_secret: Key Vault secret name that stores the Azure AD / Entra tenant ID.
        client_id_secret: Key Vault secret name that stores the app registration client ID.
        client_secret_secret: Key Vault secret name that stores the app registration client secret.
        user_id: Mailbox user principal name or object ID
            (e.g. ``"shared-mailbox@yourtenant.onmicrosoft.com"``).

    Raises:
        ValueError: If any argument is empty.

    Example:
        ```python
        from physical_operations_utils.graph_email_utils import connect

        connect(
            tenant_id_secret="AxpoGlobal-TenantId",
            client_id_secret="EmailService-ClientId",
            client_secret_secret="EmailService-ClientSecret",
            user_id="nomination.no@axpo.com",
        )
        ```
    """
    if (
        not tenant_id_secret
        or not client_id_secret
        or not client_secret_secret
        or not user_id
    ):
        raise ValueError(
            "All arguments (tenant_id_secret, client_id_secret, client_secret_secret, user_id) are required."
        )

    global _graph_client, _user_id  # noqa: PLW0603

    tenant_id = get_secret(tenant_id_secret)
    client_id = get_secret(client_id_secret)
    client_secret = get_secret(client_secret_secret)

    credential = ClientSecretCredential(
        tenant_id=tenant_id,
        client_id=client_id,
        client_secret=client_secret,
    )
    _graph_client = GraphServiceClient(
        credentials=credential,
        scopes=["https://graph.microsoft.com/.default"],
    )
    _user_id = user_id
    logger.info("Graph email client connected for user %s", user_id)

`get_emails_by_subject(subject, max_results=250)`

Search inbox for emails whose subject contains the given string.

Uses the Graph API $search operator for server-side subject search.

Parameters:

Name	Type	Description	Default
`subject`	`str`	Substring to search for in the email subject (case-insensitive).	required
`max_results`	`int`	Maximum number of emails to return. Defaults to 250.	`250`

Returns:

Type	Description
`list[dict[str, Any]]`	A list of dicts with keys: `id`, `subject`, `body`, `is_html`, `from_address`, `received_at`, `to_recipients`.

Raises:

Type	Description
`RuntimeError`	If `connect()` has not been called.
`ValueError`	If `subject` is empty.

Example

from physical_operations_utils.graph_email_utils import connect, get_emails_by_subject

connect(
    tenant_id_secret="AxpoGlobal-TenantId",
    client_id_secret="EmailService-ClientId",
    client_secret_secret="EmailService-ClientSecret",
    user_id="nomination.no@axpo.com",
)

emails = get_emails_by_subject("Daily Report")
for email in emails:
    print(email["subject"], email["received_at"])

Source code in physical_operations_utils/graph_email_utils.py

@retry(
    wait=wait_fixed(2),
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type((ConnectionError, TimeoutError)),
)
def get_emails_by_subject(
    subject: str,
    max_results: int = 250,
) -> list[dict[str, Any]]:
    """Search inbox for emails whose subject contains the given string.

    Uses the Graph API ``$search`` operator for server-side subject search.

    Args:
        subject: Substring to search for in the email subject
            (case-insensitive).
        max_results: Maximum number of emails to return. Defaults to 250.

    Returns:
        A list of dicts with keys:
            ``id``, ``subject``, ``body``, ``is_html``, ``from_address``,
            ``received_at``, ``to_recipients``.

    Raises:
        RuntimeError: If ``connect()`` has not been called.
        ValueError: If ``subject`` is empty.

    Example:
        ```python
        from physical_operations_utils.graph_email_utils import connect, get_emails_by_subject

        connect(
            tenant_id_secret="AxpoGlobal-TenantId",
            client_id_secret="EmailService-ClientId",
            client_secret_secret="EmailService-ClientSecret",
            user_id="nomination.no@axpo.com",
        )

        emails = get_emails_by_subject("Daily Report")
        for email in emails:
            print(email["subject"], email["received_at"])
        ```
    """
    _ensure_connected()
    if not subject:
        raise ValueError("subject must not be empty.")

    return _run_async(_get_emails_by_subject_async(subject, max_results))