Sending iMessage from Python

import os
import httpx
from dataclasses import dataclass

BASE_URL = "https://api.sendblue.co/api"

_client = httpx.Client(
    base_url=BASE_URL,
    timeout=httpx.Timeout(10.0, connect=5.0),
    headers={
        "sb-api-key-id": os.environ["IMESSAGE_KEY_ID"],
        "sb-api-secret-key": os.environ["IMESSAGE_SECRET"],
    },
)


class ImessageError(Exception):
    def __init__(self, message: str, status: int, retryable: bool):
        super().__init__(message)
        self.status = status
        self.retryable = retryable


@dataclass(frozen=True)
class SendResult:
    handle: str
    status: str


def send_message(
    to: str,
    body: str,
    status_callback: str | None = None,
) -> SendResult:
    response = _client.post(
        "/send-message",
        json={
            "number": to,
            "content": body,
            "status_callback": status_callback,
        },
    )

    if response.is_error:
        raise ImessageError(
            f"send failed {response.status_code}: {response.text[:300]}",
            response.status_code,
            retryable=response.status_code == 429 or response.status_code >= 500,
        )

    payload = response.json()
    return SendResult(handle=payload["message_handle"], status=payload["status"])

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception,
)


def _is_retryable(error: BaseException) -> bool:
    if isinstance(error, ImessageError):
        return error.retryable
    # Network-level failures are always worth one more go.
    return isinstance(error, httpx.TransportError)


@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential_jitter(initial=0.5, max=8),
    retry=retry_if_exception(_is_retryable),
    reraise=True,
)
def send_with_retry(**kwargs) -> SendResult:
    return send_message(**kwargs)

import hmac
import hashlib
import os
from fastapi import FastAPI, Request, Response, BackgroundTasks

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


def _valid(raw: bytes, signature: str | None) -> bool:
    if not signature:
        return False
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


@app.post("/webhooks/imessage")
async def imessage_webhook(
    request: Request,
    background: BackgroundTasks,
) -> Response:
    raw = await request.body()

    if not _valid(raw, request.headers.get("x-signature")):
        return Response(status_code=401)

    event = await request.json()

    # 200 immediately; the provider retries anything slow, and a retry
    # storm triggered by your own ORM is entirely self-inflicted.
    background.add_task(handle_event, event)
    return Response(status_code=200)


def handle_event(event: dict) -> None:
    event_id = event.get("id")
    if already_processed(event_id):
        return  # At-least-once delivery means this will happen.
    ...

import asyncio
import random

async def send_batch(
    jobs: list[dict],
    per_minute: int = 6,
    concurrency: int = 2,
) -> list[str]:
    spacing = 60 / per_minute
    limit = asyncio.Semaphore(concurrency)
    sent: list[str] = []

    async def one(job: dict, index: int) -> None:
        # Stagger the start, then jitter so the cadence is not machine-regular.
        await asyncio.sleep(index * spacing + random.uniform(0, spacing * 0.3))
        async with limit:
            try:
                result = await asyncio.to_thread(send_with_retry, **job)
                sent.append(result.handle)
            except Exception as error:
                print("send failed", job.get("to"), error)

    await asyncio.gather(*(one(job, i) for i, job in enumerate(jobs)))
    return sent