Send SMS Using Python in India

# send_sms.py: run with  SMS_API_KEY=xxxx python send_sms.py
import os

import requests

resp = requests.post(
    "https://whitelabled.securesmsc.com/api/v2/sms/send",
    json={
        "sender_id": "AVNZON",
        "recipient": "919876543210",              # 91 + 10-digit mobile
        "message": "Your OTP for login is 482913. It is valid for 10 minutes. "
                   "Do not share it with anyone. -AVNZON",
        "route": "transactional",
    },
    headers={
        "Authorization": f"Bearer {os.environ['SMS_API_KEY']}",
        "Accept": "application/json",
    },
    timeout=(5, 15),                              # (connect, read) seconds
)
print(resp.status_code, resp.json())

{
  "success": true,
  "message": "SMS queued successfully",
  "data": {
    "message_id": 1849302,
    "queue_id": 1849302,
    "status": "queued",
    "estimated_delivery": "2026-10-29T10:42:05.000000Z"
  }
}

# sms_client.py
from __future__ import annotations

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

BASE_URL = "https://whitelabled.securesmsc.com/api/v2"


def make_session(api_key: str) -> requests.Session:
    retry = Retry(
        total=None,                 # use the per-category limits below
        connect=3,                  # DNS failure / refused / connect timeout: never sent
        read=False,                 # once the request is on the wire, never resend it
        other=0,
        status=3,                   # 429/502/503: refused before processing, try again
        status_forcelist=(429, 502, 503),   # NOT 500/504: the SMS may already be queued
        allowed_methods=frozenset({"GET", "POST"}),   # POST is NOT retried by default
        backoff_factor=0.5,
        respect_retry_after_header=True,
        raise_on_status=False,      # return the last response instead of raising
    )
    session = requests.Session()
    session.mount("https://", HTTPAdapter(max_retries=retry))
    session.headers.update({
        "Authorization": f"Bearer {api_key}",
        "Accept": "application/json",
    })
    return session

# sms_client.py (continued)
from dataclasses import dataclass
from typing import Any, Optional

from urllib3.exceptions import NewConnectionError


class SmsError(Exception):
    """The gateway refused the request. Retrying will not help."""

    def __init__(self, http_status: int, body: dict):
        self.http_status = http_status
        self.error = body.get("error", "HTTP error")
        self.error_code = body.get("error_code")          # e.g. "TEMPLATE_MISMATCH"
        self.details = body.get("errors") or body.get("details") or {}
        super().__init__(f"HTTP {http_status}: {self.error}: {body.get('message', '')}")


class SmsTemporaryError(SmsError):
    """Not sent, and safe to try again later (429, 502, 503, or no connection)."""


class SmsUncertainError(Exception):
    """The request may have reached the gateway. Check status before resending."""


@dataclass(frozen=True)
class Queued:
    message_id: int
    status: str
    estimated_delivery: Optional[str] = None


def _never_sent(exc: requests.RequestException) -> bool:
    if isinstance(exc, requests.ConnectTimeout):
        return True
    reason = getattr(exc.args[0] if exc.args else None, "reason", None)
    return isinstance(reason, NewConnectionError)    # DNS failure, connection refused


class SmsClient:
    def __init__(self, api_key: str, base_url: str = BASE_URL,
                 timeout: tuple[float, float] = (5, 15)):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout                        # (connect, read) seconds
        self.session = make_session(api_key)

    def send(self, to: str, message: str, sender_id: str, **extra: Any) -> Queued:
        # extra: dlt_template_id="1107...", unicode=True, route="promotional", ...
        payload = {"sender_id": sender_id, "recipient": to, "message": message, **extra}
        data = self._request("POST", "/sms/send", json=payload)["data"]
        return Queued(data["message_id"], data["status"], data.get("estimated_delivery"))

    def send_bulk(self, recipients: list[str], message: str, sender_id: str, **extra: Any) -> dict:
        payload = {"sender_id": sender_id, "recipients": recipients, "message": message, **extra}
        return self._request("POST", "/sms/send-bulk", json=payload)

    def status(self, message_id: int | str) -> str:
        return self._request("GET", f"/sms/status/{message_id}")["data"]["status"]

    def balance(self) -> dict:
        return self._request("GET", "/balance")["data"]

    def _request(self, method: str, path: str, **kwargs: Any) -> dict:
        try:
            resp = self.session.request(method, self.base_url + path,
                                        timeout=self.timeout, **kwargs)
        except requests.RequestException as exc:
            if _never_sent(exc):
                raise SmsTemporaryError(0, {"error": "Connection failed", "message": str(exc)}) from exc
            raise SmsUncertainError(str(exc)) from exc

        try:
            body = resp.json()
        except ValueError:
            body = {"error": f"Non-JSON response ({resp.status_code})"}

        if resp.ok and body.get("success"):
            return body
        if resp.status_code in (429, 502, 503) or (method == "GET" and resp.status_code >= 500):
            raise SmsTemporaryError(resp.status_code, body)
        if resp.status_code >= 500:                   # 500/504 on a send: may be queued
            raise SmsUncertainError(f"HTTP {resp.status_code} from {path}; check before resending")
        raise SmsError(resp.status_code, body)

# templates.py
import re

TPL_APPOINTMENT = "Dear {#alphanumeric#}, your appointment on {#alphanumeric#} is confirmed. -AVNZON"

PLACEHOLDER = re.compile(r"\{#(\w+)#\}")          # {#var#}, {#numeric#}, {#num#}, {#url#} ...
MAX_LEN = {"var": 30, "url": 120, "urlott": 120, "urg": 120, "uro": 120}   # others: 40
DIGITS_ONLY = {"numeric", "num", "number"}


def fill_template(template: str, *values: object) -> str:
    slots = PLACEHOLDER.findall(template)
    if len(values) != len(slots):
        raise ValueError(f"template has {len(slots)} variables, got {len(values)}")

    filled = iter(values)

    def replace(match: re.Match) -> str:
        tag, value = match.group(1), str(next(filled))
        limit = MAX_LEN.get(tag, 40)
        if len(value) > limit:                      # never truncate: a cut URL is broken
            raise ValueError(f"{{#{tag}#}} value is {len(value)} chars; limit is {limit}")
        if tag in DIGITS_ONLY and not value.isdigit():
            raise ValueError(f"{{#{tag}#}} must be digits only, got {value!r}")
        return value

    return PLACEHOLDER.sub(replace, template)


print(fill_template(TPL_APPOINTMENT, "Priya", "31 Oct 4.30 PM"))
# Dear Priya, your appointment on 31 Oct 4.30 PM is confirmed. -AVNZON

# sms_parts.py
GSM7_BASIC = frozenset(
    "@£$¥èéùìòÇ\nØø\rÅåΔ_ΦΓΛΩΠΨΣΘΞÆæßÉ !\"#¤%&'()*+,-./0123456789:;<=>?"
    "¡ABCDEFGHIJKLMNOPQRSTUVWXYZÄÖÑܧ¿abcdefghijklmnopqrstuvwxyzäöñüà"
)
GSM7_EXTENDED = frozenset("^{}\\[~]|€\f")      # each costs 2 characters


def sms_parts(text: str) -> dict:
    if all(ch in GSM7_BASIC or ch in GSM7_EXTENDED for ch in text):
        length = sum(2 if ch in GSM7_EXTENDED else 1 for ch in text)
        encoding, single, per_part = "GSM-7", 160, 153
    else:
        # UCS-2 counts UTF-16 code units; len() counts code points (emoji differ)
        length = len(text.encode("utf-16-le")) // 2
        encoding, single, per_part = "UCS-2", 70, 67
    parts = 1 if length <= single else -(-length // per_part)   # ceiling division
    return {"encoding": encoding, "length": length, "parts": parts}


print(sms_parts("Your OTP for login is 482913. It is valid for 10 minutes. Do not share it with anyone. -AVNZON"))
# {'encoding': 'GSM-7', 'length': 94, 'parts': 1}
print(sms_parts("உங்கள் OTP 482913. 10 நிமிடங்கள் செல்லுபடியாகும். -AVNZON"))
# {'encoding': 'UCS-2', 'length': 57, 'parts': 1}
print(sms_parts("Order shipped 📦 Track: avancezone.com/t/A1023 -AVNZON"))
# {'encoding': 'UCS-2', 'length': 54, 'parts': 1}   (len() says 53: the emoji is 2 units)
print(len("செல்லுபடியாகும்"))   # 15 code points for 7 visible letters

# bulk.py
import logging
from itertools import islice
from typing import Any, Iterable, Iterator

from sms_client import SmsClient, SmsError, SmsUncertainError

log = logging.getLogger("sms.bulk")


def chunks(items: Iterable[str], size: int = 1000) -> Iterator[list[str]]:
    it = iter(items)
    while batch := list(islice(it, size)):     # itertools.batched() on Python 3.12+
        yield batch


def send_to_many(client: SmsClient, numbers: Iterable[str], message: str,
                 sender_id: str = "AVNZON", **extra: Any) -> list[list[str]]:
    unique = list(dict.fromkeys(numbers))      # de-duplicate, keep order
    failed = []
    for batch in chunks(unique):
        try:
            result = client.send_bulk(batch, message, sender_id, **extra)
            log.info("batch of %d accepted: %s", len(batch), result.get("message"))
        except (SmsError, SmsUncertainError) as exc:
            log.error("batch of %d failed: %s", len(batch), exc)
            failed.append(batch)               # investigate, then re-send only these
    return failed

# status.py
import time

FINAL = {"delivered", "failed"}


def wait_for_delivery(client, message_id: int, timeout: float = 120.0) -> str:
    """Poll with growing gaps. For support screens and reconciliation, not every message."""
    deadline = time.monotonic() + timeout
    delay = 2.0
    while True:
        status = client.status(message_id)          # queued | sent | delivered | failed
        if status in FINAL or time.monotonic() + delay > deadline:
            return status
        time.sleep(delay)
        delay = min(delay * 2, 30.0)

# settings.py
import os

SMS_API_KEY = os.environ["SMS_API_KEY"]            # fail at startup, not at first send
SMS_SENDER_ID = os.environ.get("SMS_SENDER_ID", "AVNZON")


# sms/service.py
from functools import lru_cache

from django.conf import settings

from .sms_client import SmsClient


@lru_cache(maxsize=1)
def get_sms_client() -> SmsClient:
    return SmsClient(settings.SMS_API_KEY)         # one Session, reused per process

# sms/tasks.py
from __future__ import annotations

import logging

from celery import shared_task
from django.conf import settings

from .service import get_sms_client
from .sms_client import SmsError, SmsTemporaryError, SmsUncertainError

log = logging.getLogger(__name__)


@shared_task(
    bind=True,
    autoretry_for=(SmsTemporaryError,),   # only "provably not sent" failures
    retry_backoff=30,                     # waits up to 30 s, 60 s, 120 s ... (jittered)
    retry_backoff_max=600,
    max_retries=5,
    acks_late=False,                      # a worker crash must not replay a sent SMS
)
def send_sms_task(self, to: str, message: str, template_id: str | None = None) -> int | None:
    extra = {"dlt_template_id": template_id} if template_id else {}
    try:
        queued = get_sms_client().send(to, message, settings.SMS_SENDER_ID, **extra)
    except SmsUncertainError:
        log.warning("SMS to %s may have been sent; not retrying", to[-4:])
        return None
    except SmsTemporaryError:
        raise                             # autoretry_for handles it
    except SmsError as exc:
        log.error("SMS rejected: %s %s", exc.error_code or exc.error, exc.details)
        return None
    return queued.message_id

# sms/otp.py
import hashlib
import hmac
import os
import secrets

from django.conf import settings
from django.core.cache import cache

from .service import get_sms_client
from .templates import fill_template

TPL_OTP = "Your OTP for login is {#numeric#}. It is valid for 10 minutes. Do not share it with anyone. -AVNZON"
OTP_TTL = 600                                   # seconds
MAX_GUESSES = 5
PEPPER = os.environ["OTP_PEPPER"].encode()      # server-side secret, never stored in the cache/DB


def allow(key: str, limit: int, window: int) -> bool:
    cache.add(key, 0, timeout=window)           # opens the window only if the key is new
    return cache.incr(key) <= limit


def otp_digest(phone: str, code: str) -> str:
    return hmac.new(PEPPER, f"{phone}:{code}".encode(), hashlib.sha256).hexdigest()


def issue_otp(phone: str, ip: str) -> bool:
    # 1. Rate-limit before doing anything else: per number and per client IP.
    if not (allow(f"otp:n:{phone}", 3, 600) and allow(f"otp:ip:{ip}", 20, 3600)):
        return False
    # 2. Six digits from the OS CSPRNG (never the random module); store only an HMAC.
    code = f"{secrets.randbelow(1_000_000):06d}"
    cache.set(f"otp:{phone}", otp_digest(phone, code), timeout=OTP_TTL)
    # 3. Send inline, not via Celery: a person is waiting for this one.
    get_sms_client().send(phone, fill_template(TPL_OTP, code), settings.SMS_SENDER_ID)
    return True


def verify_otp(phone: str, code: str) -> bool:
    if not allow(f"otp:try:{phone}", MAX_GUESSES, OTP_TTL):
        return False                            # too many guesses: wait for the window to pass
    stored = cache.get(f"otp:{phone}")
    if stored is None or not hmac.compare_digest(stored, otp_digest(phone, code)):
        return False
    cache.delete_many([f"otp:{phone}", f"otp:try:{phone}"])   # single use
    return True