Skip to content

prefect_email.credentials

Credential classes used to perform authenticated interactions with email services

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
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
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
27
28
29
30
31
32
33
34
35
36
37
38
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
17
18
19
20
21
22
23
24
class SMTPType(Enum):
    """
    Protocols used to secure email transmissions.
    """

    SSL = 465
    STARTTLS = 587
    INSECURE = 25