fedapay-connector 1.3.4

Un client asynchrone robuste pour l'API FedaPay, offrant une gestion automatisée des paiements avec support complet des webhooks.

FedaPay Connector

Un client asynchrone robuste pour l'API FedaPay, offrant une gestion automatisée des paiements avec support complet des webhooks.

✨ Caractéristiques

• 🔄 Pattern Singleton - Une seule instance partagée dans toute l'application
• ⚡ Entièrement Asynchrone - Performances optimales avec asyncio
• 🔒 Sécurisé - Validation des signatures et gestion sécurisée des webhooks
• 💾 Persistence Automatique - Sauvegarde et restauration des transactions
• 🎯 Callbacks Personnalisables - Hooks pour tous les événements
• 🚀 Simple à Utiliser - API intuitive et documentation complète

Installation

Via pip

pip install fedapay_connector

Via poetry

poetry add fedapay_connector

🛠️ Configuration

Prérequis

• Python 3.9+
• Un compte FedaPay avec les clés API
• Pour le serveur webhook : une URL accessible publiquement

Variables d'Environnement

Variable	Description	Requis	Défaut
FEDAPAY_API_KEY	Clé API FedaPay	✅	-
FEDAPAY_API_URL	URL API (sandbox/production)	✅	-
FEDAPAY_AUTH_KEY	Clé secrète webhook	✅	-
FEDAPAY_ENDPOINT_NAME	Endpoint webhook	❌	webhooks
FEDAPAY_DB_URL	URL sqlalchemy base de données	❌	sqlite:///fedapay_connector_persisted_data/processes.db

Exemple de .env

FEDAPAY_API_KEY=fp_key_live_123456789
FEDAPAY_API_URL=https://api.fedapay.com
FEDAPAY_AUTH_KEY=webhook_secret_123456789
FEDAPAY_ENDPOINT_NAME=webhooks

📚 Guide d'Utilisation

Modes d'Utilisation

1. Mode Simple (non recommandé)

   • Polling manuel du statut
   • Sans gestion des webhooks

2. Mode Serveur Intégré (recommandé)

   • Serveur webhook intégré
   • Gestion automatique des événements
   • Parfait pour une apllication python hors context d'API

3. Mode Serveur Intégré (options avancées) (recommandé)

   • Serveur webhook intégré
   • Gestion automatique des événements
   • Sauvegarde et restauration automatique des processus découtes apres arrêt ou redemarrage de l'app
   • Parfait pour une apllication python hors context d'API

4. Mode API Existante

   • Intégration avec FastAPI/Django/etc
   • Gestion personnalisée des webhooks

1. Mode Simple

from fedapay_connector import Pays, MethodesPaiement, FedapayConnector, PaiementSetup, UserData, EventFutureStatus, PaymentHistory, WebhookHistory
import asyncio

async def main():
    # Creation de l'instance Fedapay Connector
    fedapay = FedapayConnector(use_listen_server= False) 

    # Configuration paiement
    setup = PaiementSetup(
        pays=Pays.benin,
        method=MethodesPaiement.mtn_open
    )
    
    client = UserData(
        nom="Doe",
        prenom="John",
        email="[e-mail]",
        tel="0162626262"
    )

    # Exécution paiement
    resp = await fedapay.fedapay_pay(
        setup=setup,
        client_infos=client,
        montant_paiement=1000,
        payment_contact="0162626262"
    )

    while True:
    # vérifier le resultat manuellement par polling
    status = await fedapay.fedapay_check_transaction_status(resp.id_transaction)
    if status.status == TransactionStatus.created or status.status == TransactionStatus.pending:
        await asyncio.sleep(0.1)
    else:
        break

if __name__ == "__main__":
    asyncio.run(main())

2. Mode Serveur Intégré

Cette option nécéssite que vous ayez un reverse proxy pointant sur votre machine au port d'ecoute configurer pour le serveur (défaut : 3000) depuis une url qui sera utiliseée pour la configuration des webhook sur votre panel fedapay. lien doc fedapay

from fedapay_connector import Pays, MethodesPaiement, FedapayConnector, PaiementSetup, UserData, EventFutureStatus, PaymentHistory, WebhookHistory
import asyncio

async def main():

    # Creation des callbacks
    async def payment_callback(data:PaymentHistory):
        # s'execute chaque fois qu'un nouveau paiement est initialisé avec fedapay_pay()
            print(f"Callback de paiement reçu : {data.__dict__}")

    async def webhook_callback(data:WebhookHistory):
        # s'execute chaque fois qu'un nouveau webhook est reçu de fedapay
        print(f"Webhook reçu : {data.__dict__}")

    # Creation de l'instance Fedapay Connector
    fedapay = FedapayConnector(use_listen_server= True) 

    # Configuration des callbacks
    fedapay.set_payment_callback_function(payment_callback) # executer a chaques appels reussi a fedapay_pay()
    fedapay.set_webhook_callback_function(webhook_callback) # executer à la réception de webhooks fedapay valides

    # Démarrage du listener interne
    fedapay.start_webhook_server()

    # Configuration paiement
    setup = PaiementSetup(
        pays=Pays.benin,
        method=MethodesPaiement.mtn_open
    )
    
    client = UserData(
        nom="Doe",
        prenom="John",
        email="[e-mail]",
        tel="0162626262"
    )

    # Exécution paiement
    resp = await fedapay.fedapay_pay(
        setup=setup,
        client_infos=client,
        montant_paiement=1000,
        payment_contact="0162626262"
    )

    # Attente résultat
    status, webhooks = await fedapay.fedapay_finalise(resp.id_transaction)

    
    if status == EventFutureStatus.RESOLVED:
        print("\nTransaction réussie\n")
        print(f"\nDonnées finales : {webhooks}\n")

        # ATTENTION :  Ce cas indique le reception d'une webhook valide et la clôture de la transaction mais ne veut pas systématiquement dire due l'opération à été approuvée

        # Il faudra implementer par la suite votre gestion des webhook pour la validation ou tout autre traitement du paiement effectuer à partir de la liste d'objet WebhookTransaction reçu.

    elif status == EventFutureStatus.TIMEOUT:
        # La vérification manuelle du statut de la transaction se fait automatiquement si timeout donc si timeout est levé pas besoin de revérifier manuellement le status.

        print("\nLa transaction a expiré.\n")

    elif status == EventFutureStatus.CANCELLED:
        print("\nTransaction annulée par l'utilisateur\n")

    elif future_event_status == EventFutureStatus.CANCELLED_INTERNALLY:
            print("\nTransaction annulée en interne -- probable redemarrage ou arret de l'application\n")
    

if __name__ == "__main__":
    asyncio.run(main())

3. Mode Serveur Intégré (options avancées)

from fedapay_connector import Pays, MethodesPaiement, FedapayConnector, PaiementSetup, UserData, EventFutureStatus, PaymentHistory, WebhookHistory
import asyncio

async def main():

    # Creation des callbacks
    async def payment_callback(data:PaymentHistory):
        # s'execute chaque fois qu'un nouveau paiement est initialisé avec fedapay_pay()
            print(f"Callback de paiement reçu : {data.__dict__}")

    async def webhook_callback(data:WebhookHistory):
        # s'execute chaque fois qu'un nouveau webhook est reçu de fedapay
        print(f"Webhook reçu : {data.__dict__}")
    
    async def run_after_finalise(
        status: EventFutureStatus, data: list[WebhookHistory] | None
    ):
        # s'execute après la récupération d'écoute perdue une fois que la reponse de fedapay est reçue
        # ou que le timeout naturel survient
        if status == EventFutureStatus.RESOLVED:
            print("\nTransaction réussie\n")
            print(f"\nDonnées finales : {data}\n")

            # ATTENTION :  Ce cas indique le reception d'une webhook valide et la clôture de la transaction mais ne veut pas systématiquement dire due l'opération à été approuvée

            # Il faudra implementer par la suite votre gestion des webhook pour la validation ou tout autre traitement du paiement effectuer à partir de la liste d'objet WebhookTransaction reçu.

        elif status == EventFutureStatus.TIMEOUT:
            # La vérification manuelle du statut de la transaction se fait automatiquement si timeout donc si timeout est levé pas besoin de revérifier manuellement le status sur le coup.

            print("\nLa transaction a expiré.\n")

        elif status == EventFutureStatus.CANCELLED:
            print("\nTransaction annulée par l'utilisateur\n")

        elif future_event_status == EventFutureStatus.CANCELLED_INTERNALLY:
            print("\nTransaction annulée en interne -- probable redemarrage ou arret de l'application\n")

    # Creation de l'instance Fedapay Connector
    fedapay = FedapayConnector(use_listen_server= True) 

    # Configuration des callbacks
    fedapay.set_payment_callback_function(payment_callback) # executer a chaques appels reussi a fedapay_pay()
    fedapay.set_webhook_callback_function(webhook_callback) # executer à la réception de webhooks fedapay valides
    fedapay.set_on_persited_listening_processes_loading_finished_callback(run_after_finalise) 
    # éxectuer lors de la récupération des ecoutes d'event fedapay perduent lors d'un potentiel 
    # redemarrage de l'app pendant que des ecoutes sont actives.

    # lancement de la restauration des processus d'écoute
    await fedapay.load_persisted_listening_processes()

    # Démarrage du listener interne
    fedapay.start_webhook_server()

    # Configuration paiement
    setup = PaiementSetup(
        pays=Pays.benin,
        method=MethodesPaiement.mtn_open
    )
    
    client = UserData(
        nom="Doe",
        prenom="John",
        email="[e-mail]",
        tel="0162626262"
    )

    # Exécution paiement
    resp = await fedapay.fedapay_pay(
        setup=setup,
        client_infos=client,
        montant_paiement=1000,
        payment_contact="0162626262"
    )

    # Attente résultat
    status, webhooks = await fedapay.fedapay_finalise(resp.id_transaction)

    await run_after_finalise(status, webhooks)


if __name__ == "__main__":
    asyncio.run(main())

4. Mode API Existante (Intégration FastAPI ou framework similaire)

Dans des cas d'usage comme pour un backend FastAPI vous devrez faire l'initialisation du module dans le lifespan au demarrage de FastAPI puis l'utiliser directement dans vos logiques métiers pour le traitement des transaction.

from fastapi import FastAPI
from contextlib import asynccontextmanager
from fedapay_connector import FedapayConnector


... code du fichier main.py ...

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Creation de l'instance Fedapay Connector
    fedapay = FedapayConnector(use_listen_server= False) 

    # importer ou definissez prealablement les callabacks si voulu
    # Configuration des callbacks
    fedapay.set_payment_callback_function(payment_callback) # executer a chaques appels reussi a fedapay_pay()
    fedapay.set_webhook_callback_function(webhook_callback) # executer à la réception de webhooks fedapay valides
    fedapay.set_on_persited_listening_processes_loading_finished_callback(run_after_finalise) # éxectuer lors de la récupération des ecoutes d'event fedapay perduent lors d'un potentiel redemarrage de l'app pendant que des ecoutes sont actives.

    # lancement de la restauration des processus d'écoute
    await fedapay.load_persisted_listening_processes()

    yield

    #permet un arret propre de fedapay connector
    await fedapay.shutdown_cleanup()

app = FastAPI(lifespan=lifespan)

@app.post("/webhooks/fedapay")
async def fedapay_webhook(request: Request):
    payload = await request.body()
    headers = request.headers
    
    # Validation signature
    signature = headers.get("x-fedapay-signature")
    FedapayConnector().verify_signature(payload, signature)
    
    # Traitement webhook
    event = await request.json()
    await FedapayConnector().fedapay_save_webhook_data(event)
    
    return {"status": "success"}
    
... suite de votre code ...

Si les methodes de paiement que vous souhaiter utilisés ne sont pas disponibles en paiement sans redirection vous devrez recupérer le paiement link et le retourner au front end pour affichage dans une webview ou un element similaire pour finalisation par l'utilisateur. Le satut sera toutefois toujours capturer par le backend directement donc il n'est pas neccessaire de le recupérer coté client.

Fonctionnalités Avancées

Gestion des Webhooks

# 1. Serveur Intégré
fedapay = FedapayConnector(
    use_listen_server=True,
    listen_server_port=3000,
    listen_server_endpoint_name="webhooks"
)
fedapay.start_webhook_server()

# 2. Intégration API Existante
fedapay = FedapayConnector(use_listen_server=False)
await fedapay.fedapay_save_webhook_data(webhook_data)

Callbacks Personnalisés

async def on_payment(payment: PaymentHistory):
    """Appelé après chaque paiement"""
    print(f"Nouveau paiement: {payment.id}")
    
async def on_webhook(webhook: WebhookHistory):
    """Appelé pour chaque webhook"""
    print(f"Webhook reçu: {webhook.name}")

async def run_after_finalise(
    status: EventFutureStatus, data: list[WebhookHistory] | None
):
    """Appelé après la résolution d'écoutes récupérées """
    
    if status == EventFutureStatus.RESOLVED:
        print("\nTransaction réussie\n")
        print(f"\nDonnées finales : {data}\n")

    elif status == EventFutureStatus.TIMEOUT:
        print("\nLa transaction a expiré.\n")

    elif status == EventFutureStatus.CANCELLED:
        print("\nTransaction annulée par l'utilisateur\n")
    
    elif future_event_status == EventFutureStatus.CANCELLED_INTERNALLY:
            print("\nTransaction annulée en interne -- probable redemarrage ou arret de l'application\n")

fedapay.set_payment_callback_function(on_payment)
fedapay.set_webhook_callback_function(on_webhook)
fedapay.set_on_persited_listening_processes_loading_finished_callback(run_after_finalise)

Persistence et Restauration

Le module gère automatiquement :

• Sauvegarde des transactions en cours
• Restauration après redémarrage
• Reprise des écouteurs interrompus
• Synchronisation avec FedaPay

🔧 Dépannage

Problèmes Courants

1. Les webhooks ne sont pas reçus

   • Vérifier l'URL configurée dans FedaPay
   • Vérifier la clé secrète webhook

2. Erreurs de timeout

   • Augmenter la valeur du timeout
   • Vérifier la connexion réseau
   • Consulter les logs pour plus de détails

Contribution

Les contributions sont les bienvenues!

Licence

Ce projet est sous licence GNU Affero General Public License v3.0 or later (AGPL-3.0-or-later). Consultez le fichier LICENSE pour plus d'informations.

🔒 Sécurité

• Ne jamais exposer les clés API
• Toujours valider les signatures webhook
• Utiliser HTTPS en production
• Implémenter des timeouts appropriés