Send Email with Service Account

# py312

# requirements:

# google-auth==2.62.1

# requests==2.34.2

import base64

import mimetypes

from email.message import EmailMessage

from typing import TypedDict

from google.auth.transport.requests import AuthorizedSession

from google.oauth2 import service_account as google_sa

class google_service_account(TypedDict):

    client_email: str

    private_key: str

    token_uri: str

GMAIL_SEND = "https://gmail.googleapis.com/gmail/v1/users/me/messages/send"

SCOPE = "https://www.googleapis.com/auth/gmail.send"

def main(

    service_account: google_service_account,

    delegated_user: str,

    to: str,

    subject: str,

    body: str,

    cc: str = "",

    bcc: str = "",

    send_as: str = "",

    is_html: bool = False,

    thread_id: str = "",

    in_reply_to: str = "",

    attachments: list[dict] = [],

):

    """

    Send Email with Service Account

    Send an email from a Google Workspace mailbox with a service account and domain-wide delegation, without an OAuth connection. `delegated_user` is the mailbox to send from: the service account's client ID must be allowed the https://www.googleapis.com/auth/gmail.send scope in the Admin console (Security > Access and data control > API controls > Domain-wide delegation). Supports cc/bcc, HTML body, `attachments` as text (e.g. CSV) or base64 (`[{filename, content, content_type?}]`: content is text for text/* types and base64 otherwise, type guessed from the filename) and a `send_as` alias of that mailbox (an unverified alias falls back to the primary address). To reply in a thread, pass the original's `thread_id` and its `Message-ID` header as `in_reply_to`, and keep the same subject.

    """

    s = session(service_account, delegated_user)

    msg = EmailMessage()

    if send_as:

        msg["From"] = send_as

    msg["To"] = to

    if cc:

        msg["Cc"] = cc

    if bcc:

        msg["Bcc"] = bcc

    msg["Subject"] = subject

    if in_reply_to:

        msg["In-Reply-To"] = in_reply_to

        msg["References"] = in_reply_to

    msg.set_content(body, subtype="html" if is_html else "plain")

    for a in attachments or []:

        if not a.get("filename"):

            raise ValueError("each attachment needs a filename")

        ctype = (

            a.get("content_type")

            or mimetypes.guess_type(a["filename"])[0]

            or "text/plain"

        )

        maintype, _, subtype = ctype.split(";")[0].strip().partition("/")

        if not subtype:

            raise ValueError(f"{a['filename']}: content_type must be type/subtype")

        content = a.get("content") or ""

        if maintype == "text":

            msg.add_attachment(content, subtype=subtype, filename=a["filename"])

        else:

            try:

                data = base64.b64decode(content, validate=True)

            except ValueError:

                raise ValueError(

                    f"{a['filename']}: {ctype} content must be base64, or set a text/* content_type"

                ) from None

            msg.add_attachment(

                data,

                maintype=maintype,

                subtype=subtype,

                filename=a["filename"],

            )

    payload = {"raw": base64.urlsafe_b64encode(msg.as_bytes()).decode()}

    if thread_id:

        payload["threadId"] = thread_id

    r = s.post(GMAIL_SEND, json=payload)

    if not r.ok:

        raise Exception(f"{r.status_code} {r.text}")

    return r.json()

# Domain-wide delegation: google-auth signs the JWT bearer grant for the delegated mailbox

# and fetches the token on the first request; a new token per run, valid one hour.

def session(sa: google_service_account, user: str) -> AuthorizedSession:

    if not sa.get("client_email") or not sa.get("private_key"):

        raise ValueError("service_account needs client_email and private_key")

    if not user:

        raise ValueError("delegated_user is required: the Workspace mailbox to send from")

    info = {

        # Drop fields left empty in the form: google-auth would send private_key_id "" as the JWT kid.

        **{k: v for k, v in sa.items() if v},

        # Keys pasted into a single-line field keep their newlines as literal \n.

        "private_key": sa["private_key"].replace("\\n", "\n"),

        "token_uri": sa.get("token_uri") or "https://oauth2.googleapis.com/token",

    }

    creds = google_sa.Credentials.from_service_account_info(

        info, scopes=[SCOPE]

    ).with_subject(user)

    return AuthorizedSession(creds)