Send SMS with Python

pip install httpx

export SENDOZI_API_KEY="sk_test_..."           # sk_live_... in production
export SENDOZI_SENDER="YourBrand"              # approved for promotional
export SENDOZI_TRANSACTIONAL_SENDER="YourBrandOTP"
export SENDOZI_WEBHOOK_SECRET="whsec_..."

from __future__ import annotations

import json
import os
import random
import time
from typing import Iterator

import httpx

BASE_URL = "https://api.sendozi.com/v1"
RETRYABLE = {429, 500, 502, 503, 504}


class SendoziError(RuntimeError):
    """An API refusal. Branch on .code."""

    def __init__(self, status: int, error: dict, request_id: str):
        super().__init__(f"{error.get('code')}: {error.get('message')}")
        self.status = status
        self.code = error.get("code", "unknown")
        self.resolution = error.get("resolution", "")
        self.request_id = request_id


class Sendozi:
    def __init__(self, api_key: str | None = None, timeout: float = 30.0):
        self._http = httpx.Client(
            base_url=BASE_URL,
            headers={"Authorization": f"Bearer {api_key or os.environ['SENDOZI_API_KEY']}"},
            timeout=timeout,
        )

    def _call(self, method: str, path: str, body: dict | None = None, *,
              idempotency_key: str | None = None, params: dict | None = None,
              attempts: int = 4, timeout: float | None = None) -> dict:
        # Serialised once, so every retry sends identical bytes.
        content = json.dumps(body) if body is not None else None
        headers = {"Content-Type": "application/json"} if content else {}
        if idempotency_key:
            headers["Idempotency-Key"] = idempotency_key
        can_retry = method == "GET" or bool(idempotency_key)

        for attempt in range(1, attempts + 1):
            try:
                response = self._http.request(method, path, content=content, headers=headers,
                                              params=params, timeout=timeout or httpx.USE_CLIENT_DEFAULT)
            except httpx.TransportError:
                if not can_retry or attempt == attempts:
                    raise
                time.sleep(2 ** attempt * 0.5)
                continue

            envelope = response.json()
            if envelope.get("success"):
                return envelope["data"]

            error = SendoziError(response.status_code, envelope.get("error", {}), envelope.get("request_id", ""))
            if not can_retry or response.status_code not in RETRYABLE or attempt == attempts:
                raise error
            retry_after = response.headers.get("retry-after")
            time.sleep(int(retry_after) if retry_after and retry_after.isdigit()
                       else 2 ** attempt * 0.5 + random.random() / 4)
        raise RuntimeError("unreachable")

    def send_promotional(self, to: str | list[str], message: str, *, idempotency_key: str,
                         sender: str | None = None) -> dict:
        many = isinstance(to, list)
        body = {"route": "promotional", "sender": sender or os.environ["SENDOZI_SENDER"], "message": message,
                **({"recipients": to} if many else {"recipient": to})}
        return self._call("POST", "/sms/bulk" if many else "/sms/send", body,
                          idempotency_key=idempotency_key,
                          timeout=300.0 if many and len(to) > 10_000 else None)

    def send_transactional(self, to: str, message: str, *, idempotency_key: str,
                           sender: str | None = None) -> dict:
        """One transactional SMS, in your own words. The key keeps a retry from sending twice."""
        body = {
            "route": "transactional",
            "sender": sender or os.environ["SENDOZI_TRANSACTIONAL_SENDER"],
            "recipient": to,
            "message": message,
        }
        return self._call("POST", "/sms/send", body, idempotency_key=idempotency_key)

    def get_message(self, message_id: str) -> dict:
        return self._call("GET", f"/messages/{message_id}")

    def get_batch(self, batch_id: str) -> dict:
        return self._call("GET", f"/message-batches/{batch_id}")

    def cancel_batch(self, batch_id: str) -> dict:
        return self._call("POST", f"/message-batches/{batch_id}/cancel")

    def batch_recipients(self, batch_id: str) -> Iterator[dict]:
        cursor = None
        while True:
            params = {"limit": 500, **({"cursor": cursor} if cursor else {})}
            page = self._call("GET", f"/message-batches/{batch_id}/recipients", params=params)
            yield from page["data"]
            if not page["has_more"]:
                return
            cursor = page["next_cursor"]

import logging
import os

from django.http import JsonResponse

from .sendozi import Sendozi, SendoziError

client = Sendozi()
logger = logging.getLogger(__name__)


def request_otp(request):
    attempt = create_verification_attempt(request.user)
    try:
        result = client.send_transactional(
            to=request.user.phone,
            message=f"Your Acme code is {attempt.code}. It expires in 10 minutes.",
            idempotency_key=f"login-{attempt.id}",
        )
    except SendoziError as error:
        logger.warning("otp send failed", extra={"code": error.code, "request_id": error.request_id})
        return JsonResponse({"error": "We could not send the code. Try again shortly."}, status=502)
    return JsonResponse({"message_id": result.get("message_id")})

from celery import shared_task

from .sendozi import Sendozi


@shared_task(bind=True, max_retries=3, retry_backoff=True, retry_jitter=True)
def send_campaign(self, campaign_id: str, numbers: list[str], message: str):
    # The client already retries transient failures with the same key; a
    # Celery retry reuses the same key too, so it cannot queue the send twice.
    batch = Sendozi().send_promotional(numbers, message, idempotency_key=f"campaign-{campaign_id}")
    return {"batch_id": batch["batch_id"], "accepted": batch["accepted_count"]}

client = Sendozi()
batch = client.get_batch("bat_5f45b8c07bd844c8")
print(batch["status"], batch["progress"])

for recipient in client.batch_recipients(batch["batch_id"]):
    if recipient["status"] in ("failed", "skipped"):
        print(recipient["recipient"], recipient["error"]["code"])

import hashlib
import hmac
import os
import time

SECRET = os.environ["SENDOZI_WEBHOOK_SECRET"].encode()


def verify_sendozi(raw: bytes, timestamp: str | None, signature: str | None) -> bool:
    if not timestamp or not signature or not timestamp.isdigit():
        return False
    if abs(time.time() - int(timestamp)) > 300:
        return False
    expected = hmac.new(SECRET, timestamp.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

import json

from fastapi import FastAPI, Request, Response

from .verify import verify_sendozi

app = FastAPI()


@app.post("/webhooks/sendozi")
async def sendozi_webhook(request: Request):
    raw = await request.body()
    if not verify_sendozi(raw, request.headers.get("x-sendozi-timestamp"),
                          request.headers.get("x-sendozi-signature")):
        return Response(status_code=401)
    record_event.delay(json.loads(raw))  # dedupe on event["id"] in the task
    return Response(status_code=200)

import json

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

from .verify import verify_sendozi


@csrf_exempt
@require_POST
def sendozi_webhook(request):
    # request.body is the raw bytes that were signed.
    if not verify_sendozi(request.body, request.headers.get("X-Sendozi-Timestamp"),
                          request.headers.get("X-Sendozi-Signature")):
        return HttpResponse(status=401)
    record_event.delay(json.loads(request.body))
    return HttpResponse(status=200)