GovOPlaN/govoplan-mail
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add reusable SMTP batch sessions

Browse Source
This commit is contained in:
zemion
2026-08-20 17:39:00 +02:00
parent 169b81d9db
commit a044fee379
5 changed files with 616 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_mail/backend/capabilities.py
+143 -2
View File
@@ -1,9 +1,11 @@
from __future__ import annotations
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass
from email.message import EmailMessage
from email.utils import formatdate, make_msgid
from typing import Any
from typing import Any, Iterator
from sqlalchemy.orm import Session
@@ -40,13 +42,27 @@ from govoplan_mail.backend.sending.imap import (
append_message_to_sent,
)
from govoplan_mail.backend.sending.rate_limit import wait_for_rate_limit
from govoplan_mail.backend.sending.smtp import SmtpConfigurationError, SmtpSendError, send_email_bytes
from govoplan_mail.backend.sending.smtp import (
SmtpBatchSession,
SmtpConfigurationError,
SmtpSendError,
send_email_bytes,
)
_ACTIVE_SMTP_BATCH: ContextVar[SmtpBatchSession | None] = ContextVar(
"govoplan_mail_active_smtp_batch",
default=None,
)
@dataclass(frozen=True, slots=True)
class CampaignSmtpDeliveryResult:
envelope_recipients: list[str]
refused_recipients: dict[str, dict[str, int | str]]
connection_sequence: int = 1
session_reused: bool = False
reconnect_count: int = 0
@property
def accepted_count(self) -> int:
@@ -58,6 +74,23 @@ class CampaignImapAppendResult:
folder: str
@dataclass(frozen=True, slots=True)
class CampaignSmtpBatchState:
session: SmtpBatchSession
@property
def status(self) -> str:
return "ready"
@property
def connection_count(self) -> int:
return self.session.connection_count
@property
def reconnect_count(self) -> int:
return self.session.reconnect_count
def _sanitized_refusals(
refused_recipients: dict[str, tuple[int, bytes | str]],
) -> dict[str, dict[str, int | str]]:
@@ -95,6 +128,9 @@ def _sanitized_smtp_error(exc: SmtpSendError) -> SmtpSendError:
message,
temporary=exc.temporary,
outcome_unknown=exc.outcome_unknown,
systemic=exc.systemic,
reason_code=exc.reason_code,
phase=exc.phase,
)
@@ -279,6 +315,106 @@ def campaign_profile_delivery_summary(
}
@contextmanager
def campaign_smtp_batch(
session: Session,
*,
tenant_id: str,
campaign_id: str,
profile_id: str,
envelope_from: str,
envelope_recipients: list[str],
from_header: str | None,
expected_smtp_transport_revision: str,
smtp_server_id: str | None = None,
smtp_credential_id: str | None = None,
) -> Iterator[CampaignSmtpBatchState]:
"""Preflight and retain one authorized SMTP connection for a batch."""
selection = _selection_payload(
profile_id=profile_id,
smtp_server_id=smtp_server_id,
smtp_credential_id=smtp_credential_id,
)
try:
profile = _authorized_campaign_profile(
session,
tenant_id=tenant_id,
campaign_id=campaign_id,
profile_id=profile_id,
selection=selection,
)
except MailProfileError:
raise
except Exception:
raise SmtpConfigurationError("The selected Mail profile's SMTP configuration is unusable.") from None
if _supports_hierarchy(session):
context = _campaign_hierarchy_context(
session,
tenant_id=tenant_id,
campaign_id=campaign_id,
)
try:
selected_smtp = select_mail_transport(
session,
profile=profile,
protocol="smtp",
context=context,
server_id=smtp_server_id,
credential_id=smtp_credential_id,
)
except MailServerHierarchyError as exc:
raise MailProfileError(str(exc)) from exc
current_revision = selected_smtp.transport_revision
else:
context = None
current_revision = campaign_profile_transport_revisions(profile)["smtp"]
if current_revision != expected_smtp_transport_revision:
raise MailProfileError(
"The selected Mail profile's SMTP settings changed after this campaign was built. "
"Revalidate and rebuild the campaign before delivery."
)
try:
smtp = (
resolve_mail_transport(
session,
profile=profile,
protocol="smtp",
context=context,
server_id=smtp_server_id,
credential_id=smtp_credential_id,
).config
if context is not None
else smtp_config_from_profile(profile)
)
except MailProfileError:
raise
except Exception:
raise SmtpConfigurationError("The selected Mail profile's SMTP configuration is unusable.") from None
try:
assert_mail_policy_allows_send(
session,
tenant_id=tenant_id,
campaign_id=campaign_id,
smtp=smtp,
imap=None,
envelope_sender=envelope_from,
from_header=from_header,
recipients=envelope_recipients,
)
except MailProfileError:
raise MailProfileError("Mail delivery is blocked by the effective Mail policy.") from None
smtp_session = SmtpBatchSession(smtp)
smtp_session.preflight()
token = _ACTIVE_SMTP_BATCH.set(smtp_session)
try:
yield CampaignSmtpBatchState(session=smtp_session)
finally:
_ACTIVE_SMTP_BATCH.reset(token)
smtp_session.close()
def send_campaign_email_bytes(
session: Session,
*,
@@ -394,6 +530,7 @@ def send_campaign_email_bytes(
smtp_config=smtp,
envelope_from=envelope_from,
envelope_recipients=envelope_recipients,
batch_session=_ACTIVE_SMTP_BATCH.get(),
)
except SmtpSendError as exc:
sanitized = _sanitized_smtp_error(exc)
@@ -420,6 +557,9 @@ def send_campaign_email_bytes(
sanitized_result = CampaignSmtpDeliveryResult(
envelope_recipients=list(result.envelope_recipients),
refused_recipients=_sanitized_refusals(result.refused_recipients),
connection_sequence=getattr(result, "connection_sequence", 0),
session_reused=getattr(result, "session_reused", False),
reconnect_count=getattr(result, "reconnect_count", 0),
)
if recovery is not None:
try:
@@ -604,6 +744,7 @@ class MailCampaignCapability:
assert_campaign_mail_policy_allows_json = staticmethod(assert_campaign_mail_policy_allows_json)
mail_profile_id_from_campaign_json = staticmethod(mail_profile_id_from_campaign_json)
campaign_profile_delivery_summary = staticmethod(campaign_profile_delivery_summary)
campaign_smtp_batch = staticmethod(campaign_smtp_batch)
send_campaign_email_bytes = staticmethod(send_campaign_email_bytes)
append_campaign_message_to_sent = staticmethod(append_campaign_message_to_sent)
wait_for_rate_limit = staticmethod(wait_for_rate_limit)