bloonio_mail_relay_client

Backend Python SDK for bloonio_mail_relay — the Bloonio email PaaS. Send transactional/marketing email, manage sending domains + DKIM, view received mail, and receive signed webhook events. Framework-agnostic core + a thin FastAPI adapter.

Mirrors bloonio_auth_relay_client / bloonio_chat_relay_client (HMAC#1 signing, singleton accessor, from_env adapter).

Versioned per SemVer — pre-1.0, so breaking changes may land in minor bumps; they're called out with migration notes in CHANGELOG.md.

Install

pip install "bloonio-mail-relay-client[fastapi]"   # for FastAPI tenants
pip install "bloonio-mail-relay-client[django]"    # for Django tenants
pip install bloonio-mail-relay-client              # framework-agnostic core only

Send an email

from bloonio_mail_relay_client import MailRelayClient, MailRelaySettings

client = MailRelayClient(MailRelaySettings(
    base_url="https://mail-relay.example.com",
    tenant_id="...",          # from provisioning
    tenant_secret="sk_...",   # shown once at provisioning
))

res = client.send(
    from_addr="noreply@acme.com",   # any local-part on a verified domain
    from_name="Acme",               # optional From display name -> "Acme" <noreply@acme.com>
    to=["customer@example.com"],
    subject="Welcome",
    html="<h1>Hi!</h1>",
    text="Hi!",
)
print(res.id, res.status)           # -> "...", "sent"

# schedule for later, or cancel a scheduled send:
from datetime import datetime, timedelta, timezone
later = client.send(from_addr="noreply@acme.com", to="x@example.com", subject="Later",
                    text="...", scheduled_at=datetime.now(timezone.utc) + timedelta(hours=1))
client.cancel_email(later.id)

Send a batch

send_batch takes a list of per-message keyword mappings (each the same shape you'd pass to send) and returns one BatchSendResult per input, in order. By default a failing message is captured (ok=False, with error_code / error_message / status_code) and the batch keeps going; pass stop_on_error=True to re-raise the first MailRelayError instead.

results = client.send_batch([
    {"from_addr": "noreply@acme.com", "to": "a@example.com", "subject": "Hi A", "text": "..."},
    {"from_addr": "noreply@acme.com", "to": "b@example.com", "subject": "Hi B", "text": "..."},
])
for r in results:
    print(r.index, r.ok, r.result.id if r.ok else r.error_code)

The async client accepts a concurrency (default 1 = sequential, matching the sync semantics); with concurrency > 1 up to that many sends run at once and every result is captured (so stop_on_error is not available in that mode):

results = await client.send_batch(messages, concurrency=10)

API keys (for non-SDK / direct-HTTP callers)

key = client.create_api_key("production")
print(key.key)        # bml_… — shown ONCE; store it now (only the prefix is kept)

# a direct caller authenticates with: Authorization: Bearer bml_…
# or build an SDK client that uses the key instead of HMAC signing:
other = MailRelayClient(MailRelaySettings(base_url=..., tenant_id=..., tenant_secret="",
                                          api_key=key.key))

for k in client.iter_api_keys():
    print(k.api_key_id, k.name, k.prefix, k.status)   # metadata only, never the key
client.revoke_api_key(key.api_key_id)                 # takes effect on the next call

Templates ({{var}} substitution)

tpl = client.create_template(
    name="welcome",
    subject="Welcome, {{name}}!",
    html="<p>Hi {{name}}, your code is {{code}}.</p>",   # {{var}} VALUES are HTML-escaped
    text="Hi {{name}}, your code is {{code}}.",
)
print(tpl.variables)   # ['name', 'code']

# send by template — render with vars (no inline subject/body needed):
client.send(from_addr="noreply@acme.com", to="customer@example.com",
            template_id=tpl.template_id, vars={"name": "Sam", "code": "12345"})

for t in client.iter_templates():     # keyset-paginated
    print(t.name, t.variables)
client.update_template(tpl.template_id, subject="Welcome aboard, {{name}}!")
client.delete_template(tpl.template_id)

Your sent-mail log

# Keyset-paginated. Walk one page at a time…
page = client.list_messages(status="failed", limit=50)   # status filter optional
for m in page.items:
    print(m.message_id, m.subject, m.status)
if page.next_cursor:
    page = client.list_messages(cursor=page.next_cursor)

# …or iterate every message, following cursors automatically:
for m in client.iter_messages(status="failed"):
    print(m.message_id, m.status)

msg = client.fetch_message(res.id)                 # full detail (incl. body)
print(msg.body_html)
for r in msg.recipients:                           # per-recipient delivery status
    print(r.address, r.status)                     # delivered | bounced | complained | ...

Domains, suppressions, inbound, webhooks

dom = client.add_domain("acme.com", inbound=True)     # returns DKIM/SPF/DMARC records to publish
client.verify_domain(dom.domain_id)                   # poll until status == "verified"

# Suppression list — addresses you'll never be sent to. Hard bounces and spam
# complaints are added automatically; add your own unsubscribes here too.
client.add_suppression("ex-customer@example.com", reason="unsubscribe")
for s in client.iter_suppressions():                  # keyset-paginated; auto-follows cursors
    print(s.address, s.reason, s.source)              # reason: manual|unsubscribe|complaint|hard_bounce
client.delete_suppression("ex-customer@example.com")  # re-allow after re-confirmation

for msg in client.iter_inbound():                     # the inbound inspector (keyset-paginated)
    full = client.fetch_inbound(msg.inbound_id)
    raw  = client.fetch_inbound_raw(msg.inbound_id)            # bytes (message/rfc822)
    pdf  = client.fetch_inbound_attachment(msg.inbound_id, 0)  # bytes

hook = client.create_webhook(url="https://api.example.com/api/v1/mail-callbacks",
                             events=["email.delivered", "email.bounced", "email.received"])
print(hook.secret)   # save it — verifies inbound webhook signatures (shown once)

for ep in client.iter_webhooks():            # endpoints (keyset-paginated)
    print(ep.endpoint_id, ep.url, ep.health, ep.status)

# the delivery log — debug why a webhook did/didn't fire:
for d in client.iter_webhook_deliveries(status="exhausted"):
    print(d.event_type, d.status, d.attempts, d.last_status)   # e.g. email.bounced exhausted 5 500
full = client.fetch_webhook_delivery(d.delivery_id)
print(full.payload)                          # the exact signed event body that was sent

Analytics

a = client.fetch_analytics(days=30)          # trailing window (1–365; out-of-range is clamped)
print(a.totals["email.sent"], a.totals["email.delivered"], a.totals["email.bounced"])
print(a.rates["delivery_rate"], a.rates["bounce_rate"])   # 0.0–1.0, divide-by-zero safe
for day in a.series:                         # dense: one row per calendar day in the window
    print(day["date"], day["email.sent"], day["email.delivered"])

# open/click metrics appear only when server-side tracking is enabled; until then they're
# absent (not a misleading 0%). Check before reading: a.tracked == {"opens": False, "clicks": False}
if a.tracked.get("opens"):
    print(a.rates["open_rate"])

Receive webhooks (FastAPI)

from fastapi import FastAPI
from bloonio_mail_relay_client import WebhookEventType, WebhookEvent
from bloonio_mail_relay_client.adapters.fastapi import BloonioMailAdapter

app = FastAPI()
_seen: set[str] = set()

async def on_received(ev: WebhookEvent) -> None:
    if ev.event_id in _seen:        # idempotency — see note below
        return
    _seen.add(ev.event_id)
    print("inbound email", ev.data["inbound_id"], ev.data["subject"])

# Reads BLOONIO_MAIL_* env (incl. BLOONIO_MAIL_WEBHOOK_SIGNING_SECRET) and mounts a
# signature-verified POST endpoint at BLOONIO_MAIL_CALLBACK_PATH (default
# /api/v1/mail-callbacks). A bad OR stale signature gets 401; handlers never run on them.
BloonioMailAdapter.from_env(app, handlers={WebhookEventType.EMAIL_RECEIVED: on_received})

Replay defense & idempotency. Each delivery is signed over {timestamp}.{sha256(body)}, and the adapter rejects deliveries whose timestamp is older than BLOONIO_MAIL_WEBHOOK_TOLERANCE_SECONDS (default 300s) — so a sniffed POST can't be replayed forever. Set it to None/large if you verify queue-delayed events. The window is not idempotency: the relay retries on non-2xx, so the same event can legitimately arrive more than once. Always dedupe handlers on ev.event_id (persist it; the snippet's in-memory set is illustrative only). Verifying manually instead of via the adapter? Call verify_webhook(secret, body, ts, sig) from core.hmac — it applies the same window.

Async variant: AsyncMailRelayClient — full method parity with the sync client (await + async with / aclose()), including send_batch(..., concurrency=N).

Receive webhooks (Django)

# urls.py
from django.urls import path
from bloonio_mail_relay_client import WebhookEventType
from bloonio_mail_relay_client.adapters.django import build_callback_view

def on_received(ev):            # sync or async — both work
    print("inbound email", ev.data["inbound_id"])

urlpatterns = [
    path(
        "api/v1/mail-callbacks",
        build_callback_view(handlers={WebhookEventType.EMAIL_RECEIVED: on_received}),
    ),
]

Same config and semantics as the FastAPI adapter: settings=None reads BLOONIO_MAIL_* env vars; a bad OR stale signature gets 401 (handlers never run), an invalid payload 400, and handler exceptions are logged but still return 200 so the relay doesn't retry. The replay-defense + event_id dedupe notes above apply unchanged. Sending needs no adapter — build MailRelayClient(MailRelaySettings()) anywhere (views, Celery tasks, management commands).