Emails

Emails(baze)

Class used for handling emails.

Source code in echo_baze/baze_root.py

def __init__(self, baze: e_bz.Baze) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    baze : Baze
        Top level object carrying all functionality and the connection handler.

    """
    # check inputs
    if not isinstance(baze, e_bz.Baze):
        raise ValueError(f"baze must be of type Baze, not {type(baze)}")

    self.baze: e_bz.Baze = baze

send(subject, message, users=None, roles=None, emails=None, files=None, recipient_errors='ignore')

Sends an email using Bazefield bulk email API. This feature can be accessed in the portal under Administration > Bulk Email.

Parameters:

• subject

  (str) –

  Subject of the email.

• message

  (str) –

  Body of the email. Can be formatted with HTML. All \n will be replaced with
  automatically.

• users

  (list[str] | None, default: None ) –

  List of user names to send emails to. By default None

• roles

  (list[str] | None, default: None ) –

  List of role names to send emails to. By default None

• emails

  (list[EmailStr] | None, default: None ) –

  List of email addresses to send emails to. By default None

• files

  (dict[str, Path] | None, default: None ) –

  Dict containing the files to be sent. Must be in the format {file_name: file_path}, where file_name includes the extension. If set to None, no attachments will be added. By default None

• recipient_errors

  (Literal['ignore', 'raise'], default: 'ignore' ) –

  What to do if one of the recipients is not found.

  If "ignore", the email will be sent to the other recipients.

  If "raise", a ValueError will be raised.

  By default "ignore"

Source code in echo_baze/emails.py

@validate_call
def send(
    self,
    subject: str,
    message: str,
    users: list[str] | None = None,
    roles: list[str] | None = None,
    emails: list[EmailStr] | None = None,
    files: dict[str, Path] | None = None,
    recipient_errors: Literal["ignore", "raise"] = "ignore",
) -> None:
    r"""Sends an email using Bazefield bulk email API. This feature can be accessed in the portal under `Administration > Bulk Email`.

    Parameters
    ----------
    subject : str
        Subject of the email.
    message : str
        Body of the email. Can be formatted with HTML.
        All \n will be replaced with <br> automatically.
    users : list[str] | None, optional
        List of user names to send emails to. By default None
    roles : list[str] | None, optional
        List of role names to send emails to. By default None
    emails : list[EmailStr] | None, optional
        List of email addresses to send emails to. By default None
    files : dict[str, Path] | None, optional
        Dict containing the files to be sent. Must be in the format {file_name: file_path}, where file_name includes the extension.
        If set to None, no attachments will be added.
        By default None
    recipient_errors : Literal["ignore", "raise"], optional
        What to do if one of the recipients is not found.

        If "ignore", the email will be sent to the other recipients.

        If "raise", a ValueError will be raised.

        By default "ignore"

    """
    # checking if there is at least one user or role
    if users is None and roles is None and emails is None:
        raise ValueError("Please specify at least one role, user or email!")
    if (
        (users is not None and len(users) == 0)
        and (roles is not None and len(roles) == 0)
        and (emails is not None and len(emails)) == 0
    ):
        raise ValueError("Please specify at least one role, user or email!")

    # getting users and role ids
    user_ids = self.baze.users.get_ids()
    role_ids = self.baze.roles.get_ids()

    if recipient_errors == "raise":
        if users is not None and (wrong_users := [user for user in users if user not in user_ids]):
            raise ValueError(f"Users {wrong_users} do not exist!")
        if roles is not None and (wrong_roles := [role for role in roles if role not in role_ids]):
            raise ValueError(f"Roles {wrong_roles} do not exist!")

    # filtering users and roles
    user_ids = [user_id for user_name, user_id in user_ids.items() if users is not None and user_name in users]
    role_ids = [role_id for role_name, role_id in role_ids.items() if roles is not None and role_name in roles]

    # raising error if no users or roles are found
    if not user_ids and not role_ids and not emails:
        raise ValueError("No users or roles found!")

    # checking if all files exist
    if files is not None:
        for file_name in files:
            if not isinstance(files[file_name], Path):
                files[file_name] = Path(files[file_name])
            if not files[file_name].is_file():
                raise ValueError(f"File '{file_name}' with path '{files[file_name]}' does not exist!")

    # replacing newlines with <br>
    message = message.replace("\n", "<br>")
    # URL encoding
    message = quote(message)

    # creating the payload
    payload = {
        "UserIds": json.dumps(user_ids),
        "RoleIds": json.dumps(role_ids),
        "Recipients": json.dumps(emails if emails is not None else []),
        "Subject": subject,
        "Message": message,
        "IsHtml": "true",
        "UrlEncoded": "true",
        "attachmentUrls": "",
    }

    # adding files to payload
    if files is not None:
        for i, t in enumerate(files.items()):
            file_name, file_path = t
            mime_type, _ = mimetypes.guess_type(file_path)
            if mime_type is None:
                raise TypeError(f"Could not guess MIME type for file {file_path}.")
            payload[f"File{i}"] = (file_name, open(file_path, "rb"), mime_type)  # noqa

    mp_encoder = MultipartEncoder(fields=payload)

    # sending request
    response = self.baze.conn.post(
        endpoint="email",
        data=mp_encoder.to_string(),  # The MultipartEncoder is posted as data, don't use files=...!
        headers={"Content-Type": mp_encoder.content_type},  # The MultipartEncoder provides the content-type header with the boundary
    )
    self._handle_http_errors(response)

    logger.info(f"Email sent to users {users} and roles {roles}")