cinetpay-python 0.1.0

SDK Python pour l'API CinetPay v1 — paiements et transferts mobile money en Afrique. Compatible Django, FastAPI, Flask et tout projet Python.

cinetpay-python

SDK Python pour l'API CinetPay v1 — paiements et transferts mobile money en Afrique.

Compatible Django, FastAPI, Flask et tout projet Python 3.10+.

Caractéristiques

• Sync + Async : CinetPayClient et AsyncCinetPayClient
• Multi-pays : credentials api_key / api_password par 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

Installation

pip install cinetpay-python

Environnements

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é.

Démarrage rapide

Synchrone

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="[e-mail]",
        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

Asynchrone

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())

API

Paiement

# 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)

Transfert

from cinetpay import TransferRequest

transfer = client.transfer.create(
    TransferRequest(
        currency="XOF",
        merchant_transaction_id="TR-001",
        phone_number="[phone]",
        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")

Solde

balance = client.balance.get("CI")
print(f"{balance.available_balance} {balance.currency}")

Webhook

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")

Configuration

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(),
)

Token store Redis

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(),
))

Gestion des erreurs

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

Utilitaires

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()

Context Manager

# Sync
with CinetPayClient(config) as client:
    balance = client.balance.get("CI")

# Async
async with AsyncCinetPayClient(config) as client:
    balance = await client.balance.get("CI")

Sécurité

Protection des clés API

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

Credentials masqués

creds = CountryCredentials(api_key="sk_test_abc", api_password="secret")
print(creds)  # CountryCredentials(api_key='***', api_password='***')
print(client)  # CinetPayClient(countries=['CI', 'SN'])

Autres protections

• 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

Support

Pour toute question sur l'API CinetPay : [e-mail]

Licence

MIT