SDK Python pour l'API CinetPay v1 — paiements et transferts mobile money en Afrique.
Compatible Django, FastAPI, Flask et tout projet Python 3.10+.
- Sync + Async :
CinetPayClientetAsyncCinetPayClient - Multi-pays : credentials
api_key/api_passwordpar pays - Auto-détection : sandbox (
sk_test_) vs production (sk_live_) - Token cache : JWT mis en cache 23h, thread-safe (stampede guard)
- Validation : données validées avant envoi (montants, emails, URLs)
- Webhook : vérification timing-safe (
hmac.compare_digest) - Typé : type hints complets,
py.typed(PEP 561), compatible mypy - Sécurisé : HTTPS obligatoire, credentials masqués dans
repr(), SSRF protection
pip install cinetpay-python
| Préfixe clé API | URL API | Environnement |
|---|---|---|
sk_test_... |
https://api.cinetpay.net |
Sandbox |
sk_live_... |
https://api.cinetpay.co |
Production |
Le SDK détecte automatiquement l'environnement à partir du préfixe de la clé.
from cinetpay import CinetPayClient, ClientConfig, CountryCredentials, PaymentRequest import os client = CinetPayClient(ClientConfig( credentials={ "CI": CountryCredentials( api_key=os.environ["CINETPAY_API_KEY_CI"], api_password=os.environ["CINETPAY_API_PASSWORD_CI"], ), }, debug=True, )) # Initialiser un paiement payment = client.payment.initialize( PaymentRequest( currency="XOF", merchant_transaction_id="ORDER-001", amount=5000, lang="fr", designation="Achat en ligne", client_email="client@email.com", client_first_name="Jean", client_last_name="Dupont", success_url="https://monsite.com/success", failed_url="https://monsite.com/failed", notify_url="https://monsite.com/webhook", channel="PUSH", ), "CI", ) print(payment.payment_url) # Rediriger le client print(payment.payment_token) # Pour le Seamless frontend
import asyncio from cinetpay import AsyncCinetPayClient, ClientConfig, CountryCredentials async def main(): async with AsyncCinetPayClient(ClientConfig( credentials={ "CI": CountryCredentials( api_key="sk_test_...", api_password="your_password", ), }, )) as client: balance = await client.balance.get("CI") print(f"Solde: {balance.available_balance} {balance.currency}") asyncio.run(main())
# Initialiser payment = client.payment.initialize(PaymentRequest(...), "CI") print(payment.payment_url) print(payment.payment_token) # Vérifier le statut status = client.payment.get_status("ORDER-001", "CI") print(status.status) # SUCCESS, FAILED, PENDING, ... print(status.user.name)
from cinetpay import TransferRequest transfer = client.transfer.create( TransferRequest( currency="XOF", merchant_transaction_id="TR-001", phone_number="+2250707000001", amount=500, payment_method="OM_CI", reason="Remboursement", notify_url="https://monsite.com/webhook", ), "CI", ) print(transfer.status) # Vérifier le statut status = client.transfer.get_status(transfer.transaction_id, "CI")
balance = client.balance.get("CI") print(f"{balance.available_balance} {balance.currency}")
from cinetpay import verify_notification, parse_notification # Flask @app.route("/webhook", methods=["POST"]) def webhook(): payload = parse_notification(request.json) # Vérifier le token (timing-safe) expected = get_stored_notify_token(payload.merchant_transaction_id) if not verify_notification(expected, payload.notify_token): return "Invalid token", 401 # Confirmer le statut status = client.payment.get_status(payload.transaction_id, "CI") if status.status == "SUCCESS": # Livrer la commande pass return "OK", 200
# FastAPI @app.post("/webhook") async def webhook(request: Request): body = await request.json() payload = parse_notification(body) if not verify_notification(stored_token, payload.notify_token): raise HTTPException(401, "Invalid token") status = await client.payment.get_status(payload.transaction_id, "CI") return {"status": status.status}
# Django def webhook(request): import json payload = parse_notification(json.loads(request.body)) if not verify_notification(stored_token, payload.notify_token): return HttpResponse(status=401) status = client.payment.get_status(payload.transaction_id, "CI") return HttpResponse("OK")
from cinetpay import ClientConfig, CountryCredentials config = ClientConfig( # Credentials par pays (obligatoire) credentials={ "CI": CountryCredentials(api_key="sk_test_...", api_password="..."), "SN": CountryCredentials(api_key="sk_test_...", api_password="..."), }, # URL de base (auto-détecté depuis le préfixe de la clé) # base_url="https://api.cinetpay.co", # forcer la production # TTL du cache token en secondes (défaut: 82800 = 23h) token_ttl=82800, # Timeout des requêtes en secondes (défaut: 30.0) timeout=30.0, # Active les logs (défaut: False) debug=True, # Token store personnalisé (défaut: MemoryTokenStore) # token_store=RedisTokenStore(), )
import redis from cinetpay import ClientConfig, CountryCredentials class RedisTokenStore: def __init__(self): self.r = redis.Redis() def get(self, key: str) -> str | None: val = self.r.get(key) return val.decode() if val else None def set(self, key: str, value: str, ttl_seconds: int) -> None: self.r.setex(key, ttl_seconds, value) def delete(self, key: str) -> None: self.r.delete(key) client = CinetPayClient(ClientConfig( credentials={"CI": CountryCredentials(...)}, token_store=RedisTokenStore(), ))
from cinetpay import ( CinetPayError, ApiError, AuthenticationError, NetworkError, ValidationError, ) try: payment = client.payment.initialize(request, "CI") except ValidationError as e: # Données invalides — avant tout appel réseau print(e) # [amount] must be an integer between 100 and 2500000 except ApiError as e: # Erreur API CinetPay print(e.api_code) # 1200 print(e.api_status) # TRANSACTION_EXIST print(e.description) # La transaction existe déjà except AuthenticationError: # Credentials invalides except NetworkError as e: # Problème réseau print(e.cause) except CinetPayError: # Catch-all pour toutes les erreurs du SDK
from cinetpay import is_final_status, PAYMENT_METHODS_BY_COUNTRY, COUNTRY_CODES # Vérifier si un statut est final is_final_status("SUCCESS") # True is_final_status("PENDING") # False # Opérateurs par pays PAYMENT_METHODS_BY_COUNTRY["CI"] # ("OM_CI", "MOOV_CI", "MTN_CI", "WAVE_CI") # Pays supportés COUNTRY_CODES # ("CI", "BF", "ML", "SN", "TG", "GN", "CM", "BJ", "CD", "NE") # Révoquer un token client.revoke_token("CI") client.revoke_all_tokens()
# Sync with CinetPayClient(config) as client: balance = client.balance.get("CI") # Async async with AsyncCinetPayClient(config) as client: balance = await client.balance.get("CI")
NE FAITES PAS FAITES
────────────────────────────────────────────────────────────────────────
api_key="clé-en-dur" api_key=os.environ["CINETPAY_API_KEY_CI"]
Mélanger sk_test_ et sk_live_ Utiliser le même env pour tous les pays
Commiter le .env dans git Ajouter .env dans .gitignore
print(credentials) Le repr() masque automatiquement les clés
creds = CountryCredentials(api_key="sk_test_abc", api_password="secret") print(creds) # CountryCredentials(api_key='***', api_password='***') print(client) # CinetPayClient(countries=['CI', 'SN'])
- HTTPS obligatoire (sauf localhost)
- SSRF : warning si le hostname n'est pas un domaine CinetPay connu
- Erreurs sanitisées : les messages d'erreur d'authentification ne contiennent jamais les credentials
- Token stampede guard :
threading.Lock(sync) /asyncio.Lock(async) empêche les appels auth simultanés
Pour toute question sur l'API CinetPay : support@cinetpay.com
MIT