Skip to content

Email

The email adapter provides integration with email services for sending transactional and notification emails.

Ports

Abstract port interface defining the email adapter contract.

archipy.adapters.email.ports.EmailPort

Interface for email sending operations.

This interface defines the contract for email adapters, ensuring a consistent approach to sending emails across different implementations. It provides a comprehensive set of features including support for:

  • Multiple recipients (To, CC, BCC)
  • HTML and plain text content
  • File and in-memory attachments
  • Template-based email rendering

Implementing classes should handle the details of connecting to an email service, managing connections, and ensuring reliable delivery.

Examples:

>>> from archipy.adapters.email.ports import EmailPort
>>>
>>> class CustomEmailAdapter(EmailPort):
...     def __init__(self, config):
...         self.config = config
...
...     def send_email(
...         self,
...         to_email,
...         subject,
...         body,
...         cc=None,
...         bcc=None,
...         attachments=None,
...         html=False,
...         template=None,
...         template_vars=None,
...     ):
...         # Implementation details...
...         pass
Source code in archipy/adapters/email/ports.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 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
class EmailPort:
    """Interface for email sending operations.

    This interface defines the contract for email adapters, ensuring
    a consistent approach to sending emails across different implementations.
    It provides a comprehensive set of features including support for:

    - Multiple recipients (To, CC, BCC)
    - HTML and plain text content
    - File and in-memory attachments
    - Template-based email rendering

    Implementing classes should handle the details of connecting to an
    email service, managing connections, and ensuring reliable delivery.

    Examples:
        >>> from archipy.adapters.email.ports import EmailPort
        >>>
        >>> class CustomEmailAdapter(EmailPort):
        ...     def __init__(self, config):
        ...         self.config = config
        ...
        ...     def send_email(
        ...         self,
        ...         to_email,
        ...         subject,
        ...         body,
        ...         cc=None,
        ...         bcc=None,
        ...         attachments=None,
        ...         html=False,
        ...         template=None,
        ...         template_vars=None,
        ...     ):
        ...         # Implementation details...
        ...         pass
    """

    @abstractmethod
    def send_email(
        self,
        to_email: EmailStr | list[EmailStr],
        subject: str,
        body: str,
        cc: EmailStr | list[EmailStr] | None = None,
        bcc: EmailStr | list[EmailStr] | None = None,
        attachments: list[str | EmailAttachmentDTO] | None = None,
        html: bool = False,
        template: str | None = None,
        template_vars: dict | None = None,
    ) -> None:
        """Send an email with various options and features.

        This method handles the composition and delivery of an email with
        support for multiple recipients, HTML content, templates, and attachments.

        Args:
            to_email: Primary recipient(s) of the email
            subject: Email subject line
            body: Email body content (either plain text or HTML)
            cc: Carbon copy recipient(s)
            bcc: Blind carbon copy recipient(s)
            attachments: List of file paths or EmailAttachmentDTO objects
            html: If True, treats body as HTML content, otherwise plain text
            template: A template string to render using template_vars
            template_vars: Variables to use when rendering the template

        Returns:
            None

        Examples:
            >>> # Simple text email
            >>> adapter.send_email(to_email="user@example.com", subject="Hello", body="This is a test email")
            >>>
            >>> # HTML email with attachment
            >>> adapter.send_email(
            ...     to_email=["user1@example.com", "user2@example.com"],
            ...     subject="Report",
            ...     body="<h1>Monthly Report</h1><p>Please see attached</p>",
            ...     html=True,
            ...     attachments=["path/to/report.pdf"],
            ... )
            >>>
            >>> # Template-based email
            >>> template = "Hello {{ name }}, your account expires on {{ date }}"
            >>> adapter.send_email(
            ...     to_email="user@example.com",
            ...     subject="Account Expiration",
            ...     body="",  # Body will be rendered from template
            ...     template=template,
            ...     template_vars={"name": "John", "date": "2023-12-31"},
            ... )
        """
        raise NotImplementedError

archipy.adapters.email.ports.EmailPort.send_email abstractmethod 

send_email(
    to_email: EmailStr | list[EmailStr],
    subject: str,
    body: str,
    cc: EmailStr | list[EmailStr] | None = None,
    bcc: EmailStr | list[EmailStr] | None = None,
    attachments: list[str | EmailAttachmentDTO]
    | None = None,
    html: bool = False,
    template: str | None = None,
    template_vars: dict | None = None,
) -> None

Send an email with various options and features.

This method handles the composition and delivery of an email with support for multiple recipients, HTML content, templates, and attachments.

Parameters:

Name Type Description Default
to_email EmailStr | list[EmailStr]

Primary recipient(s) of the email

 required
subject str

Email subject line

 required
body str

Email body content (either plain text or HTML)

 required
cc EmailStr | list[EmailStr] | None

Carbon copy recipient(s)

 None
bcc EmailStr | list[EmailStr] | None

Blind carbon copy recipient(s)

 None
attachments list[str | EmailAttachmentDTO] | None

List of file paths or EmailAttachmentDTO objects

 None
html bool

If True, treats body as HTML content, otherwise plain text

 False
template str | None

A template string to render using template_vars

 None
template_vars dict | None

Variables to use when rendering the template

 None

Returns:

Type Description
None

None

Examples:

>>> # Simple text email
>>> adapter.send_email(to_email="user@example.com", subject="Hello", body="This is a test email")
>>>
>>> # HTML email with attachment
>>> adapter.send_email(
...     to_email=["user1@example.com", "user2@example.com"],
...     subject="Report",
...     body="<h1>Monthly Report</h1><p>Please see attached</p>",
...     html=True,
...     attachments=["path/to/report.pdf"],
... )
>>>
>>> # Template-based email
>>> template = "Hello {{ name }}, your account expires on {{ date }}"
>>> adapter.send_email(
...     to_email="user@example.com",
...     subject="Account Expiration",
...     body="",  # Body will be rendered from template
...     template=template,
...     template_vars={"name": "John", "date": "2023-12-31"},
... )
Source code in archipy/adapters/email/ports.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 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
@abstractmethod
def send_email(
    self,
    to_email: EmailStr | list[EmailStr],
    subject: str,
    body: str,
    cc: EmailStr | list[EmailStr] | None = None,
    bcc: EmailStr | list[EmailStr] | None = None,
    attachments: list[str | EmailAttachmentDTO] | None = None,
    html: bool = False,
    template: str | None = None,
    template_vars: dict | None = None,
) -> None:
    """Send an email with various options and features.

    This method handles the composition and delivery of an email with
    support for multiple recipients, HTML content, templates, and attachments.

    Args:
        to_email: Primary recipient(s) of the email
        subject: Email subject line
        body: Email body content (either plain text or HTML)
        cc: Carbon copy recipient(s)
        bcc: Blind carbon copy recipient(s)
        attachments: List of file paths or EmailAttachmentDTO objects
        html: If True, treats body as HTML content, otherwise plain text
        template: A template string to render using template_vars
        template_vars: Variables to use when rendering the template

    Returns:
        None

    Examples:
        >>> # Simple text email
        >>> adapter.send_email(to_email="user@example.com", subject="Hello", body="This is a test email")
        >>>
        >>> # HTML email with attachment
        >>> adapter.send_email(
        ...     to_email=["user1@example.com", "user2@example.com"],
        ...     subject="Report",
        ...     body="<h1>Monthly Report</h1><p>Please see attached</p>",
        ...     html=True,
        ...     attachments=["path/to/report.pdf"],
        ... )
        >>>
        >>> # Template-based email
        >>> template = "Hello {{ name }}, your account expires on {{ date }}"
        >>> adapter.send_email(
        ...     to_email="user@example.com",
        ...     subject="Account Expiration",
        ...     body="",  # Body will be rendered from template
        ...     template=template,
        ...     template_vars={"name": "John", "date": "2023-12-31"},
        ... )
    """
    raise NotImplementedError

options: show_root_toc_entry: false heading_level: 3

Adapters

Concrete email adapter implementing SMTP-based email sending with ArchiPy conventions.

archipy.adapters.email.adapters.logger module-attribute 

logger = getLogger(__name__)

archipy.adapters.email.adapters.EmailConnectionManager

Manages SMTP connections with connection pooling and timeout handling.

Source code in archipy/adapters/email/adapters.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class EmailConnectionManager:
    """Manages SMTP connections with connection pooling and timeout handling."""

    def __init__(self, config: EmailConfig) -> None:
        self.config = config
        self.smtp_connection: smtplib.SMTP | None = None
        self.last_used: datetime | None = None

    def connect(self) -> None:
        """Establish SMTP connection with authentication."""
        if not self.config.SMTP_SERVER:
            raise InvalidArgumentError("SMTP_SERVER is required for email connection")

        try:
            self.smtp_connection = smtplib.SMTP(
                self.config.SMTP_SERVER,
                self.config.SMTP_PORT,
                timeout=self.config.CONNECTION_TIMEOUT,
            )
            self.smtp_connection.starttls()
            if self.config.USERNAME and self.config.PASSWORD:
                self.smtp_connection.login(self.config.USERNAME, self.config.PASSWORD)
            self.last_used = datetime.now()
        except Exception as e:
            BaseUtils.capture_exception(e)
            self.smtp_connection = None

    def disconnect(self) -> None:
        """Close SMTP connection safely."""
        try:
            if self.smtp_connection:
                self.smtp_connection.quit()
                self.smtp_connection = None
        except Exception as e:
            BaseUtils.capture_exception(e)
        finally:
            self.smtp_connection = None

    def refresh_if_needed(self) -> None:
        """Refresh connection if needed based on timeout."""
        if not self.smtp_connection or not self.last_used:
            self.connect()
            return

        time_diff = (datetime.now() - self.last_used).total_seconds()
        if time_diff > 300:  # Refresh after 5 minutes
            self.disconnect()
            self.connect()

archipy.adapters.email.adapters.EmailConnectionManager.config instance-attribute 

config = config

archipy.adapters.email.adapters.EmailConnectionManager.smtp_connection instance-attribute 

smtp_connection: SMTP | None = None

archipy.adapters.email.adapters.EmailConnectionManager.last_used instance-attribute 

last_used: datetime | None = None

archipy.adapters.email.adapters.EmailConnectionManager.connect 

connect() -> None

Establish SMTP connection with authentication.

Source code in archipy/adapters/email/adapters.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def connect(self) -> None:
    """Establish SMTP connection with authentication."""
    if not self.config.SMTP_SERVER:
        raise InvalidArgumentError("SMTP_SERVER is required for email connection")

    try:
        self.smtp_connection = smtplib.SMTP(
            self.config.SMTP_SERVER,
            self.config.SMTP_PORT,
            timeout=self.config.CONNECTION_TIMEOUT,
        )
        self.smtp_connection.starttls()
        if self.config.USERNAME and self.config.PASSWORD:
            self.smtp_connection.login(self.config.USERNAME, self.config.PASSWORD)
        self.last_used = datetime.now()
    except Exception as e:
        BaseUtils.capture_exception(e)
        self.smtp_connection = None

archipy.adapters.email.adapters.EmailConnectionManager.disconnect 

disconnect() -> None

Close SMTP connection safely.

Source code in archipy/adapters/email/adapters.py 
59
60
61
62
63
64
65
66
67
68
def disconnect(self) -> None:
    """Close SMTP connection safely."""
    try:
        if self.smtp_connection:
            self.smtp_connection.quit()
            self.smtp_connection = None
    except Exception as e:
        BaseUtils.capture_exception(e)
    finally:
        self.smtp_connection = None

archipy.adapters.email.adapters.EmailConnectionManager.refresh_if_needed 

refresh_if_needed() -> None

Refresh connection if needed based on timeout.

Source code in archipy/adapters/email/adapters.py 
70
71
72
73
74
75
76
77
78
79
def refresh_if_needed(self) -> None:
    """Refresh connection if needed based on timeout."""
    if not self.smtp_connection or not self.last_used:
        self.connect()
        return

    time_diff = (datetime.now() - self.last_used).total_seconds()
    if time_diff > 300:  # Refresh after 5 minutes
        self.disconnect()
        self.connect()

archipy.adapters.email.adapters.EmailConnectionPool

Connection pool for managing multiple SMTP connections.

Source code in archipy/adapters/email/adapters.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class EmailConnectionPool:
    """Connection pool for managing multiple SMTP connections."""

    def __init__(self, config: EmailConfig) -> None:
        self.config = config
        self.pool: Queue[EmailConnectionManager] = Queue(maxsize=config.POOL_SIZE)
        self._initialize_pool()

    def _initialize_pool(self) -> None:
        for _ in range(self.config.POOL_SIZE):
            connection = EmailConnectionManager(self.config)
            self.pool.put(connection)

    def get_connection(self) -> EmailConnectionManager:
        """Get a connection from the pool."""
        connection = self.pool.get()
        connection.refresh_if_needed()
        return connection

    def return_connection(self, connection: EmailConnectionManager) -> None:
        """Return a connection to the pool."""
        connection.last_used = datetime.now()
        self.pool.put(connection)

archipy.adapters.email.adapters.EmailConnectionPool.config instance-attribute 

config = config

archipy.adapters.email.adapters.EmailConnectionPool.pool instance-attribute 

pool: Queue[EmailConnectionManager] = Queue(
    maxsize=POOL_SIZE
)

archipy.adapters.email.adapters.EmailConnectionPool.get_connection 

get_connection() -> EmailConnectionManager

Get a connection from the pool.

Source code in archipy/adapters/email/adapters.py 
95
96
97
98
99
def get_connection(self) -> EmailConnectionManager:
    """Get a connection from the pool."""
    connection = self.pool.get()
    connection.refresh_if_needed()
    return connection

archipy.adapters.email.adapters.EmailConnectionPool.return_connection 

return_connection(
    connection: EmailConnectionManager,
) -> None

Return a connection to the pool.

Source code in archipy/adapters/email/adapters.py 
101
102
103
104
def return_connection(self, connection: EmailConnectionManager) -> None:
    """Return a connection to the pool."""
    connection.last_used = datetime.now()
    self.pool.put(connection)

archipy.adapters.email.adapters.AttachmentHandler

Enhanced attachment handler with better type safety and validation.

Source code in archipy/adapters/email/adapters.py 
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
class AttachmentHandler:
    """Enhanced attachment handler with better type safety and validation."""

    @staticmethod
    def create_attachment(
        source: str | bytes | BinaryIO | HttpUrl,
        filename: str,
        attachment_type: EmailAttachmentType,
        content_type: str | None = None,
        content_disposition: EmailAttachmentDispositionType = EmailAttachmentDispositionType.ATTACHMENT,
        content_id: str | None = None,
        max_size: int | None = None,
    ) -> EmailAttachmentDTO:
        """Create an attachment with validation."""
        if max_size is None:
            max_size = BaseConfig.global_config().EMAIL.ATTACHMENT_MAX_SIZE
        try:
            processed_content = AttachmentHandler._process_source(source, attachment_type)

            return EmailAttachmentDTO(
                content=processed_content,
                filename=filename,
                content_type=content_type,
                content_disposition=content_disposition,
                content_id=content_id,
                attachment_type=attachment_type,
                max_size=max_size,
            )
        except Exception as exception:
            raise InvalidArgumentError(f"Failed to create attachment: {exception!s}") from exception

    @staticmethod
    def _process_source(source: str | bytes | BinaryIO | HttpUrl, attachment_type: EmailAttachmentType) -> bytes:
        """Process different types of attachment sources."""
        if attachment_type == EmailAttachmentType.FILE:
            if isinstance(source, str):
                return Path(source).read_bytes()
            if isinstance(source, os.PathLike):
                return Path(os.fspath(source)).read_bytes()
            raise ValueError(f"File attachment type requires string path, got {type(source)}")
        elif attachment_type == EmailAttachmentType.BASE64:
            if isinstance(source, str | bytes):
                return base64.b64decode(source)
            raise ValueError(f"Base64 attachment type requires str or bytes, got {type(source)}")
        elif attachment_type == EmailAttachmentType.URL:
            if isinstance(source, str | HttpUrl):
                response = requests.get(str(source), timeout=30)
                response.raise_for_status()
                return bytes(response.content)
            raise ValueError(f"URL attachment type requires str or HttpUrl, got {type(source)}")
        elif attachment_type == EmailAttachmentType.BINARY:
            if isinstance(source, bytes):
                return source
            if isinstance(source, BinaryIO):
                return source.read()
            if hasattr(source, "read"):
                read_method = source.read
                if callable(read_method):
                    read_callable = cast("Callable[[], Any]", read_method)
                    result = read_callable()
                    if isinstance(result, bytes):
                        return result
                    if isinstance(result, str):
                        return result.encode("utf-8")
                    raise ValueError(f"read() method returned unexpected type: {type(result)}")
            raise ValueError(f"Invalid binary source type: {type(source)}")
        raise ValueError(f"Unsupported attachment type: {attachment_type}")

    @staticmethod
    def process_attachment(msg: MIMEMultipart, attachment: EmailAttachmentDTO) -> None:
        """Process and attach the attachment to the email message."""
        content = AttachmentHandler._get_content(attachment)
        part = AttachmentHandler._create_mime_part(content, attachment)

        # Add headers
        part.add_header("Content-Disposition", attachment.content_disposition.value, filename=attachment.filename)

        if attachment.content_id:
            part.add_header("Content-ID", attachment.content_id)

        msg.attach(part)

    @staticmethod
    def _get_content(attachment: EmailAttachmentDTO) -> bytes:
        """Get content as bytes from attachment."""
        if isinstance(attachment.content, str | bytes):
            return attachment.content if isinstance(attachment.content, bytes) else attachment.content.encode()
        return attachment.content.read()

    @staticmethod
    def _create_mime_part(
        content: bytes,
        attachment: EmailAttachmentDTO,
    ) -> MIMEText | MIMEImage | MIMEAudio | MIMEBase:
        """Create appropriate MIME part based on content type."""
        if not attachment.content_type:
            raise ValueError("Content type is required for attachment")
        main_type, sub_type = attachment.content_type.split("/", 1)

        if main_type == "text":
            return MIMEText(content.decode(), sub_type)
        if main_type == "image":
            return MIMEImage(content, _subtype=sub_type)
        if main_type == "audio":
            return MIMEAudio(content, _subtype=sub_type)
        part = MIMEBase(main_type, sub_type)
        part.set_payload(content)
        encoders.encode_base64(part)
        return part

archipy.adapters.email.adapters.AttachmentHandler.create_attachment staticmethod 

create_attachment(
    source: str | bytes | BinaryIO | HttpUrl,
    filename: str,
    attachment_type: EmailAttachmentType,
    content_type: str | None = None,
    content_disposition: EmailAttachmentDispositionType = EmailAttachmentDispositionType.ATTACHMENT,
    content_id: str | None = None,
    max_size: int | None = None,
) -> EmailAttachmentDTO

Create an attachment with validation.

Source code in archipy/adapters/email/adapters.py 
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
@staticmethod
def create_attachment(
    source: str | bytes | BinaryIO | HttpUrl,
    filename: str,
    attachment_type: EmailAttachmentType,
    content_type: str | None = None,
    content_disposition: EmailAttachmentDispositionType = EmailAttachmentDispositionType.ATTACHMENT,
    content_id: str | None = None,
    max_size: int | None = None,
) -> EmailAttachmentDTO:
    """Create an attachment with validation."""
    if max_size is None:
        max_size = BaseConfig.global_config().EMAIL.ATTACHMENT_MAX_SIZE
    try:
        processed_content = AttachmentHandler._process_source(source, attachment_type)

        return EmailAttachmentDTO(
            content=processed_content,
            filename=filename,
            content_type=content_type,
            content_disposition=content_disposition,
            content_id=content_id,
            attachment_type=attachment_type,
            max_size=max_size,
        )
    except Exception as exception:
        raise InvalidArgumentError(f"Failed to create attachment: {exception!s}") from exception

archipy.adapters.email.adapters.AttachmentHandler.process_attachment staticmethod 

process_attachment(
    msg: MIMEMultipart, attachment: EmailAttachmentDTO
) -> None

Process and attach the attachment to the email message.

Source code in archipy/adapters/email/adapters.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
@staticmethod
def process_attachment(msg: MIMEMultipart, attachment: EmailAttachmentDTO) -> None:
    """Process and attach the attachment to the email message."""
    content = AttachmentHandler._get_content(attachment)
    part = AttachmentHandler._create_mime_part(content, attachment)

    # Add headers
    part.add_header("Content-Disposition", attachment.content_disposition.value, filename=attachment.filename)

    if attachment.content_id:
        part.add_header("Content-ID", attachment.content_id)

    msg.attach(part)

archipy.adapters.email.adapters.EmailAdapter

Bases: EmailPort

Email adapter implementing EmailPort for sending emails with SMTP.

Source code in archipy/adapters/email/adapters.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
class EmailAdapter(EmailPort):
    """Email adapter implementing EmailPort for sending emails with SMTP."""

    def __init__(self, config: EmailConfig | None = None) -> None:
        self.config = config or BaseConfig.global_config().EMAIL
        self.connection_pool = EmailConnectionPool(self.config)

    @override
    def send_email(
        self,
        to_email: EmailStr | list[EmailStr],
        subject: str,
        body: str,
        cc: EmailStr | list[EmailStr] | None = None,
        bcc: EmailStr | list[EmailStr] | None = None,
        attachments: list[str | EmailAttachmentDTO] | None = None,
        html: bool = False,
        template: str | None = None,
        template_vars: dict | None = None,
    ) -> None:
        """Send email with advanced features and connection pooling."""
        connection: EmailConnectionManager | None = None
        try:
            connection = self.connection_pool.get_connection()
            msg = self._create_message(
                to_email=to_email,
                subject=subject,
                body=body,
                cc=cc,
                bcc=bcc,
                attachments=attachments,
                html=html,
                template=template,
                template_vars=template_vars,
            )

            recipients = self._get_all_recipients(to_email, cc, bcc)

            for attempt in range(self.config.MAX_RETRIES):
                try:
                    if connection.smtp_connection:
                        connection.smtp_connection.send_message(msg, to_addrs=recipients)
                        logger.debug(f"Email sent successfully to {to_email}")
                        return
                    else:
                        connection.connect()
                except Exception as e:
                    if attempt == self.config.MAX_RETRIES - 1:
                        BaseUtils.capture_exception(e)
                    connection.connect()  # Retry with fresh connection

        except Exception as e:
            BaseUtils.capture_exception(e)
        finally:
            if connection:
                self.connection_pool.return_connection(connection)

    def _create_message(
        self,
        to_email: EmailStr | list[EmailStr],
        subject: str,
        body: str,
        cc: EmailStr | list[EmailStr] | None = None,
        bcc: EmailStr | list[EmailStr] | None = None,
        attachments: list[str | EmailAttachmentDTO] | None = None,
        html: bool = False,
        template: str | None = None,
        template_vars: dict | None = None,
    ) -> MIMEMultipart:
        msg = MIMEMultipart()
        msg["From"] = self.config.USERNAME or "no-reply@example.com"
        msg["To"] = to_email if isinstance(to_email, str) else ", ".join(to_email)
        msg["Subject"] = subject

        if cc:
            msg["Cc"] = cc if isinstance(cc, str) else ", ".join(cc)
        if bcc:
            msg["Bcc"] = bcc if isinstance(bcc, str) else ", ".join(bcc)

        if template:
            body = Template(template).render(**(template_vars or {}))

        msg.attach(MIMEText(body, "html" if html else "plain"))

        if attachments:
            for attachment in attachments:
                if isinstance(attachment, str):
                    # Treat as file path
                    attachment_obj = AttachmentHandler.create_attachment(
                        source=attachment,
                        filename=Path(attachment).name,
                        attachment_type=EmailAttachmentType.FILE,
                    )
                else:
                    attachment_obj = attachment
                AttachmentHandler.process_attachment(msg, attachment_obj)

        return msg

    @staticmethod
    def _get_all_recipients(
        to_email: EmailStr | list[EmailStr],
        cc: EmailStr | list[EmailStr] | None,
        bcc: EmailStr | list[EmailStr] | None,
    ) -> list[str]:
        """Get list of all recipients."""
        recipients = []

        # Add primary recipients
        if isinstance(to_email, str):
            recipients.append(to_email)
        else:
            recipients.extend(to_email)

        # Add CC recipients
        if cc:
            if isinstance(cc, str):
                recipients.append(cc)
            else:
                recipients.extend(cc)

        # Add BCC recipients
        if bcc:
            if isinstance(bcc, str):
                recipients.append(bcc)
            else:
                recipients.extend(bcc)

        return recipients

archipy.adapters.email.adapters.EmailAdapter.config instance-attribute 

config = config or EMAIL

archipy.adapters.email.adapters.EmailAdapter.connection_pool instance-attribute 

connection_pool = EmailConnectionPool(config)

archipy.adapters.email.adapters.EmailAdapter.send_email 

send_email(
    to_email: EmailStr | list[EmailStr],
    subject: str,
    body: str,
    cc: EmailStr | list[EmailStr] | None = None,
    bcc: EmailStr | list[EmailStr] | None = None,
    attachments: list[str | EmailAttachmentDTO]
    | None = None,
    html: bool = False,
    template: str | None = None,
    template_vars: dict | None = None,
) -> None

Send email with advanced features and connection pooling.

Source code in archipy/adapters/email/adapters.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
@override
def send_email(
    self,
    to_email: EmailStr | list[EmailStr],
    subject: str,
    body: str,
    cc: EmailStr | list[EmailStr] | None = None,
    bcc: EmailStr | list[EmailStr] | None = None,
    attachments: list[str | EmailAttachmentDTO] | None = None,
    html: bool = False,
    template: str | None = None,
    template_vars: dict | None = None,
) -> None:
    """Send email with advanced features and connection pooling."""
    connection: EmailConnectionManager | None = None
    try:
        connection = self.connection_pool.get_connection()
        msg = self._create_message(
            to_email=to_email,
            subject=subject,
            body=body,
            cc=cc,
            bcc=bcc,
            attachments=attachments,
            html=html,
            template=template,
            template_vars=template_vars,
        )

        recipients = self._get_all_recipients(to_email, cc, bcc)

        for attempt in range(self.config.MAX_RETRIES):
            try:
                if connection.smtp_connection:
                    connection.smtp_connection.send_message(msg, to_addrs=recipients)
                    logger.debug(f"Email sent successfully to {to_email}")
                    return
                else:
                    connection.connect()
            except Exception as e:
                if attempt == self.config.MAX_RETRIES - 1:
                    BaseUtils.capture_exception(e)
                connection.connect()  # Retry with fresh connection

    except Exception as e:
        BaseUtils.capture_exception(e)
    finally:
        if connection:
            self.connection_pool.return_connection(connection)

options: show_root_toc_entry: false heading_level: 3