Edits history of script submission #23943 for ' Send Email with Service Account (gmail)'

• python3

  Ap­pro­ved version

  Edit in Windmill

  Embed

  # 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)
  

  Submitted by hugo989 1 day ago