Mail Subsystem¶

Mail helpers cover the entire pipeline: composing messages with MailBuilder, constraining filesystem access, and delivering payloads via transports such as SMTP. Use this reference to wire the pieces together without spelunking the implementation.

Tip

Pair this reference with Mail for the feature guide.

Quick overview¶

MailBuilder validates addresses, merges HTML/plain bodies, and exposes a fluent API for attachments.
MailFilesystemGuards wraps PathGuardrails so templates, attachments, and inline assets stay within allowed roots.
MailThrottle enforces a token-bucket rate limit before any transport call, acting as a kill switch against accidental mail spam (runaway loops, recursion, hot-path notifications).
MailTransport defines the delivery contract; concrete backends include SMTPTransport, SesTransport, ResendTransport, and GmailTransport.
Exceptions distinguish validation (MailValidationError), configuration (MailConfigurationError), delivery failures (MailTransportError), and throttling rejections (MailThrottledError).

Builder workflow¶

Instantiate a builder, set the addressing headers, choose plain or HTML bodies, and call build() to obtain an EmailMessage. The builder keeps validators in play for every step so you catch issues before hitting the transport layer.

from kstlib.mail import MailBuilder

message = (
    MailBuilder()
    .sender("alerts@example.com")
    .to("ops@example.com")
    .subject("Heartbeat failed")
    .message("Plain fallback", content_type="plain")
    .message("<p>HTML body</p>")
    .build()
)

build() only assembles the MIME payload. Call send() (after assigning a MailTransport) when you want the builder to dispatch via SMTP or other transports.

Templates and placeholders¶

Use message(template=..., placeholders=...) to render HTML or plain content from reusable files. Guardrails ensure template paths live under approved directories, but you still control the OS-level permissions on those folders.

examples/mail/html_template.py¶

"""Render an HTML mail from a reusable template file."""

from __future__ import annotations

from pathlib import Path

from kstlib.mail import MailBuilder

TEMPLATE_PATH = Path(__file__).with_name("templates").joinpath("newsletter.html")


def build_html_message() -> None:
    """Populate the HTML template with placeholders and dump the MIME payload."""
    message = (
        MailBuilder()
        .sender("sender@example.com")
        .to("subscriber@example.com")
        .subject("Monthly Update")
        .message(
            template=TEMPLATE_PATH,
            placeholders={
                "subject": "kstlib Monthly Update",
                "headline": "A brand new mail builder",
                "body_text": "This edition demonstrates HTML templating with placeholders.",
                "signature": "The kstlib Team",
            },
        )
        .build()
    )
    print(message.as_string())


if __name__ == "__main__":  # pragma: no cover - manual example
    build_html_message()

Inline resources require an HTML body and a cid that matches the img src="cid:..." reference. Attachments and inline paths flow through the guardrails so test environments can keep relaxed policies while production stays strict.

examples/mail/attachments_inline.py¶

"""Demonstrate attachments and inline resources with :class:`MailBuilder`."""

from __future__ import annotations

from base64 import b64decode
from pathlib import Path
from tempfile import TemporaryDirectory

from kstlib.mail import MailBuilder, MailFilesystemGuards

_LOGO_BASE64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9Y9dOAoAAAAASUVORK5CYII="


def build_message_with_attachments() -> None:
    """Create a message that includes an attachment and an inline PNG resource."""
    with TemporaryDirectory() as tmp_dir:
        workdir = Path(tmp_dir)
        guards = MailFilesystemGuards.relaxed_for_testing(workdir)

        report_path = guards.attachments_root / "daily-report.txt"
        report_path.parent.mkdir(parents=True, exist_ok=True)
        report_path.write_text("Daily metrics: 42 conversions", encoding="utf-8")

        logo_path = guards.inline_root / "logo.png"
        logo_path.parent.mkdir(parents=True, exist_ok=True)
        logo_path.write_bytes(b64decode(_LOGO_BASE64))

        message = (
            MailBuilder(filesystem=guards)
            .sender("sender@example.com")
            .to("ops@example.com")
            .subject("Daily metrics report")
            .message(
                '<p>Please find the report attached.</p><img src="cid:company-logo" alt="logo" />',
                content_type="html",
            )
            .attach(report_path)
            .attach_inline("company-logo", logo_path)
            .build()
        )

        print(message.as_string())


if __name__ == "__main__":  # pragma: no cover - manual example
    build_message_with_attachments()

Filesystem guardrails¶

MailFilesystemGuards.default() loads configuration from kstlib.conf.yml (if available) and provisions PathGuardrails for attachments, inline assets, and templates. Override roots or policies per builder by instantiating your own guard object:

from pathlib import Path
from kstlib.mail import MailBuilder, MailFilesystemGuards

guards = MailFilesystemGuards.relaxed_for_testing(Path("/tmp/mail"))
builder = MailBuilder(filesystem=guards)

The guardrails never replace OS-level permissions; they simply provide a consistent abstraction to catch obvious mistakes (directory traversal, missing roots) before interacting with the transport layer.

Builder¶

MailBuilder¶

class kstlib.mail.MailBuilder(*, transport=None, preset=None, encoding='utf-8', filesystem=None, limits=None, throttle=None)[source]

Bases: object

Compose and send emails using a fluent interface.

Supports plain text and HTML bodies, file attachments, inline images, and template-based content with placeholder substitution.

Transport resolution cascade, highest priority first:

transport= kwarg (explicit instance, backward compatible)
preset= kwarg (named preset under mail.presets in config)
mail.default in kstlib.conf.yml (auto-resolved preset name)
None: no transport. .build() still works. .send() raises MailConfigurationError.

Preset envelope defaults:

A preset may declare a defaults subsection with sender and reply_to keys. These are applied automatically when the builder is initialised with preset= or when the config’s mail.default resolves to such a preset. User-provided values via .sender() or .reply_to() always override the preset defaults (user wins).

Deliberately scoped to sender and reply_to only: to/cc/bcc are excluded on purpose to prevent silent accidental sends to the preset’s audience. Unsupported keys inside defaults are logged once as a WARNING and ignored (forward compatibility).

Example YAML:

mail:
  presets:
    corporate:
      transport: smtp
      host: smtp-secure.corp.local
      defaults:
        sender: "Service Notifications <notify@corp.local>"
        reply_to: "Service Notifications <notify@corp.local>"

Example

Build an email without sending (useful for inspection):

>>> from kstlib.mail import MailBuilder
>>> mail = (
...     MailBuilder()
...     .sender("noreply@example.com")
...     .to("user@example.com")
...     .subject("Welcome!")
...     .message("<h1>Hello</h1>", content_type="html")
... )
>>> msg = mail.build()
>>> msg["Subject"]
'Welcome!'

With a configured transport for actual delivery (explicit):

>>> from kstlib.mail import MailBuilder
>>> from kstlib.mail.transports import SMTPTransport
>>> transport = SMTPTransport(host="smtp.example.com", port=587)
>>> mail = MailBuilder(transport=transport)
>>> # mail.sender(...).to(...).subject(...).message(...).send()

Config-driven via a named preset (defaults.sender is pre-filled):

>>> mail = MailBuilder(preset="corporate")  
>>> # mail.to(...).subject(...).message(...).send()

Config-driven via mail.default preset:

>>> mail = MailBuilder()  
>>> # Uses preset referenced by mail.default in kstlib.conf.yml

Initialise the builder with optional transport, preset, charset, and guardrails.

Parameters:

transport (MailTransport | None) – Explicit transport instance. Takes priority over preset and the config default. Backward compatible: passing this is equivalent to the pre-preset API. When set, preset envelope defaults are not applied.
preset (str | None) – Name of a preset declared under mail.presets in kstlib.conf.yml. Resolved immediately. Raises MailConfigurationError if the preset does not exist. Any defaults.sender / defaults.reply_to declared on the preset are applied right after transport resolution.
encoding (str) – Character encoding for message bodies (default: utf-8).
filesystem (MailFilesystemGuards | None) – Filesystem guardrails for attachments, inline resources, and templates.
limits (MailLimits | None) – Message and attachment limits.
throttle (bool | dict[str, Any] | None) – Anti-spam kill switch. False disables the throttle on this builder. A dict (e.g. {"rate": 100, "per": 3600.0, "on_exceed": "warn"}) builds a per-instance custom throttle. None (default) resolves the throttle from configuration via the cascade mail.presets.<name>.throttle > mail.throttle > code defaults. See MailThrottle for the available knobs.

Raises:

MailConfigurationError – If preset is passed but does not resolve to a valid preset in configuration, if a preset default has a non-string value, or if the resolved throttle configuration is invalid.
MailValidationError – If a preset default holds an unparseable email address.

transport(self, transport: 'MailTransport') → 'MailBuilder' -> MailBuilder[source]

Attach a transport backend to this builder.

sender(self, value: 'str') → 'MailBuilder' -> MailBuilder[source]

Set the sender address.

reply_to(self, value: 'str | None') → 'MailBuilder' -> MailBuilder[source]

Set an optional reply-to address.

to(self, *values: 'str') → 'MailBuilder' -> MailBuilder[source]

Append recipients to the To header.

cc(self, *values: 'str') → 'MailBuilder' -> MailBuilder[source]

Append recipients to the Cc header.

bcc(self, *values: 'str') → 'MailBuilder' -> MailBuilder[source]

Append recipients to the Bcc header.

subject(self, value: 'str') → 'MailBuilder' -> MailBuilder[source]

Set the message subject.

message(self, content: 'str | None' = None, *, content_type: "Literal['plain', 'html']" = 'html', template: 'str | Path | None' = None, placeholders: 'Mapping[str, Any] | None' = None, **extra_placeholders: 'Any') → 'MailBuilder' -> MailBuilder[source]

Populate the message body either via raw content or a template.

Warning

HTML content is not sanitized. The caller is responsible for ensuring the HTML is safe. This is by design: mail templates are authored by the application developer, not by end users.

Raises:: MailValidationError – If content_type is unsupported.

attach(self, *paths: 'str | Path') → 'MailBuilder' -> MailBuilder[source]

Attach binary files to the message.

Parameters:: *paths (str | Path) – One or more file paths to attach.
Returns:: Self for method chaining.
Raises:: MailValidationError – If no paths provided, attachment limit exceeded, or file size exceeds configured limits.
Return type:: MailBuilder

attach_inline(self, cid: 'str', path: 'str | Path') → 'MailBuilder' -> MailBuilder[source]

Attach an inline resource (e.g. image referenced with cid:).

Raises:: MailValidationError – If cid is empty or file size exceeds limits.

build(self) → 'EmailMessage' -> EmailMessage[source]

Assemble and return an EmailMessage without sending it.

Raises:: MailValidationError – If sender, recipient, or body is missing.

send(self) → 'EmailMessage' -> EmailMessage[source]

Build and send the email using the configured transport.

If a MailThrottle is attached (default, unless disabled via throttle=False), it is consulted before the transport. When the bucket is empty, the throttle either raises MailThrottledError (mode raise) or logs a security warning and returns the built message without sending (mode warn).

Returns:

The constructed EmailMessage. In warn mode, the returned message may have been dropped silently (a WARNING [SECURITY] log is emitted).

Raises:

MailConfigurationError – If no transport has been configured.
MailTransportError – If the transport fails to deliver the message.
MailThrottledError – If the throttle is in raise mode and the bucket is empty.

Return type:

EmailMessage

notify(func: Callable[P, R], /) → Callable[P, R][source]

notify(func: None = None, /, *, subject: str | None = None, on_error_only: bool = False, on_success_only: bool = False, mode: str | None = None, collector: NotifyCollector | None = None, include_return: bool = False, include_traceback: bool = True) → Callable[[Callable[P, R]], Callable[P, R]]

Send email notifications on function execution.

Sends a notification email after the decorated function completes, reporting success or failure with execution metrics. Optionally accumulates results into a NotifyCollector for run summaries.

Can be used with or without parentheses:

@mail.notify
def task(): ...

@mail.notify(subject="Step 1", on_error_only=True)
def task(): ...

@mail.notify(mode="ok")
def check(): ...

Filtering modes (mutually exclusive with on_*_only):

mode="both" (default): notify on both success and failure.
mode="ok" (alias of on_success_only=True): notify only on successful execution.
mode="ko" (alias of on_error_only=True): notify only on exception.

mode is case-insensitive.

When collector is provided, every notification that passes the active filter is also recorded into the collector (so the same gate applies to mail send and to capture, guaranteeing one entry per execution under the double-decorator pattern).

Parameters:

func – The function to decorate (when used without parentheses).
subject – Override the builder’s subject for this notification.
on_error_only – Only send notification if the function raises.
on_success_only – Only send notification on successful return.
mode – Filtering mode ("ok", "ko", or "both", case-insensitive). Mutually exclusive with on_*_only.
collector – Optional collector that records every result that passes the active filter.
include_return – Include return value in success notifications.
include_traceback – Include traceback in failure notifications.

Returns:

Decorated function that sends notifications.

Raises:

MailValidationError – If mode and on_*_only are combined, if on_error_only and on_success_only are both true, or if mode is not in {"ok", "ko", "both"}.

Example

>>> from kstlib.mail import MailBuilder
>>> mail = MailBuilder().sender("bot@x.com").to("admin@x.com")
>>> _ = mail.subject("Daily ETL")
>>> @mail.notify(on_error_only=True)
... def extract():
...     return {"rows": 100}

send_summary(self, collector: 'NotifyCollector', *, subject: 'str | None' = None, format: "Literal['html', 'plain', 'monitor_table']" = 'html') → 'EmailMessage' -> EmailMessage[source]

Send a summary email built from a NotifyCollector.

Operates on an isolated snapshot so the original builder state is preserved for subsequent sends.

Parameters:

collector (NotifyCollector) – The collector whose recorded results are rendered.
subject (str | None) – Optional subject override for the summary email. Falls back to the builder’s current subject when omitted.
format (Literal['html', 'plain', 'monitor_table']) – Output format for the body. "html" uses NotifyCollector.render_html(), "plain" uses NotifyCollector.render_plain(), "monitor_table" renders the collector’s to_monitor_table() with inline CSS for email-safe HTML.

Returns:

The EmailMessage that was sent.

Raises:

MailValidationError – If format is not one of the supported values.

Return type:

EmailMessage

NotifyResult¶

class kstlib.mail.NotifyResult(function_name, success, started_at, ended_at, duration_ms, return_value=None, exception=None, traceback_str=None)[source]

Bases: object

Result of a notified function execution.

function_name

Name of the decorated function.

Type:: str

success

Whether the function completed without exception.

Type:: bool

started_at

UTC timestamp when execution started.

Type:: datetime.datetime

ended_at

UTC timestamp when execution ended.

Type:: datetime.datetime

duration_ms

Execution duration in milliseconds.

Type:: float

return_value

Function return value (if success and include_return=True).

Type:: Any

exception

Exception raised (if failure).

Type:: BaseException | None

traceback_str

Formatted traceback string (if failure and include_traceback=True).

Type:: str | None

function_name: str

success: bool

started_at: datetime

ended_at: datetime

duration_ms: float

return_value: Any

exception: BaseException | None

traceback_str: str | None

__init__(self, function_name: 'str', success: 'bool', started_at: 'datetime', ended_at: 'datetime', duration_ms: 'float', return_value: 'Any' = None, exception: 'BaseException | None' = None, traceback_str: 'str | None' = None) → None -> None

NotifyCollector¶

class kstlib.mail.NotifyCollector(*, maxsize=1000, redact_user_data=True)[source]

Bases: object

Thread-safe bounded collector of NotifyResult instances.

Designed to be passed to kstlib.mail.MailBuilder.notify() via the collector= kwarg, accumulating results across multiple decorated function calls. Provides several rendering helpers for summary emails.

Capacity is bounded to maxsize items with FIFO eviction when full.

Parameters:: maxsize (int) – Maximum number of results to retain. Older entries are evicted FIFO when the bound is reached. Must be positive.
Raises:: ValueError – If maxsize is not a positive integer.

Examples

>>> from kstlib.mail import NotifyCollector
>>> collector = NotifyCollector(maxsize=500)
>>> collector.ok_count
0
>>> collector.ko_count
0

__init__(self, *, maxsize: 'int' = 1000, redact_user_data: 'bool' = True) → 'None' -> None[source]

Initialize the NotifyCollector.

Parameters:

maxsize (int) – Maximum number of results retained (FIFO eviction).
redact_user_data (bool) – When True (default), exception messages, return values, and tracebacks from the decorated user functions are replaced with a redaction placeholder before they appear in the rendered summary. Set to False to opt-in to the legacy verbatim behaviour – only do so if the decorated functions never raise with sensitive content in their messages and never return objects whose repr() would expose secrets.

add(self, result: 'NotifyResult') → 'None' -> None[source]

Append a result to the collector (thread-safe, FIFO bounded).

Parameters:: result (NotifyResult) – The notify result to record.

reset(self) → 'None' -> None[source]

Remove every recorded result (thread-safe).

property maxsize: int: Return the configured maximum capacity.

property results: list[NotifyResult]: Return a snapshot copy of recorded results in insertion order.

property ok_count: int: Return the number of successful results.

property ko_count: int: Return the number of failed results.

property total_count: int: Return the total number of recorded results.

render_html(self, *, include_tracebacks: 'bool' = True) → 'str' -> str[source]

Render results as a standalone HTML table with inline CSS.

Produces a self-contained HTML fragment suitable for embedding in an email body. Each row is colored green for success or red for failure on the Status column. When include_tracebacks is true and any failure carries a traceback, a collapsible <details> block lists them under the table.

Parameters:: include_tracebacks (bool) – Whether to append a collapsible block with the full traceback of each failed result.
Returns:: HTML string with the summary table (and optional traceback block).
Return type:: str

render_plain(self) → 'str' -> str[source]

Render results as a plain text summary.

Returns:: Plain text string with header, per-result rows, and totals.
Return type:: str

to_monitor_table(self) → 'MonitorTable' -> MonitorTable[source]

Build a MonitorTable of the recorded results.

Imports are routed through the public kstlib.monitoring API to avoid cross-feature private imports.

Returns:: A populated MonitorTable with five columns (Function, Status, Started, Duration (ms), Detail).
Return type:: MonitorTable

to_context(self) → 'dict[str, Any]' -> dict[str, Any][source]

Build a Jinja-friendly context dict from the recorded results.

Returns:: results, ok_count, ko_count, total_count, ok_ratio, started_at, ended_at, total_duration_ms.
Return type:: Dict with keys

Filesystem¶

MailFilesystemGuards¶

class kstlib.mail.MailFilesystemGuards(*, attachments, inline=None, templates=None)[source]

Bases: object

Resolve mail templates and attachments using secure guardrails.

Example

>>> guards = MailFilesystemGuards.default()  
>>> safe_attachment = guards.resolve_attachment("reports/daily.csv")  

__init__(self, *, attachments: 'PathGuardrails', inline: 'PathGuardrails | None' = None, templates: 'PathGuardrails | None' = None) → 'None' -> None[source]

Initialise guardrails, defaulting inline/templates to attachment settings.

property attachments_root: Path: Return the root used for attachments (resolved path).

property inline_root: Path: Return the root dedicated to inline resources.

property templates_root: Path: Return the root used for templates (resolved path).

classmethod default() → 'MailFilesystemGuards' -> MailFilesystemGuards[source]

Construct guards from the loaded configuration or fallback defaults.

classmethod from_sources(*, config: 'OptionalMappingSection' = None, roots: 'MailGuardRootsOverrides | None' = None, external: 'MailExternalOverrides | None' = None, policy: 'GuardPolicy | None' = None) → 'MailFilesystemGuards' -> MailFilesystemGuards[source]

Build guards from optional config mappings and overrides.

classmethod relaxed_for_testing(root: 'Path') → 'MailFilesystemGuards' -> MailFilesystemGuards[source]

Create a guards instance with relaxed path policy for tests and examples.

resolve_attachment(self, candidate: 'str | Path') → 'Path' -> Path[source]

Resolve candidate as a secure attachment path.

resolve_inline(self, candidate: 'str | Path') → 'Path' -> Path[source]

Resolve candidate as a secure inline resource path.

resolve_template(self, candidate: 'str | Path') → 'Path' -> Path[source]

Resolve candidate as a secure template file path.

Throttling¶

Mail throttling is a token-bucket rate limit applied before any transport call. It is config-driven via mail.throttle.* (rate, period, mode, key) and resolved through the cascade kwargs > preset > mail.throttle > defaults. The throttle is intended as a defensive kill switch: it stops runaway loops, recursive notification paths, and hot-path @mail.notify decorators from saturating the SMTP backend or external provider quotas.

When the bucket empties, the active mode decides what happens: "raise" raises MailThrottledError, "warn" emits a WARNING [SECURITY] log and lets the call through. Silent drop is intentionally rejected at init: a security event must never be silent.

MailThrottle¶

class kstlib.mail.MailThrottle(*, rate=20, per=60.0, on_exceed='raise')[source]

Bases: object

Token-bucket throttle for mail sends, config-driven.

Singleton per preset (or per builder for explicit transports). Wraps kstlib.resilience.RateLimiter with mail-specific exception type and on_exceed behavior.

The throttle is a kill switch against accidental or buggy mail spam: runaway batch loops, recursion, exception handlers that mail, the @mail.notify decorator on a hot function. It is enforced before the transport is invoked.

Mode "drop" (silent) is intentionally rejected at init: a security event must never be silent. Valid modes are "raise" and "warn", and both emit a WARNING [SECURITY] log per blocked send.

Parameters:

rate (int) – Maximum mails per period. Must be int in [HARD_MIN_THROTTLE_RATE, HARD_MAX_THROTTLE_RATE].
per (float) – Period in seconds. Must be int or float in [HARD_MIN_THROTTLE_PER, HARD_MAX_THROTTLE_PER].
on_exceed (OnExceedMode) – Policy when the bucket is empty. "raise" raises MailThrottledError after the security warning; "warn" logs the warning and returns silently.

Raises:

MailConfigurationError – If a parameter is out of bounds, of a wrong type, or if on_exceed is "drop" (rejected).

Examples

Strict mode (default):

throttle = MailThrottle(rate=20, per=60.0)
if throttle.consume(subject_for_log="Daily report"):
    send_mail()
# MailThrottledError raised on the 21st call within 60s

Warn-only mode (operational critical channels):

throttle = MailThrottle(rate=5, per=60.0, on_exceed="warn")
allowed = throttle.consume("Critical alert")
if not allowed:
    pass  # silently dropped, [SECURITY] logged

Inspect the resolved configuration on a fresh instance:

>>> throttle = MailThrottle(rate=20, per=60.0)
>>> throttle.rate
20
>>> throttle.per
60.0
>>> throttle.on_exceed
'raise'

__init__(self, *, rate: 'int' = 20, per: 'float' = 60.0, on_exceed: 'OnExceedMode' = 'raise') → 'None' -> None[source]

Validate parameters and build the underlying RateLimiter.

property rate: int: Configured rate (mails per period).

property per: float: Configured period in seconds.

property on_exceed: Literal['raise', 'warn']: Configured policy when the bucket is empty.

consume(self, subject_for_log: 'str | None' = None) → 'bool' -> bool[source]

Try to consume one token, applying the on_exceed policy on failure.

Parameters:: subject_for_log (str | None) – Optional subject of the mail being sent. Truncated to 80 chars in the log; if it contains a null byte, replaced by a placeholder. Used for debug only, never for routing.
Returns:: True if the caller should proceed with the send, False if the send should be dropped silently (warn mode only).
Raises:: MailThrottledError – If throttled and on_exceed='raise'.
Return type:: bool

__repr__(self) → 'str' -> str[source]

Return a non-sensitive repr (no builder content, no recipients).

Transports¶

MailTransport¶

class kstlib.mail.MailTransport[source]

Bases: ABC

Abstract sync transport for delivering emails.

Subclass this for synchronous transport implementations like SMTP.

Examples

Implementing a custom sync transport:

class MyTransport(MailTransport):
    def send(self, message: EmailMessage) -> None:
        # Send the message
        pass

abstract send(self, message: 'EmailMessage') → 'None' -> None[source]

Deliver the message to the underlying service.

Parameters:: message (EmailMessage) – The email message to send.
Raises:: MailTransportError – If delivery fails.

AsyncMailTransport¶

class kstlib.mail.AsyncMailTransport[source]

Bases: ABC

Abstract async transport for delivering emails.

Subclass this for asynchronous transport implementations like HTTP APIs.

Examples

Implementing a custom async transport:

class MyAsyncTransport(AsyncMailTransport):
    async def send(self, message: EmailMessage) -> None:
        async with httpx.AsyncClient() as client:
            await client.post(...)

abstract async send(self, message: 'EmailMessage') → 'None' -> None[source]

Deliver the message asynchronously.

Parameters:: message (EmailMessage) – The email message to send.
Raises:: MailTransportError – If delivery fails.

AsyncTransportWrapper¶

class kstlib.mail.AsyncTransportWrapper(transport, *, executor=None)[source]

Bases: AsyncMailTransport

Wrap a sync transport for async usage.

Executes the sync transport’s send method in a thread pool executor to avoid blocking the event loop.

Parameters:

transport (MailTransport) – The sync transport to wrap.
executor (ThreadPoolExecutor | None) – Optional thread pool executor. If None, uses the default.

Examples

Wrapping an SMTP transport:

from kstlib.mail.transport import AsyncTransportWrapper
from kstlib.mail.transports import SMTPTransport

smtp = SMTPTransport(host="smtp.example.com", port=587)
async_smtp = AsyncTransportWrapper(smtp)

# Now usable in async context
await async_smtp.send(message)

With custom executor:

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=2)
async_smtp = AsyncTransportWrapper(smtp, executor=executor)

__init__(self, transport: 'MailTransport', *, executor: 'ThreadPoolExecutor | None' = None) → 'None' -> None[source]

Initialize the async wrapper.

Parameters:

transport (MailTransport) – The sync transport to wrap.
executor (ThreadPoolExecutor | None) – Optional custom thread pool executor.

property transport: MailTransport: Return the wrapped sync transport.

async send(self, message: 'EmailMessage') → 'None' -> None[source]

Send message asynchronously via the wrapped transport.

Runs the sync transport’s send method in a thread pool to avoid blocking the async event loop.

Parameters:: message (EmailMessage) – The email message to send.
Raises:: MailTransportError – If the underlying transport fails.

SMTPTransport¶

class kstlib.mail.transports.smtp.SMTPTransport(host, port=587, *, credentials=None, security=None, timeout=None)[source]

Bases: MailTransport

Deliver messages using the standard SMTP protocol.

__init__(self, host: 'str', port: 'int' = 587, *, credentials: 'SMTPCredentials | None' = None, security: 'SMTPSecurity | None' = None, timeout: 'float | None' = None) → 'None' -> None[source]

Configure connection parameters for the SMTP backend.

send(self, message: 'EmailMessage') → 'None' -> None[source]

Send message through the configured SMTP server.

When TRACE logging is enabled, detailed session information is logged including SMTP commands, SSL/TLS handshake details, and message envelope.

SMTPCredentials¶

class kstlib.mail.transports.smtp.SMTPCredentials(username, password=None)[source]

Bases: object

SMTP authentication bundle.

username: str

password: str | None

__repr__(self) → 'str' -> str[source]

Redact password from repr output.

__init__(self, username: 'str', password: 'str | None' = None) → None -> None

SesTransport¶

class kstlib.mail.transports.ses.SesTransport(*, region='eu-west-3', aws_access_key_id=None, aws_secret_access_key=None, timeout=30.0)[source]

Bases: AsyncMailTransport

Async transport for sending emails via AWS SES.

Uses send_raw_email to pass the full MIME message directly to SES, which preserves all headers, attachments, and HTML content without conversion.

Parameters:

region (str) – AWS region for the SES endpoint (default: eu-west-3).
aws_access_key_id (str | None) – Explicit AWS access key. If omitted, boto3 uses its default credential chain (env vars, instance profile, etc.).
aws_secret_access_key (str | None) – Explicit AWS secret key. Must be provided together with aws_access_key_id.
timeout (float) – Boto3 connect/read timeout in seconds (default: 30.0).

Raises:

MailConfigurationError – If region is empty, timeout is not positive, or only one of the two credential arguments is given.

Examples

Send with EC2 instance profile (no explicit credentials):

transport = SesTransport(region="eu-west-3")
await transport.send(message)

Send with explicit credentials:

transport = SesTransport(
    region="us-east-1",
    aws_access_key_id="AKIA...",
    aws_secret_access_key="secret...",
)
await transport.send(message)

__init__(self, *, region: 'str' = 'eu-west-3', aws_access_key_id: 'str | None' = None, aws_secret_access_key: 'str | None' = None, timeout: 'float' = 30.0) → 'None' -> None[source]

Initialize the SES transport with AWS region, optional credentials, and a send timeout.

__repr__(self) → 'str' -> str[source]

Redact AWS credentials from repr output.

property last_response: SesResponse | None: Return the response from the last successful send.

async send(self, message: 'EmailMessage') → 'None' -> None[source]

Send an email via AWS SES using send_raw_email.

The message is sent as raw MIME bytes, preserving all headers, body parts, and attachments exactly as built by EmailMessage.

Boto3 is synchronous, so the actual API call runs inside run_in_executor to keep the event loop free.

When TRACE logging is enabled, request metadata is logged.

Parameters:

message (EmailMessage) – The email message to send.

Raises:

MailTransportError – If the SES API call fails.
MailConfigurationError – If boto3 is not installed or AWS credentials cannot be resolved.

SesResponse¶

class kstlib.mail.transports.ses.SesResponse(message_id)[source]

Bases: object

Response from AWS SES after sending an email.

message_id

The unique message ID assigned by SES.

Type:: str

message_id: str

__init__(self, message_id: 'str') → None -> None

ResendTransport¶

class kstlib.mail.transports.resend.ResendTransport(api_key, *, base_url='https://api.resend.com/emails', timeout=30.0)[source]

Bases: AsyncMailTransport

Async transport for sending emails via Resend.com API.

Resend provides a simple REST API for sending transactional emails. This transport converts EmailMessage objects to Resend’s JSON format and sends them asynchronously using httpx.

Parameters:

api_key (str) – Resend API key (starts with re_).
base_url (str) – API base URL (default: https://api.resend.com/emails).
timeout (float) – Request timeout in seconds (default: 30.0).

Examples

Basic send:

transport = ResendTransport(api_key="re_123456789")

message = EmailMessage()
message["From"] = "sender@example.com"
message["To"] = "recipient@example.com"
message["Subject"] = "Hello"
message.set_content("Plain text body")

await transport.send(message)

With HTML content:

message = EmailMessage()
message["From"] = "sender@example.com"
message["To"] = "recipient@example.com"
message["Subject"] = "Welcome"
message.set_content("Plain text fallback")
message.add_alternative("<h1>Welcome!</h1>", subtype="html")

await transport.send(message)

__init__(self, api_key: 'str', *, base_url: 'str' = 'https://api.resend.com/emails', timeout: 'float' = 30.0) → 'None' -> None[source]

Initialize the Resend transport.

Parameters:

api_key (str) – Resend API key.
base_url (str) – API endpoint URL.
timeout (float) – Request timeout in seconds.

Raises:

MailConfigurationError – If api_key is empty.

__repr__(self) → 'str' -> str[source]

Redact API key from repr output.

property last_response: ResendResponse | None: Return the response from the last successful send.

async send(self, message: 'EmailMessage') → 'None' -> None[source]

Send an email via the Resend API.

Converts the EmailMessage to Resend’s JSON format and posts it to the API. Supports plain text, HTML, and attachments.

When TRACE logging is enabled, detailed HTTP request/response information is logged including headers and body.

Parameters:

message (EmailMessage) – The email message to send.

Raises:

MailTransportError – If the API request fails.
MailConfigurationError – If required fields are missing.

ResendResponse¶

class kstlib.mail.transports.resend.ResendResponse(id)[source]

Bases: object

Response from Resend API after sending an email.

id

The unique ID assigned to the sent email.

Type:: str

id: str

__init__(self, id: 'str') → None -> None

GmailTransport¶

class kstlib.mail.transports.gmail.GmailTransport(token, *, base_url='https://gmail.googleapis.com/gmail/v1/users/me/messages/send', timeout=30.0)[source]

Bases: AsyncMailTransport

Async transport for sending emails via Gmail API.

Uses OAuth2 Bearer token authentication. The token must have the ‘https://www.googleapis.com/auth/gmail.send’ scope.

Parameters:

token (Token) – OAuth2 token from kstlib.auth module.
base_url (str) – API endpoint URL (default: Gmail send endpoint).
timeout (float) – Request timeout in seconds (default: 30.0).

Raises:

MailConfigurationError – If token is missing or invalid.

Examples

Basic send:

from kstlib.auth import Token
from kstlib.mail.transports import GmailTransport

token = Token(access_token="ya29.xxx")
transport = GmailTransport(token=token)

message = EmailMessage()
message["From"] = "sender@gmail.com"
message["To"] = "recipient@example.com"
message["Subject"] = "Hello"
message.set_content("Email body")

await transport.send(message)

With token refresh callback:

async def refresh_token() -> Token:
    # Refresh logic using kstlib.auth
    return new_token

transport = GmailTransport(
    token=token,
    on_token_refresh=refresh_token,
)

__init__(self, token: 'Token', *, base_url: 'str' = 'https://gmail.googleapis.com/gmail/v1/users/me/messages/send', timeout: 'float' = 30.0) → 'None' -> None[source]

Initialize the Gmail transport.

Parameters:

token (Token) – OAuth2 token with gmail.send scope.
base_url (str) – API endpoint URL.
timeout (float) – Request timeout in seconds.

Raises:

MailConfigurationError – If token is None or has no access_token.

__repr__(self) → 'str' -> str[source]

Redact token from repr output.

property token: Token: Return the current OAuth2 token.

property last_response: GmailResponse | None: Return the response from the last successful send.

update_token(self, token: 'Token') → 'None' -> None[source]

Update the OAuth2 token (e.g., after refresh).

Parameters:: token (Token) – New OAuth2 token.

async send(self, message: 'EmailMessage') → 'None' -> None[source]

Send an email via the Gmail API.

Encodes the EmailMessage in base64url format and posts it to the Gmail API. The sender must match the authenticated user’s email address (or an alias).

When TRACE logging is enabled, detailed HTTP request/response information is logged including headers and body.

Parameters:

message (EmailMessage) – The email message to send.

Raises:

MailTransportError – If the API request fails.
MailConfigurationError – If httpx is not installed.

Exceptions¶

MailError¶

class kstlib.mail.MailError[source]

Bases: KstlibError

Base class for mail related errors.

MailValidationError¶

class kstlib.mail.MailValidationError[source]

Bases: MailError

Raised when provided mail data fails validation checks.

MailConfigurationError¶

class kstlib.mail.MailConfigurationError[source]

Bases: MailError

Raised when the mail builder is missing required configuration.

MailTransportError¶

class kstlib.mail.MailTransportError[source]

Bases: MailError

Raised when a transport backend cannot deliver a message.

MailThrottledError¶

class kstlib.mail.MailThrottledError[source]

Bases: MailError

Raised when the mail throttle bucket is empty and on_exceed is ‘raise’.

Carries the throttle parameters in the message to help the caller decide on a backoff strategy. The throttle is config-driven via mail.throttle.* and acts as an anti-spam kill switch.