Fameen Messaging — SDK Python officiel

SDK Python officiel de l'API Fameen Messaging : envoyez des SMS, des messages WhatsApp et des emails depuis vos applications Python, authentifiez vos utilisateurs par code de vérification (OTP), suivez leur statut et recevez des webhooks signés.

• Paquet PyPI : fameen-messaging (module fameen_messaging) — version 1.0.4
• Python ≥ 3.9 — dépendance unique : httpx
• Client synchrone (FameenMessaging) et asynchrone (AsyncFameenMessaging)
• Réessais automatiques (réseau, 429, 5xx idempotents), erreurs typées, vérification de webhooks en temps constant

Installation

pip install fameen-messaging

Démarrage rapide

import os
from fameen_messaging import FameenMessaging

client = FameenMessaging(api_key=os.environ["FAMEEN_API_KEY"])  # clé "fam_…", jamais en dur

# SMS
message = client.sms.send("+224620000000", "Bonjour {prenom} !")
print(message.sid, message.status)  # msg_…  queued

# WhatsApp
client.whatsapp.send("+224620000000", "Votre commande est prête ✅")

# Email
client.email.send("client@exemple.com", "Corps du message", subject="Bienvenue !")

# Envoi unifié — canal explicite ou déduit (« @ » dans `to` → email, sinon SMS)
client.messages.create("client@exemple.com", "Bonjour !", subject="Info")
client.messages.create("+224620000000", "Bonjour !", channel="whatsapp")

# Suivi
statut = client.messages.get(message.sid)
page = client.messages.list(channel="sms", status="delivered", page=1, limit=30)
for m in page.data:
    print(m.sid, m.status, m.delivered_at)

# Solde du portefeuille
solde = client.wallet.balance()
print(solde.sms_credits, solde.wa_credits, solde.email_credits, solde.billing.mode)

Le contenu accepte les variables de personnalisation {prenom}, {nom}, {email}, {phone} (max 5 000 caractères ; subject ≤ 255).

WhatsApp — à faire une fois avant le premier envoi

whatsapp.send(...) échoue tant que votre numéro WhatsApp Business n'est pas connecté : il n'existe aucun numéro partagé de repli, Meta imposant que chaque entreprise émette depuis le sien.

1. Tableau de bord → Paramètres → WhatsApp → Connecter WhatsApp. Une fenêtre Meta (Embedded Signup) vous fait choisir ou créer votre compte WhatsApp Business et votre numéro ; la connexion se finalise au retour.
2. Prérequis Meta : un compte Meta Business et un numéro non déjà utilisé sur WhatsApp (ni l'app classique, ni WhatsApp Business), joignable pour recevoir un code.

La fenêtre de 24 h — la règle qui surprend le plus. Meta n'autorise le message libre que dans les 24 h suivant le dernier message reçu de ce contact. En dehors (ou pour un premier contact), seul un gabarit approuvé passe ; un message libre est refusé par Meta et la ressource finit en failed. Faites approuver vos gabarits depuis Paramètres → WhatsApp → Gabarits avant de planifier des envois sortants.

Détail complet : https://fameenbusiness.com/communication/api

Médias (pièces jointes)

WhatsApp et email acceptent des pièces jointes (PDF, images, vidéo, audio). Passez les octets du fichier (bytes) — le SDK les encode en base64 ; l'API héberge le fichier et le distribue. SMS non supporté. Quand un média est fourni, message peut être vide.

from fameen_messaging import FameenMessaging, file_attachment

client = FameenMessaging(api_key="fam_…")

# WhatsApp : un seul média par message, message = légende (facultative)
client.whatsapp.send(
    "+224620000000", "Votre facture",
    media=open("facture.pdf", "rb").read(), file_name="facture.pdf",
)

# Email : plusieurs pièces jointes (file_attachment lit le fichier pour vous)
client.email.send(
    "client@exemple.com", "Bonjour, voir en pièces jointes.",
    subject="Vos documents",
    attachments=[file_attachment("facture.pdf"), file_attachment("cgv.pdf")],
)

Chaque pièce jointe est un dict {"content": bytes|base64, "filename": ..., "content_type": ..., "type": ...} où type vaut image | video | audio | document (déduit du type MIME si absent). Max 16 Mo par fichier. Mêmes paramètres sur le client asynchrone.

Codes de vérification (OTP)

Authentifiez un utilisateur par code à usage unique sur SMS, WhatsApp ou email. Le code est généré, stocké haché et vérifié côté serveur : il ne transite jamais par votre application et n'apparaît dans aucune réponse. Ni génération, ni stockage, ni expiration à gérer.

# 1. Envoyer le code (canal déduit du destinataire si absent)
v = client.otp.send("+224620000000", channel="sms")
# v.verification_id, v.status == "pending", v.expires_at, v.attempts_remaining

# 2. Contrôler le code saisi par l'utilisateur
r = client.otp.verify("483920", verification_id=v.verification_id)
if r.approved:
    ...  # utilisateur authentifié
else:
    # r.reason : "invalid_code" | "expired" | "max_attempts"
    print(f"Échec ({r.reason}), {r.attempts_remaining} tentative(s) restante(s)")

Un code erroné ne lève pas d'exception : la réponse porte status="rejected" et reason. Seules les erreurs de transport ou d'authentification lèvent.

Si vous ne conservez pas l'identifiant, vérifiez par destinataire — la vérification en cours la plus récente est utilisée :

client.otp.verify("483920", to="+224620000000")

Options d'envoi : code_length (4–8), ttl_seconds (60–3600), max_attempts (1–10), template (doit contenir {{code}} ; marqueurs {{code}}, {{minutes}}, {{seconds}}, {{company}}), subject (email), status_callback et idempotency_key. Sans ces paramètres, les réglages du compte s'appliquent.

À savoir :

• L'envoi consomme un crédit du canal utilisé. Toute clé créée depuis le tableau de bord couvre les trois canaux ; channel_not_allowed (403) ne concerne que d'anciennes clés restreintes.
• Un code validé est à usage unique ; le revérifier renvoie rejected.
• Demander un nouveau code pour le même destinataire annule le précédent.
• client.otp.get(verification_id) retourne l'état courant, jamais le code.
• Disponible à l'identique sur le client asynchrone (await client.otp.send(...)).

Client asynchrone

Mêmes méthodes, en async, sur httpx.AsyncClient :

import asyncio
import os
from fameen_messaging import AsyncFameenMessaging

async def main():
    async with AsyncFameenMessaging(api_key=os.environ["FAMEEN_API_KEY"]) as client:
        message = await client.sms.send("+224620000000", "Bonjour !")
        solde = await client.wallet.balance()
        print(message.sid, solde.sms_credits)

asyncio.run(main())

Authenticité du paquet

Depuis la 1.0.2, fameen-messaging est publié par Trusted Publishing : la CI de GitHub Actions s'authentifie auprès de PyPI par échange de jeton OIDC, sans qu'aucun secret n'existe côté dépôt. PyPI génère alors des attestations de provenance (in-toto, signées via Sigstore et inscrites au journal de transparence Rekor), qui lient chaque archive au commit et au workflow qui l'ont produite.

Elles sont consultables depuis la page de la version sur pypi.org, section Provenance : nom du dépôt, du workflow et empreinte de la distribution.

Les versions ≤ 1.0.2 ont été publiées au jeton et n'ont pas d'attestation.

Idempotence

Passez une idempotency_key (en-tête Idempotency-Key, fenêtre de 24 h côté serveur) : tout réessai renvoie la réponse d'origine au lieu de créer un doublon — et cela rend les réessais automatiques du SDK sûrs sur les POST :

client.sms.send("+224620000000", "Commande confirmée", idempotency_key="commande-42-confirmation")

Erreurs

Toutes les erreurs du SDK héritent de FameenError :

from fameen_messaging import FameenAPIError, FameenConnectionError

try:
    client.sms.send("+224620000000", "Bonjour !")
except FameenAPIError as err:
    print(err.status, err.code, err)      # ex. 402 insufficient_credits Crédits insuffisants…
    if err.code == "insufficient_credits":
        ...  # rechargez le portefeuille
    if err.retry_after is not None:
        ...  # 429 : attendez err.retry_after secondes
except FameenConnectionError:
    ...  # l'API n'a pas pu être jointe (DNS, timeout, coupure réseau)

Codes stables (err.code) : invalid_request (400), subscription_expired (400, facturation à la consommation échue), unauthorized (401), insufficient_credits (402), channel_not_allowed (403), not_found (404), rate_limited (429), internal_error (5xx). Si le corps d'erreur est illisible, le code est déduit du statut HTTP et vaut unknown_error par défaut — cette valeur n'est jamais émise par l'API.

Une validation locale est faite avant tout appel réseau (lève TypeError) : to et message non vides ; to contenant « @ » refusé si le canal explicite n'est pas email.

Réessais automatiques

max_retries = 2 par défaut ; backoff exponentiel retry_base × 2^tentative + aléa(0..retry_base) (base 0,5 s).

Limite de débit

60 requêtes/minute par compte — toutes les clés d'un compte partagent ce quota. Chaque réponse expose X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset (epoch secondes) ; le SDK les mémorise :

info = client.last_rate_limit  # RateLimitInfo(limit=60, remaining=42, reset=1770000000) ou None

Webhooks

L'API notifie votre status_callback à chaque changement de statut (event ∈ queued | sent | delivered | failed). Chaque requête est signée : HMAC-SHA256 (hex) du corps brut avec le secret whsec_… du compte, dans l'en-tête X-Fameen-Signature (en-tête informatif : X-Fameen-Event).

⚠️ Vérifiez toujours la signature sur le corps brut (les octets reçus, avant tout parsing JSON). La comparaison est faite en temps constant.

from fameen_messaging import verify_webhook_signature, construct_webhook_event, WebhookVerificationError

ok = verify_webhook_signature(corps_brut, signature, secret)   # -> bool
event = construct_webhook_event(corps_brut, signature, secret)  # -> WebhookEvent (vérifie PUIS parse)

Django

# views.py
import os
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from fameen_messaging import construct_webhook_event, WebhookVerificationError

@csrf_exempt
def fameen_webhook(request):
    try:
        event = construct_webhook_event(
            request.body,  # corps brut : request.body, PAS request.POST
            request.headers.get("X-Fameen-Signature"),
            os.environ["FAMEEN_WEBHOOK_SECRET"],
        )
    except WebhookVerificationError:
        return HttpResponse(status=401)

    if event.event == "delivered":
        ...  # mettez à jour votre base avec event.sid
    return HttpResponse(status=200)

Flask

import os
from flask import Flask, request
from fameen_messaging import construct_webhook_event, WebhookVerificationError

app = Flask(__name__)

@app.post("/webhooks/fameen")
def fameen_webhook():
    try:
        event = construct_webhook_event(
            request.get_data(),  # corps brut : get_data(), PAS request.json
            request.headers.get("X-Fameen-Signature"),
            os.environ["FAMEEN_WEBHOOK_SECRET"],
        )
    except WebhookVerificationError:
        return "", 401

    print(event.sid, event.status)
    return "", 200

FastAPI

import os
from fastapi import FastAPI, Request, Response
from fameen_messaging import construct_webhook_event, WebhookVerificationError

app = FastAPI()

@app.post("/webhooks/fameen")
async def fameen_webhook(request: Request):
    payload = await request.body()  # corps brut, avant tout parsing
    try:
        event = construct_webhook_event(
            payload,
            request.headers.get("x-fameen-signature"),
            os.environ["FAMEEN_WEBHOOK_SECRET"],
        )
    except WebhookVerificationError:
        return Response(status_code=401)

    print(event.sid, event.status)
    return Response(status_code=200)

Configuration

FameenMessaging(
    api_key="fam_…",          # requis
    base_url="https://fameenbusiness.com/api/v1",  # défaut ; « / » finaux retirés
    timeout=30.0,              # timeout httpx par tentative, en secondes
    max_retries=2,             # réessais automatiques
    retry_base=0.5,            # base du backoff exponentiel (s) — utile en test
    transport=None,            # transport httpx injectable (httpx.MockTransport en test)
)

AsyncFameenMessaging accepte exactement les mêmes options.

Modèles de données

Les retours sont des dataclasses gelées, construites de manière tolérante depuis le JSON (champs inconnus ignorés, champs manquants → None/0). Les clés camelCase deviennent snake_case (externalId → external_id) ; la clé from (mot réservé Python) devient from_.

messages.history() est déprécié (lignes brutes, DeprecationWarning) — préférez messages.list().

Tests (développement)

python -m venv .venv
.venv/Scripts/python -m pip install httpx pytest pytest-asyncio   # (Linux/macOS : .venv/bin/python)
.venv/Scripts/python -m pytest -q

Les tests utilisent httpx.MockTransport : aucun appel réseau.

Licence

MIT — © Fameen Groupe.