Separate backend each threads on send_queued_mail

post_office/mail.py

@@ -41,11 +41,12 @@
 
 def _send_email(email: Email, log_level: int) -> tuple[bool, Optional[Exception]]:
     try:
-        email.dispatch(log_level=log_level, commit=False, disconnect_after_delivery=False)
-        logger.debug('Successfully sent email #%d' % email.id)
+        connection = connections[email.backend_alias or 'default']
+        email.dispatch(log_level=log_level, commit=False, disconnect_after_delivery=False, connection=connection)
+        logger.debug(f'Successfully sent email #{email.id}')
         return True, None
     except Exception as e:
-        logger.exception('Failed to send email #%d' % email.id)
+        logger.exception(f'Failed to send email #{email.id}')
         return False, e
 
 
@@ -162,17 +163,17 @@ def send(
     try:
         recipients = parse_emails(recipients)
     except ValidationError as e:
-        raise ValidationError('recipients: %s' % e.message)
+        raise ValidationError(f'recipients: {e.message}')
 
     try:
         cc = parse_emails(cc)
     except ValidationError as e:
-        raise ValidationError('c: %s' % e.message)
+        raise ValidationError(f'c: {e.message}')
 
     try:
         bcc = parse_emails(bcc)
     except ValidationError as e:
-        raise ValidationError('bcc: %s' % e.message)
+        raise ValidationError(f'bcc: {e.message}')
 
     if sender is None:
         sender = settings.DEFAULT_FROM_EMAIL
@@ -206,7 +207,7 @@ def send(
             template = get_email_template(template, language)
 
     if backend and backend not in get_available_backends().keys():
-        raise ValueError('%s is not a valid backend alias' % backend)
+        raise ValueError(f'{backend} is not a valid backend alias')
 
     email = create(
         sender,
@@ -296,7 +297,7 @@ def send_queued(processes: int = 1, log_level: Optional[int] = None) -> tuple[in
     total_sent, total_failed, total_requeued = 0, 0, 0
     total_email = len(queued_emails)
 
-    logger.info('Started sending %s emails with %s processes.' % (total_email, processes))
+    logger.info(f'Started sending {total_email} emails with {processes} processes.')
 
     if log_level is None:
         log_level = get_log_level()
@@ -363,7 +364,7 @@ def _send_bulk(
     failed_emails = []  # This is a list of two tuples (email, exception)
     email_count = len(emails)
 
-    logger.info('Process started, sending %s emails' % email_count)
+    logger.info(f'Process started, sending {email_count} emails')
 
     emails_to_send = []
 
@@ -376,7 +377,7 @@ def _send_bulk(
             email.prepare_email_message()
             emails_to_send.append(email)
         except Exception as e:
-            logger.exception('Failed to prepare email #%d' % email.id)
+            logger.exception(f'Failed to prepare email #{email.id}')
             failed_emails.append((email, e))
 
     number_of_threads = min(get_threads_per_process(), email_count)

post_office/models.py

@@ -104,7 +104,7 @@ def __init__(self, *args, **kwargs):
         self._cached_email_message = None
 
     def __repr__(self):
-        return '<%s: %s>' % (self.__class__.__name__, self.to)
+        return f'<{self.__class__.__name__}: {self.to}>'
 
     def __str__(self):
         return f'{", ".join(self.to)}'
@@ -206,12 +206,20 @@ def prepare_email_message(self):
         self._cached_email_message = msg
         return msg
 
-    def dispatch(self, log_level=None, disconnect_after_delivery=True, commit=True):
+    def dispatch(self, log_level=None, disconnect_after_delivery=True, commit=True, connection=None):
         """
         Sends email and log the result.
+
+        If ``connection`` is provided, it overrides the connection embedded in
+        the email message by ``prepare_email_message()``. This allows callers
+        (e.g. worker threads) to supply a thread-local connection rather than
+        reusing one that was opened in a different thread.
         """
         try:
-            self.email_message().send()
+            msg = self.email_message()
+            if connection is not None:
+                msg.connection = connection
+            msg.send()
             status = STATUS.sent
             message = ''
             exception_type = ''
@@ -325,7 +333,7 @@ class Meta:
         ordering = ['name']
 
     def __str__(self):
-        return '%s %s' % (self.name, self.language)
+        return f'{self.name} {self.language}'
 
     def natural_key(self):
         return (self.name, self.language, self.default_template)

tests/test_commands.py

@@ -3,7 +3,7 @@
 
 from django.core.files.base import ContentFile
 from django.core.management import call_command
-from django.test import TestCase, TransactionTestCase
+from django.test import TestCase
 from django.test.utils import override_settings
 from django.utils.timezone import now
 

tests/test_mail.py

@@ -131,31 +131,38 @@ def test_send_bulk(self):
         self.assertEqual(len(mail.outbox), 1)
         self.assertEqual(mail.outbox[0].subject, 'send bulk')
 
-    @override_settings(EMAIL_BACKEND='tests.test_mail.ConnectionTestingBackend')
+    @override_settings(
+        EMAIL_BACKEND='tests.test_mail.ConnectionTestingBackend',
+        POST_OFFICE={
+            'BACKENDS': {
+                'connection_tester': 'tests.test_mail.ConnectionTestingBackend',
+            },
+            'THREADS_PER_PROCESS': 1,
+        },
+    )
     def test_send_bulk_reuses_open_connection(self):
         """
-        Ensure _send_bulk() only opens connection once to send multiple emails.
+        Ensure _send_bulk() opens one connection per thread, not one per email.
+        With THREADS_PER_PROCESS=1: main thread opens one during prepare, worker
+        thread opens one during send — total 2 opens for any number of emails.
         """
         global connection_counter
         self.assertEqual(connection_counter, 0)
-        email = Email.objects.create(
-            to=['to@example.com'],
-            from_email='bob@example.com',
-            subject='',
-            message='',
-            status=STATUS.queued,
-            backend_alias='connection_tester',
-        )
-        email_2 = Email.objects.create(
-            to=['to@example.com'],
-            from_email='bob@example.com',
-            subject='',
-            message='',
-            status=STATUS.queued,
-            backend_alias='connection_tester',
+        email_1, email_2, email_3 = Email.objects.bulk_create(
+            [
+                Email(
+                    to=['to@example.com'],
+                    from_email='bob@example.com',
+                    subject='',
+                    message='',
+                    status=STATUS.queued,
+                    backend_alias='connection_tester',
+                )
+                for _ in range(3)
+            ]
         )
-        _send_bulk([email, email_2])
-        self.assertEqual(connection_counter, 1)
+        _send_bulk([email_1, email_2, email_3])
+        self.assertEqual(connection_counter, 2)
 
     def test_get_queued(self):
         """
@@ -381,9 +388,9 @@ def test_create_with_template_and_empty_context(self):
         email = create(sender='from@example.com', recipients=['to@example.com'], template=template, context=context)
         today = timezone.datetime.today()
         current_year = today.year
-        self.assertEqual(email.subject, 'Subject %d' % current_year)
-        self.assertEqual(email.message, 'Content %d' % current_year)
-        self.assertEqual(email.html_message, 'HTML %d' % current_year)
+        self.assertEqual(email.subject, f'Subject {current_year}')
+        self.assertEqual(email.message, f'Content {current_year}')
+        self.assertEqual(email.html_message, f'HTML {current_year}')
         self.assertEqual(email.context, None)
         self.assertIsNotNone(email.template)
 

tests/test_models.py

@@ -1,7 +1,7 @@
 import json
 import os
-
 from datetime import datetime, timedelta
+from unittest.mock import MagicMock
 
 from django.conf import settings as django_settings, settings
 from django.core import mail
@@ -11,6 +11,7 @@
 from django.forms.models import modelform_factory
 from django.test import TestCase
 from django.utils import timezone
+from django.core.mail.backends.locmem import EmailBackend as LocMemEmailBackend
 
 from post_office.models import Email, Log, PRIORITY, STATUS, EmailTemplate, Attachment
 from post_office.mail import send
@@ -111,6 +112,30 @@ def test_dispatch_with_override_recipients(self):
         self.assertEqual(mail.outbox[0].to, ['override@gmail.com'])
         settings.POST_OFFICE = previous_settings
 
+    def test_dispatch_uses_provided_connection(self):
+        """
+        Ensure dispatch() overrides msg.connection with the explicitly passed
+        connection, even when prepare_email_message() already embedded one.
+        """
+        email = Email.objects.create(
+            to=['to@example.com'],
+            from_email='from@example.com',
+            subject='Test provided connection',
+            message='Message',
+            backend_alias='locmem',
+        )
+
+        mocked_connection = MagicMock()
+        mocked_connection.send_messages.return_value = 1
+
+        # sanity check, original connection embedded in email_message() should be a LocMemEmailBackend instance
+        self.assertTrue(isinstance(email.email_message().connection, LocMemEmailBackend))
+
+        email.dispatch(connection=mocked_connection)
+        # message object's connection should be overridden by the provided one
+        self.assertEqual(email.email_message().connection, mocked_connection)
+        mocked_connection.send_messages.assert_called_once()
+
     def test_status_and_log(self):
         """
         Ensure that status and log are set properly on successful sending