Send SMS with Python

pip install httpx

export SENDOZI_API_KEY="sk_test_your_sandbox_key"
export SENDOZI_SENDER_ID="Sendozi"

from __future__ import annotations

import os
import re
from dataclasses import dataclass

import httpx

BASE_URL = "https://api.sendozi.com"
NIGERIAN_MOBILE = re.compile(r"^\+234[789][01]\d{8}$")
_STRIP = re.compile(r"[\s()\-.]")


class SendoziError(RuntimeError):
    """Carries the API error code so callers can branch on it."""

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


def normalise_phone(raw: str) -> str | None:
    """08012345678 and 2348012345678 both become +2348012345678."""
    value = _STRIP.sub("", str(raw).strip())

    if value.startswith("00"):
        value = "+" + value[2:]
    if re.fullmatch(r"0\d{10}", value):
        value = "+234" + value[1:]
    elif re.fullmatch(r"234\d{10}", value):
        value = "+" + value
    elif re.fullmatch(r"[789][01]\d{8}", value):
        value = "+234" + value

    return value if NIGERIAN_MOBILE.fullmatch(value) else None


@dataclass
class Sendozi:
    api_key: str = os.environ.get("SENDOZI_API_KEY", "")
    sender_id: str = os.environ.get("SENDOZI_SENDER_ID", "Sendozi")
    timeout: float = 15.0

    def _post(self, path: str, payload: dict, idempotency_key: str | None = None) -> dict:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        if idempotency_key:
            headers["Idempotency-Key"] = idempotency_key

        response = httpx.post(f"{BASE_URL}{path}", headers=headers, json=payload, timeout=self.timeout)
        envelope = response.json()

        if not envelope.get("success"):
            error = envelope.get("error", {})
            raise SendoziError(
                error.get("code", "request_failed"),
                error.get("message", "The request could not be completed."),
                error.get("resolution", ""),
                envelope.get("request_id", ""),
                response.status_code,
            )
        return envelope["data"]

    def send_sms(self, to: str, message: str, *, sms_type: str = "transactional",
                 sender: str | None = None, idempotency_key: str | None = None) -> dict:
        recipient = normalise_phone(to)
        if recipient is None:
            raise ValueError(f"Not a Nigerian mobile number: {to}")

        return self._post(
            "/v1/sms/send",
            {
                "sender": sender or self.sender_id,
                "recipient": recipient,
                "message": message,
                "sms_type": sms_type,
            },
            idempotency_key,
        )

    def send_bulk_sms(self, to: list[str], message: str, *, sms_type: str = "promotional",
                      sender: str | None = None, idempotency_key: str | None = None) -> dict:
        recipients = [normalise_phone(number) for number in to]
        invalid = [raw for raw, clean in zip(to, recipients) if clean is None]
        if invalid:
            raise ValueError(f"Invalid numbers: {', '.join(invalid)}")

        return self._post(
            "/v1/sms/bulk",
            {
                "sender": sender or self.sender_id,
                "recipients": recipients,
                "message": message,
                "sms_type": sms_type,
            },
            idempotency_key,
        )

from django.http import JsonResponse

from .sendozi import Sendozi, SendoziError

client = Sendozi()


def request_otp(request):
    attempt = create_verification_attempt(request.user)

    try:
        message = client.send_sms(
            to=request.user.phone,
            message=f"Your Acme code is {attempt.code}. It expires in 10 minutes.",
            sms_type="transactional",
            # Derived from the attempt, so a retry replays instead of resending.
            idempotency_key=f"otp-{attempt.id}",
        )
    except SendoziError as error:
        # Log the code and request_id; show the user something human.
        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": message["message_id"], "status": message["status"]})

from celery import shared_task

from .sendozi import Sendozi, SendoziError

PERMANENT = {
    "invalid_request", "invalid_recipient", "message_policy_violation",
    "unauthorized", "forbidden", "kyc_required", "sender_id_not_approved",
    "channel_not_active", "recipient_suppressed", "api_key_blocked",
    "wallet_frozen", "insufficient_balance", "idempotency_conflict",
}


@shared_task(bind=True, max_retries=4, retry_backoff=True, retry_jitter=True)
def send_sms_task(self, phone: str, message: str, event_id: str):
    try:
        # Same idempotency key on every retry: that is what makes this safe.
        return Sendozi().send_sms(phone, message, idempotency_key=f"event-{event_id}")
    except SendoziError as error:
        if error.code in PERMANENT:
            raise  # retrying will fail identically
        raise self.retry(exc=error)

import hashlib
import hmac
import os

from fastapi import FastAPI, Request, Response

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


@app.post("/hooks/sendozi")
async def sendozi_webhook(request: Request):
    raw = await request.body()
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    provided = request.headers.get("x-sendozi-signature", "")

    if not hmac.compare_digest(expected, provided):
        return Response(status_code=401)

    record_delivery.delay(await request.json())  # process out of band
    return Response(status_code=200)

import hashlib
import hmac
import json
import os

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

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


@csrf_exempt
@require_POST
def sendozi_webhook(request):
    # request.body is the raw bytes, which is what was signed.
    expected = hmac.new(SECRET, request.body, hashlib.sha256).hexdigest()
    provided = request.headers.get("X-Sendozi-Signature", "")

    if not hmac.compare_digest(expected, provided):
        return HttpResponse(status=401)

    record_delivery.delay(json.loads(request.body))
    return HttpResponse(status=200)