prefect_email

EmailServerCredentials

Bases: Block

Block used to manage generic email server authentication. It is recommended you use a Google App Password if you use Gmail.

Attributes:

Name	Type	Description
username	Optional[str]	The username to use for authentication to the server. Unnecessary if SMTP login is not required.
password	SecretStr	The password to use for authentication to the server. Unnecessary if SMTP login is not required.
smtp_server	Union[SMTPServer, str]	Either the hostname of the SMTP server, or one of the keys from the built-in SMTPServer Enum members, like "gmail".
smtp_type	Union[SMTPType, str]	Either "SSL", "STARTTLS", or "INSECURE".
smtp_port	Optional[int]	If provided, overrides the smtp_type's default port number.
verify	Optional[bool]	If False, SSL certificates will not be verified. Default to True.

Example

Load stored email server credentials:

from prefect_email import EmailServerCredentials
email_credentials_block = EmailServerCredentials.load("BLOCK_NAME")

Source code in prefect_email/credentials.py

class EmailServerCredentials(Block):
    """
    Block used to manage generic email server authentication.
    It is recommended you use a
    [Google App Password](https://support.google.com/accounts/answer/185833)
    if you use Gmail.

    Attributes:
        username: The username to use for authentication to the server.
            Unnecessary if SMTP login is not required.
        password: The password to use for authentication to the server.
            Unnecessary if SMTP login is not required.
        smtp_server: Either the hostname of the SMTP server, or one of the
            keys from the built-in SMTPServer Enum members, like "gmail".
        smtp_type: Either "SSL", "STARTTLS", or "INSECURE".
        smtp_port: If provided, overrides the smtp_type's default port number.
        verify: If `False`, SSL certificates will not be verified. Default to `True`.

    Example:
        Load stored email server credentials:
        ```python
        from prefect_email import EmailServerCredentials
        email_credentials_block = EmailServerCredentials.load("BLOCK_NAME")
        ```
    """  # noqa E501

    _block_type_name = "Email Server Credentials"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/82bc6ed16ca42a2252a5512c72233a253b8a58eb-250x250.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-email/credentials/#prefect_email.credentials.EmailServerCredentials"  # noqa

    username: Optional[str] = Field(
        default=None,
        description=(
            "The username to use for authentication to the server. "
            "Unnecessary if SMTP login is not required."
        ),
    )
    password: SecretStr = Field(
        default_factory=partial(SecretStr, ""),
        description=(
            "The password to use for authentication to the server. "
            "Unnecessary if SMTP login is not required."
        ),
    )
    smtp_server: Union[SMTPServer, str] = Field(
        default=SMTPServer.GMAIL,
        description=(
            "Either the hostname of the SMTP server, or one of the "
            "keys from the built-in SMTPServer Enum members, like 'gmail'."
        ),
        title="SMTP Server",
    )
    smtp_type: Union[SMTPType, str] = Field(
        default=SMTPType.SSL,
        description=("Either 'SSL', 'STARTTLS', or 'INSECURE'."),
        title="SMTP Type",
    )
    smtp_port: Optional[int] = Field(
        default=None,
        description=("If provided, overrides the smtp_type's default port number."),
        title="SMTP Port",
    )

    verify: Optional[bool] = Field(
        default=True,
        description=(
            "If `False`, SSL certificates will not be verified. Default to `True`."
        ),
    )

    @field_validator("smtp_server", mode="before")
    def _cast_smtp_server(cls, value: str):
        """
        Cast the smtp_server to an SMTPServer Enum member, if valid.
        """
        return _cast_to_enum(value, SMTPServer)

    @field_validator("smtp_type", mode="before")
    @classmethod
    def _cast_smtp_type(cls, value: str):
        """
        Cast the smtp_type to an SMTPType Enum member, if valid.
        """
        if isinstance(value, int):
            return SMTPType(value)
        return _cast_to_enum(value, SMTPType, restrict=True)

    def get_server(self) -> SMTP:
        """
        Gets an authenticated SMTP server.

        Returns:
            SMTP: An authenticated SMTP server.

        Example:
            Gets a GMail SMTP server through defaults.
            ```python
            from prefect import flow
            from prefect_email import EmailServerCredentials

            @flow
            def example_get_server_flow():
                email_server_credentials = EmailServerCredentials(
                    username="username@gmail.com",
                    password="password",
                )
                server = email_server_credentials.get_server()
                return server

            example_get_server_flow()
            ```
        """
        smtp_server = self.smtp_server
        if isinstance(smtp_server, SMTPServer):
            smtp_server = smtp_server.value

        smtp_type = self.smtp_type
        smtp_port = self.smtp_port
        if smtp_port is None:
            smtp_port = smtp_type.value

        if smtp_type == SMTPType.INSECURE:
            server = SMTP(smtp_server, smtp_port)
        else:
            context = (
                ssl.create_default_context()
                if self.verify
                else ssl._create_unverified_context(protocol=ssl.PROTOCOL_TLS_CLIENT)
            )
            if smtp_type == SMTPType.SSL:
                server = SMTP_SSL(smtp_server, smtp_port, context=context)
            elif smtp_type == SMTPType.STARTTLS:
                server = SMTP(smtp_server, smtp_port)
                server.starttls(context=context)
        if self.username is not None:
            if not self.verify or smtp_type == SMTPType.INSECURE:
                try:
                    logger = get_run_logger()
                except Exception:
                    logger = internal_logger
                logger.warning(
                    """SMTP login is not secure without a verified SSL/TLS or SECURE connection. 
                    Without such a connection, the password may be sent in plain text, 
                    making it vulnerable to interception."""
                )
            server.login(self.username, self.password.get_secret_value())

        return server

get_server()

Gets an authenticated SMTP server.

Returns:

Name	Type	Description
SMTP	SMTP	An authenticated SMTP server.

Example

Gets a GMail SMTP server through defaults.

from prefect import flow
from prefect_email import EmailServerCredentials

@flow
def example_get_server_flow():
    email_server_credentials = EmailServerCredentials(
        username="username@gmail.com",
        password="password",
    )
    server = email_server_credentials.get_server()
    return server

example_get_server_flow()

Source code in prefect_email/credentials.py

def get_server(self) -> SMTP:
    """
    Gets an authenticated SMTP server.

    Returns:
        SMTP: An authenticated SMTP server.

    Example:
        Gets a GMail SMTP server through defaults.
        ```python
        from prefect import flow
        from prefect_email import EmailServerCredentials

        @flow
        def example_get_server_flow():
            email_server_credentials = EmailServerCredentials(
                username="username@gmail.com",
                password="password",
            )
            server = email_server_credentials.get_server()
            return server

        example_get_server_flow()
        ```
    """
    smtp_server = self.smtp_server
    if isinstance(smtp_server, SMTPServer):
        smtp_server = smtp_server.value

    smtp_type = self.smtp_type
    smtp_port = self.smtp_port
    if smtp_port is None:
        smtp_port = smtp_type.value

    if smtp_type == SMTPType.INSECURE:
        server = SMTP(smtp_server, smtp_port)
    else:
        context = (
            ssl.create_default_context()
            if self.verify
            else ssl._create_unverified_context(protocol=ssl.PROTOCOL_TLS_CLIENT)
        )
        if smtp_type == SMTPType.SSL:
            server = SMTP_SSL(smtp_server, smtp_port, context=context)
        elif smtp_type == SMTPType.STARTTLS:
            server = SMTP(smtp_server, smtp_port)
            server.starttls(context=context)
    if self.username is not None:
        if not self.verify or smtp_type == SMTPType.INSECURE:
            try:
                logger = get_run_logger()
            except Exception:
                logger = internal_logger
            logger.warning(
                """SMTP login is not secure without a verified SSL/TLS or SECURE connection. 
                Without such a connection, the password may be sent in plain text, 
                making it vulnerable to interception."""
            )
        server.login(self.username, self.password.get_secret_value())

    return server

SMTPServer

Bases: Enum

Server used to send email.

Source code in prefect_email/credentials.py

class SMTPServer(Enum):
    """
    Server used to send email.
    """

    AOL = "smtp.aol.com"
    ATT = "smtp.mail.att.net"
    COMCAST = "smtp.comcast.net"
    ICLOUD = "smtp.mail.me.com"
    GMAIL = "smtp.gmail.com"
    OUTLOOK = "smtp-mail.outlook.com"
    YAHOO = "smtp.mail.yahoo.com"

SMTPType

Bases: Enum

Protocols used to secure email transmissions.

Source code in prefect_email/credentials.py

class SMTPType(Enum):
    """
    Protocols used to secure email transmissions.
    """

    SSL = 465
    STARTTLS = 587
    INSECURE = 25

email_send_message(subject, msg, email_server_credentials, msg_plain=None, email_from=None, email_to=None, email_to_cc=None, email_to_bcc=None, attachments=None) async

Sends an email message from an authenticated email service over SMTP. Sending messages containing HTML code is supported - the default MIME type is set to the text/html.

Parameters:

Name	Type	Description	Default
subject	str	The subject line of the email.	required
msg	str	The contents of the email, added as html; can be used in combination with msg_plain.	required
msg_plain	Optional[str]	The contents of the email as plain text, can be used in combination with msg.	None
email_to	Optional[Union[str, List[str]]]	The email addresses to send the message to, separated by commas. If a list is provided, will join the items, separated by commas.	None
email_to_cc	Optional[Union[str, List[str]]]	Additional email addresses to send the message to as cc, separated by commas. If a list is provided, will join the items, separated by commas.	None
email_to_bcc	Optional[Union[str, List[str]]]	Additional email addresses to send the message to as bcc, separated by commas. If a list is provided, will join the items, separated by commas.	None
attachments	Optional[List[str]]	Names of files that should be sent as attachment.	None

Returns:

Name	Type	Description
MimeText		The MIME Multipart message of the email.

Example

Sends a notification email to someone@gmail.com.

from prefect import flow
from prefect_email import EmailServerCredentials, email_send_message

@flow
def example_email_send_message_flow():
    email_server_credentials = EmailServerCredentials(
        username="username@email.com",
        password="password",
    )
    subject = email_send_message(
        email_server_credentials=email_server_credentials,
        subject="Example Flow Notification",
        msg="This proves email_send_message works!",
        email_to="someone@email.com",
    )
    return subject

example_email_send_message_flow()

Source code in prefect_email/message.py

@task
async def email_send_message(
    subject: str,
    msg: str,
    email_server_credentials: "EmailServerCredentials",
    msg_plain: Optional[str] = None,
    email_from: Optional[str] = None,
    email_to: Optional[Union[str, List[str]]] = None,
    email_to_cc: Optional[Union[str, List[str]]] = None,
    email_to_bcc: Optional[Union[str, List[str]]] = None,
    attachments: Optional[List[str]] = None,
):
    """
    Sends an email message from an authenticated email service over SMTP.
    Sending messages containing HTML code is supported - the default MIME
    type is set to the text/html.

    Args:
        subject: The subject line of the email.
        msg: The contents of the email, added as html; can be used in
            combination with msg_plain.
        msg_plain: The contents of the email as plain text,
            can be used in combination with msg.
        email_to: The email addresses to send the message to, separated by commas.
            If a list is provided, will join the items, separated by commas.
        email_to_cc: Additional email addresses to send the message to as cc,
            separated by commas. If a list is provided, will join the items,
            separated by commas.
        email_to_bcc: Additional email addresses to send the message to as bcc,
            separated by commas. If a list is provided, will join the items,
            separated by commas.
        attachments: Names of files that should be sent as attachment.

    Returns:
        MimeText: The MIME Multipart message of the email.

    Example:
        Sends a notification email to someone@gmail.com.
        ```python
        from prefect import flow
        from prefect_email import EmailServerCredentials, email_send_message

        @flow
        def example_email_send_message_flow():
            email_server_credentials = EmailServerCredentials(
                username="username@email.com",
                password="password",
            )
            subject = email_send_message(
                email_server_credentials=email_server_credentials,
                subject="Example Flow Notification",
                msg="This proves email_send_message works!",
                email_to="someone@email.com",
            )
            return subject

        example_email_send_message_flow()
        ```
    """
    message = MIMEMultipart()
    message["Subject"] = subject
    message["From"] = email_from or email_server_credentials.username

    email_to_dict = {"To": email_to, "Cc": email_to_cc, "Bcc": email_to_bcc}
    if all(val is None for val in email_to_dict.values()):
        raise ValueError(
            "One of email_to, email_to_cc, or email_to_bcc must be specified"
        )

    for key, val in email_to_dict.items():
        if isinstance(val, list):
            val = ", ".join(val)
        message[key] = val

    # First add the message in plain text, then the HTML version;
    # email clients try to render the last part first
    if msg_plain:
        message.attach(MIMEText(msg_plain, "plain"))
    if msg:
        message.attach(MIMEText(msg, "html"))

    for filepath in attachments or []:
        with open(filepath, "rb") as attachment:
            part = MIMEBase("application", "octet-stream")
            part.set_payload(attachment.read())

        encoders.encode_base64(part)
        filename = os.path.basename(filepath)
        part.add_header(
            "Content-Disposition",
            f"attachment; filename= {filename}",
        )
        message.attach(part)

    with email_server_credentials.get_server() as server:
        partial_send_message = partial(server.send_message, message)
        await to_thread.run_sync(partial_send_message)

    return message