Python SDK

nexiom-connect is the official SDK for Python. It works in Django, Flask, FastAPI, Celery workers, or plain scripts, has a sync and an async client, and retries safely for you.

Install#

Python 3.10 or later is required. The SDK uses httpx and anyio.

pip install nexiom-connect

Initialize the client#

Create one client on your server and reuse it: it keeps connections open and is safe to share between threads. It reads the API key from NEXIOM_API_KEY; never ship the key to the browser.

from nexiom_connect import NexiomConnect

nexiom = NexiomConnect()  # Reads NEXIOM_API_KEY.

Pass options by name, for example NexiomConnect(timeout=10). Close the client with nexiom.close() or a with block when your program is done with it.

Use the async client#

AsyncNexiomConnect has the same methods; await them. It runs on asyncio and trio, and takes an httpx.AsyncClient as http_client.

from nexiom_connect import AsyncNexiomConnect


async def send_welcome(address: str) -> str:
    async with AsyncNexiomConnect() as nexiom:
        email = await nexiom.emails.send(
            from_="hello@mail.example.com",
            to=address,
            subject="Welcome",
            text="Thanks for joining us.",
        )

    return email["messageId"]

Send an email#

Arguments are keyword-only and snake_case. from is a Python keyword, so the sender is from_. Methods return the response data as a dict.

email = nexiom.emails.send(
    from_="hello@mail.example.com",
    from_name="Acme",
    to="reader@example.com",
    subject="Your receipt",
    html="<p>Thanks for your order.</p>",
    text="Thanks for your order.",
    idempotency_key=f"receipt-{order.id}",
)

print(email["messageId"])

Pass an idempotency_key that identifies the email in your system. If you omit it, the SDK generates one per call, which protects its own retries but not a retry of your whole operation. See Sending email.

To send a template, pass template_id and template_variables instead of subject and content.

Schedule an email#

Pass scheduled_at as a timezone-aware datetime or an ISO 8601 timestamp with an offset, up to 30 days ahead. A naive datetime is refused, since its moment is ambiguous. Keep the returned messageId to move or cancel the send before it goes out.

from datetime import datetime, timedelta, timezone

now = datetime.now(timezone.utc)

email = nexiom.emails.send(
    from_="hello@mail.example.com",
    to="reader@example.com",
    subject="Your trial ends soon",
    text="Your trial ends soon.",
    scheduled_at=now + timedelta(days=1),
)

# Move it one day later, or cancel it before it is sent.
nexiom.emails.reschedule(email["messageId"], scheduled_at=now + timedelta(days=2))
nexiom.emails.cancel(email["messageId"])

See Reschedule and Cancel for the rules.

Read email logs#

emails.list and emails.get need a full-access key. list pages with a cursor; get takes a delivery ID, not a message ID.

page = nexiom.emails.list(status="bounced", limit=50)

if page["items"]:
    delivery = nexiom.emails.get(page["items"][0]["id"])

    print(delivery["recipient"], delivery["delivery_reason"])

# nextCursor is None on the last page.
if page["nextCursor"] is not None:
    nexiom.emails.list(status="bounced", cursor=page["nextCursor"])

See List emails and Retrieve an email for every field.

Manage contacts#

Define a contact property before you store a value for it.

nexiom.contacts.properties.create(name="plan", type="string")

contact = nexiom.contacts.create(
    email="reader@example.com",
    first_name="Sam",
    properties={"plan": "pro"},
)

nexiom.contacts.update(
    contact["id"],
    email="reader@example.com",
    email_status="unsubscribed",
)

page = nexiom.contacts.list(search="example.com", limit=50)

contacts.get, contacts.delete, and contacts.properties are available too. Responses use snake_case fields such as first_name. On updates, user_id=None clears the external user ID. See the Contacts API for every field.

Send a WhatsApp message#

messaging.send sends on every messaging channel; channel selects the arguments. For WhatsApp, send an approved template from your connected number, named by its phone number in from_. Requires a key with the whatsapp.send permission.

message = nexiom.messaging.send(
    channel="whatsapp",
    from_="+213555000111",
    to="+213555123456",
    template={
        "id": "mtp_example",
        "variables": {"body": {1: "Amina", 2: "A-1234"}},
    },
    idempotency_key="order-A-1234-shipped",
)

current = nexiom.messaging.get(message["id"])

print(current["status"])

See Sending WhatsApp messages.

Verify webhooks#

webhooks.construct_event checks a webhook request's signature and timestamp, and returns the event. Pass the raw body, not parsed JSON. Webhooks() needs no API key. In a Django view:

import os

from django.http import HttpRequest, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from nexiom_connect import NexiomWebhookVerificationError, Webhooks

webhooks = Webhooks()


@csrf_exempt
@require_POST
def nexiom_webhook(request: HttpRequest) -> HttpResponse:
    try:
        event = webhooks.construct_event(
            request.body,
            request.headers,
            os.environ["NEXIOM_WEBHOOK_SECRET"],
        )
    except NexiomWebhookVerificationError as error:
        return HttpResponse(error.reason, status=400)

    if event["type"] == "email.bounced":
        print(event["data"]["recipient"])

    return HttpResponse(status=204)

It raises NexiomWebhookVerificationError when the request is not genuine. In Flask, pass request.get_data() and request.headers; in FastAPI, await request.body() and request.headers. See Verify webhook requests for a complete FastAPI route.

Handle errors#

API and transport failures raise nexiom_connect.NexiomError.

from nexiom_connect import ErrorKind, NexiomError

try:
    email = nexiom.emails.send(**message, idempotency_key=idempotency_key)
except NexiomError as error:
    if error.kind is ErrorKind.API and error.status == 429:
        ...  # Rate limited: try again later with the same idempotency key.

    logger.error("%s %s %s %s", error.status, error.code, error, error.request_id)
    raise

Invalid arguments — for example a missing from_, or an unknown argument such as fromName — raise NexiomValidationError or TypeError before any request is made. Errors can be pickled, so they cross process boundaries in Celery.

Retries and timeouts#

The SDK retries reads, sends and cancels on 408, 429, and 5xx responses and on network failures, honouring Retry-After. Sends are retried with the same idempotency key, so a retry never sends twice. Other writes, such as creating a contact or rescheduling an email, are not retried automatically. If Retry-After asks for a longer wait than the remaining timeout, the SDK raises that error straight away. The timeout covers every attempt and reading the response.

Override per call by passing timeout=... or max_retries=... to any method.

Types#

The package ships type information and passes mypy --strict. Responses are typed with TypedDicts from nexiom_connect.types, such as SendEmailResponse, Contact, Message, and WebhookEvent, so your editor completes field names and your type checker catches typos.