email utils

Utils to send emails with attachments.

send_email_from_axponordic_domain(sender, subject, body, body_contains_html, recipients, exception_on_invalid_addresses, cc=None, attachment_file_path=None, api_key=None)

Sends an email via Mailgun using the axponordic.com domain.

This function handles both HTML and plain-text bodies (with automatic newline-to-
conversion), supports optional CC and file attachments, and validates recipients and sender addresses based on the active environment. In non-production environments, only @axpo.com recipients and specific whitelisted SMS addresses are allowed.

If no API key is provided, it is retrieved from the Azure Key Vault via get_secret("Mailgun-APIKey").

Parameters:

Name	Type	Description	Default
sender	str	The email address sending the message. Must end with @axponordic.com.	required
subject	str	The subject line of the email.	required
body	str	The email body, either plain text or HTML.	required
body_contains_html	bool	Whether the body already contains HTML. If False, newlines are converted to <br> and wrapped in <p>.	required
recipients	List[str]	A list of recipient email addresses.	required
cc	List[str] | None	Optional list of CC addresses.	None
exception_on_invalid_addresses	bool	If True, raises a ValueError on invalid addresses.	required
attachment_file_path	str | None	Optional path to a file to attach. The file must be	None
api_key	str | None	Optional Mailgun API key. If not provided, fetched from Azure Key Vault.	None

Returns:

Type	Description
Response	requests.Response: The response returned by the Mailgun requests.post call.

Raises:

Type	Description
ValueError	If the sender address does not end with @axponordic.com, or if a non-production recipient is not from an approved domain.

Example

from physical_operations_utils.email_utils import send_email_from_axponordic_domain

send_email_from_axponordic_domain(
    sender="alerts@axponordic.com",
    subject="Forecast Update",
    body="New forecast available.",
    body_contains_html=False,
    recipients=["team@axpo.com"],
    cc=None,
)

Source code in physical_operations_utils/email_utils.py

def send_email_from_axponordic_domain(
    sender: str,
    subject: str,
    body: str,
    body_contains_html: bool,
    recipients: List[str],
    exception_on_invalid_addresses: bool,
    cc: List[str] | None = None,
    attachment_file_path: str | None = None,
    api_key: str | None = None,
) -> requests.Response:
    """
    Sends an email via Mailgun using the axponordic.com domain.

    This function handles both HTML and plain-text bodies (with automatic newline-to-<br> conversion),
    supports optional CC and file attachments, and validates recipients and sender addresses based
    on the active environment. In non-production environments, only `@axpo.com` recipients and specific
    whitelisted SMS addresses are allowed.

    If no API key is provided, it is retrieved from the Azure Key Vault via `get_secret("Mailgun-APIKey")`.

    Args:
        sender (str): The email address sending the message. Must end with `@axponordic.com`.
        subject (str): The subject line of the email.
        body (str): The email body, either plain text or HTML.
        body_contains_html (bool): Whether the body already contains HTML. If False, newlines are converted to `<br>` and wrapped in `<p>`.
        recipients (List[str]): A list of recipient email addresses.
        cc ((List[str]) | None): Optional list of CC addresses.
        exception_on_invalid_addresses (bool): If True, raises a ValueError on invalid addresses.
        attachment_file_path (str | None): Optional path to a file to attach. The file must be
        api_key (str | None): Optional Mailgun API key. If not provided, fetched from Azure Key Vault.

    Returns:
        requests.Response: The response returned by the Mailgun `requests.post` call.

    Raises:
        ValueError: If the sender address does not end with `@axponordic.com`, or if a non-production
                    recipient is not from an approved domain.

    Example:
        ```python
        from physical_operations_utils.email_utils import send_email_from_axponordic_domain

        send_email_from_axponordic_domain(
            sender="alerts@axponordic.com",
            subject="Forecast Update",
            body="New forecast available.",
            body_contains_html=False,
            recipients=["team@axpo.com"],
            cc=None,
        )
        ```
    """
    if api_key is None:
        api_key = get_secret("Mailgun-APIKey")
    if not sender.endswith("@axponordic.com"):
        raise ValueError(
            f"Sender email address must end with '@axponordic.com'. Got: '{sender}'"
        )
    if os.getenv("ENVIRONMENT") != "prod":
        subject = f"PHYSICAL_DESK_INTEGRATION_TEST_{subject}"
        for recipient in recipients:
            if not (recipient.endswith("@axpo.com")) and (
                recipient not in NONPROD_WHITELISTED_SMS_ADDRESSES
            ):
                raise ValueError(
                    f"Environment is {os.getenv('ENVIRONMENT')}. Recipient email address must end with '@axpo.com' if environment is not prod. Got: '{recipient}'"
                )
        if cc is not None:
            for recipient in cc:
                if not (recipient.endswith("@axpo.com")) and (
                    recipient not in NONPROD_WHITELISTED_SMS_ADDRESSES
                ):
                    raise ValueError(
                        f"Environment is {os.getenv('ENVIRONMENT')}. CC email address must end with '@axpo.com' if environment is not prod. Got: '{recipient}'"
                    )

    valid_recipients, invalid_recipients = validate_email_addresses(
        addresses=recipients, raise_exception=exception_on_invalid_addresses
    )
    valid_cc, invalid_cc = (
        validate_email_addresses(
            addresses=cc, raise_exception=exception_on_invalid_addresses
        )
        if cc
        else ([], [])
    )
    if invalid_recipients:
        logging.warning(
            f"Invalid recipient email addresses: {', '.join(invalid_recipients)}"
        )
    if invalid_cc:
        logging.warning(f"Invalid CC email addresses: {', '.join(invalid_cc)}")

    if not valid_recipients:
        raise ValueError("No valid email addresses provided.")

    files = None
    if attachment_file_path:
        _validate_file_size(attachment_file_path)
        _validate_file_type(attachment_file_path)
        with open(attachment_file_path, "rb") as file:
            files = [
                ("attachment", (os.path.basename(attachment_file_path), file.read()))
            ]
    res = requests.post(
        "https://api.eu.mailgun.net/v3/axponordic.com/messages",
        auth=("api", api_key),
        files=files,
        data={
            "from": sender,
            "to": ";".join(valid_recipients),
            "subject": subject,
            "cc": ";".join(valid_cc) if cc else "",
            "html": (
                _parse_newlines_to_html_br(body) if not body_contains_html else body
            ),
        },
    )
    return res

validate_email_addresses(addresses, raise_exception=True)

Validates a list of email addresses using a basic regex. By default, raises a ValueError if any address is invalid.

Parameters:

Name	Type	Description	Default
addresses	List[str]	A list of email addresses to validate.	required
raise_exception	bool	If True, raises a ValueError on invalid addresses.	True

Returns:

Type	Description
Tuple[List[str], List[str]]	Tuple[List[str], List[str]]: A tuple containing two lists: Valid email addresses (stripped of surrounding whitespace) and Invalid email addresses (original input)

Raises:

Type	Description
ValueError	If raise_exception is True and there are invalid addresses or if the input list is empty.

Example

```python from physical_operations_utils.email_utils import _validate_email_addresses

valid, invalid = validate_email_addresses([" john.doe@example.com", "invalid@", "jane@site.org"], False) print(valid, invalid) # Prints (["john.doe@example.com", "jane@site.org"], ["invalid@"])

valid, invalid = validate_email_addresses([" john.doe@example.com", "invalid@", "jane@site.org"], True) # raises ValueError

Source code in physical_operations_utils/email_utils.py

def validate_email_addresses(
    addresses: List[str], raise_exception: bool = True
) -> Tuple[List[str], List[str]]:
    """
    Validates a list of email addresses using a basic regex. By default, raises a ValueError if any address is invalid.

    Args:
        addresses (List[str]): A list of email addresses to validate.
        raise_exception (bool, optional): If True, raises a ValueError on invalid addresses.

    Returns:
        Tuple[List[str], List[str]]: A tuple containing two lists: Valid email addresses (stripped of surrounding whitespace)
            and Invalid email addresses (original input)

    Raises:
        ValueError: If `raise_exception` is True and there are invalid addresses or if the input list is empty.

    Example:
        ```python
        from physical_operations_utils.email_utils import _validate_email_addresses

        valid, invalid = validate_email_addresses(["  john.doe@example.com", "invalid@", "jane@site.org"], False)
        print(valid, invalid) # Prints (["john.doe@example.com", "jane@site.org"], ["invalid@"])

        valid, invalid = validate_email_addresses(["  john.doe@example.com", "invalid@", "jane@site.org"], True) # raises ValueError
    """
    if len(addresses) == 0:
        raise ValueError("No email addresses provided.")
    valid_addresses = []
    invalid_addresses = []

    for address in addresses:
        stripped = address.strip()
        if EMAIL_REGEX.match(stripped):
            valid_addresses.append(stripped)
        else:
            invalid_addresses.append(address)

    if invalid_addresses and raise_exception:
        raise ValueError(f"Invalid email addresses: {', '.join(invalid_addresses)}")

    return valid_addresses, invalid_addresses