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 frontendimport 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 SDKfrom 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