# 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
```bash
pip install fedapay_connector
```
### Via poetry
```bash
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
```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
```python
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="john@example.com",
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](https://docs.fedapay.com/integration-api/fr/webhooks-fr)
```python
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="john@example.com",
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")
if __name__ == "__main__":
asyncio.run(main())
```
### 3. Mode Serveur Intégré (options avancées)
```python
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")
# 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="john@example.com",
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.
```python
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
```python
# 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
```python
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")
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
Raw data
{
"_id": null,
"home_page": null,
"name": "fedapay-connector",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": "ASSOGBA Dayane <assogbadayane197@gmail.com>",
"keywords": "API, asynchrone, connector, fedapay, paiement",
"author": null,
"author_email": "ASSOGBA Dayane <assogbadayane197@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/df/c1/ac41b0e5de1f6b3351b6e703e615145c7aa4f88838bfab91b35418718df8/fedapay_connector-1.3.2.tar.gz",
"platform": null,
"description": "# FedaPay Connector\n\n\n\n\n\n\nUn client asynchrone robuste pour l'API FedaPay, offrant une gestion automatis\u00e9e des paiements avec support complet des webhooks.\n\n## \u2728 Caract\u00e9ristiques\n\n- \ud83d\udd04 **Pattern Singleton** - Une seule instance partag\u00e9e dans toute l'application\n- \u26a1 **Enti\u00e8rement Asynchrone** - Performances optimales avec asyncio\n- \ud83d\udd12 **S\u00e9curis\u00e9** - Validation des signatures et gestion s\u00e9curis\u00e9e des webhooks\n- \ud83d\udcbe **Persistence Automatique** - Sauvegarde et restauration des transactions\n- \ud83c\udfaf **Callbacks Personnalisables** - Hooks pour tous les \u00e9v\u00e9nements\n- \ud83d\ude80 **Simple \u00e0 Utiliser** - API intuitive et documentation compl\u00e8te\n\n## Installation\n\n### Via pip\n```bash\npip install fedapay_connector\n```\n\n### Via poetry\n```bash\npoetry add fedapay_connector\n```\n\n## \ud83d\udee0\ufe0f Configuration\n\n### Pr\u00e9requis\n\n- Python 3.9+\n- Un compte FedaPay avec les cl\u00e9s API\n- Pour le serveur webhook : une URL accessible publiquement\n\n### Variables d'Environnement\n\n| Variable | Description | Requis | D\u00e9faut |\n|----------|-------------|--------|---------|\n| `FEDAPAY_API_KEY` | Cl\u00e9 API FedaPay | \u2705 | - |\n| `FEDAPAY_API_URL` | URL API (sandbox/production) | \u2705 | - |\n| `FEDAPAY_AUTH_KEY` | Cl\u00e9 secr\u00e8te webhook | \u2705 | - |\n| `FEDAPAY_ENDPOINT_NAME` | Endpoint webhook | \u274c | `webhooks` |\n| `FEDAPAY_DB_URL` | URL sqlalchemy base de donn\u00e9es | \u274c | `sqlite:///fedapay_connector_persisted_data/processes.db` |\n\n### Exemple de .env\n```env\nFEDAPAY_API_KEY=fp_key_live_123456789\nFEDAPAY_API_URL=https://api.fedapay.com\nFEDAPAY_AUTH_KEY=webhook_secret_123456789\nFEDAPAY_ENDPOINT_NAME=webhooks\n```\n\n## \ud83d\udcda Guide d'Utilisation\n\n### Modes d'Utilisation\n\n1. **Mode Simple** (non recommand\u00e9)\n - Polling manuel du statut\n - Sans gestion des webhooks\n\n2. **Mode Serveur Int\u00e9gr\u00e9** (recommand\u00e9)\n - Serveur webhook int\u00e9gr\u00e9\n - Gestion automatique des \u00e9v\u00e9nements\n - Parfait pour une apllication python hors context d'API\n\n3. **Mode Serveur Int\u00e9gr\u00e9 (options avanc\u00e9es)** (recommand\u00e9)\n - Serveur webhook int\u00e9gr\u00e9\n - Gestion automatique des \u00e9v\u00e9nements\n - Sauvegarde et restauration automatique des processus d\u00e9coutes apres arr\u00eat ou redemarrage de l'app\n - Parfait pour une apllication python hors context d'API\n\n4. **Mode API Existante** \n - Int\u00e9gration avec FastAPI/Django/etc\n - Gestion personnalis\u00e9e des webhooks\n\n### 1. Mode Simple \n\n```python\nfrom fedapay_connector import Pays, MethodesPaiement, FedapayConnector, PaiementSetup, UserData, EventFutureStatus, PaymentHistory, WebhookHistory\nimport asyncio\n\nasync def main():\n # Creation de l'instance Fedapay Connector\n fedapay = FedapayConnector(use_listen_server= False) \n\n # Configuration paiement\n setup = PaiementSetup(\n pays=Pays.benin,\n method=MethodesPaiement.mtn_open\n )\n \n client = UserData(\n nom=\"Doe\",\n prenom=\"John\",\n email=\"john@example.com\",\n tel=\"0162626262\"\n )\n\n # Ex\u00e9cution paiement\n resp = await fedapay.fedapay_pay(\n setup=setup,\n client_infos=client,\n montant_paiement=1000,\n payment_contact=\"0162626262\"\n )\n\n while True:\n # v\u00e9rifier le resultat manuellement par polling\n status = await fedapay.fedapay_check_transaction_status(resp.id_transaction)\n if status.status == TransactionStatus.created or status.status == TransactionStatus.pending:\n await asyncio.sleep(0.1)\n else:\n break\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n### 2. Mode Serveur Int\u00e9gr\u00e9\n\nCette option n\u00e9c\u00e9ssite que vous ayez un reverse proxy pointant sur votre machine au port d'ecoute configurer pour le serveur\n(d\u00e9faut : 3000) depuis une url qui sera utilise\u00e9e pour la configuration des webhook sur votre panel fedapay. \n[lien doc fedapay](https://docs.fedapay.com/integration-api/fr/webhooks-fr)\n\n```python\nfrom fedapay_connector import Pays, MethodesPaiement, FedapayConnector, PaiementSetup, UserData, EventFutureStatus, PaymentHistory, WebhookHistory\nimport asyncio\n\nasync def main():\n\n # Creation des callbacks\n async def payment_callback(data:PaymentHistory):\n # s'execute chaque fois qu'un nouveau paiement est initialis\u00e9 avec fedapay_pay()\n print(f\"Callback de paiement re\u00e7u : {data.__dict__}\")\n\n async def webhook_callback(data:WebhookHistory):\n # s'execute chaque fois qu'un nouveau webhook est re\u00e7u de fedapay\n print(f\"Webhook re\u00e7u : {data.__dict__}\")\n\n # Creation de l'instance Fedapay Connector\n fedapay = FedapayConnector(use_listen_server= True) \n\n # Configuration des callbacks\n fedapay.set_payment_callback_function(payment_callback) # executer a chaques appels reussi a fedapay_pay()\n fedapay.set_webhook_callback_function(webhook_callback) # executer \u00e0 la r\u00e9ception de webhooks fedapay valides\n\n # D\u00e9marrage du listener interne\n fedapay.start_webhook_server()\n\n # Configuration paiement\n setup = PaiementSetup(\n pays=Pays.benin,\n method=MethodesPaiement.mtn_open\n )\n \n client = UserData(\n nom=\"Doe\",\n prenom=\"John\",\n email=\"john@example.com\",\n tel=\"0162626262\"\n )\n\n # Ex\u00e9cution paiement\n resp = await fedapay.fedapay_pay(\n setup=setup,\n client_infos=client,\n montant_paiement=1000,\n payment_contact=\"0162626262\"\n )\n\n # Attente r\u00e9sultat\n status, webhooks = await fedapay.fedapay_finalise(resp.id_transaction)\n\n \n if status == EventFutureStatus.RESOLVED:\n print(\"\\nTransaction r\u00e9ussie\\n\")\n print(f\"\\nDonn\u00e9es finales : {webhooks}\\n\")\n\n # ATTENTION : Ce cas indique le reception d'une webhook valide et la cl\u00f4ture de la transaction mais ne veut pas syst\u00e9matiquement dire due l'op\u00e9ration \u00e0 \u00e9t\u00e9 approuv\u00e9e\n\n # Il faudra implementer par la suite votre gestion des webhook pour la validation ou tout autre traitement du paiement effectuer \u00e0 partir de la liste d'objet WebhookTransaction re\u00e7u.\n\n elif status == EventFutureStatus.TIMEOUT:\n # La v\u00e9rification manuelle du statut de la transaction se fait automatiquement si timeout donc si timeout est lev\u00e9 pas besoin de rev\u00e9rifier manuellement le status.\n\n print(\"\\nLa transaction a expir\u00e9.\\n\")\n\n elif status == EventFutureStatus.CANCELLED:\n print(\"\\nTransaction annul\u00e9e par l'utilisateur\\n\")\n \n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n\n### 3. Mode Serveur Int\u00e9gr\u00e9 (options avanc\u00e9es)\n\n```python\nfrom fedapay_connector import Pays, MethodesPaiement, FedapayConnector, PaiementSetup, UserData, EventFutureStatus, PaymentHistory, WebhookHistory\nimport asyncio\n\nasync def main():\n\n # Creation des callbacks\n async def payment_callback(data:PaymentHistory):\n # s'execute chaque fois qu'un nouveau paiement est initialis\u00e9 avec fedapay_pay()\n print(f\"Callback de paiement re\u00e7u : {data.__dict__}\")\n\n async def webhook_callback(data:WebhookHistory):\n # s'execute chaque fois qu'un nouveau webhook est re\u00e7u de fedapay\n print(f\"Webhook re\u00e7u : {data.__dict__}\")\n \n async def run_after_finalise(\n status: EventFutureStatus, data: list[WebhookHistory] | None\n ):\n # s'execute apr\u00e8s la r\u00e9cup\u00e9ration d'\u00e9coute perdue une fois que la reponse de fedapay est re\u00e7ue\n # ou que le timeout naturel survient\n if status == EventFutureStatus.RESOLVED:\n print(\"\\nTransaction r\u00e9ussie\\n\")\n print(f\"\\nDonn\u00e9es finales : {data}\\n\")\n\n # ATTENTION : Ce cas indique le reception d'une webhook valide et la cl\u00f4ture de la transaction mais ne veut pas syst\u00e9matiquement dire due l'op\u00e9ration \u00e0 \u00e9t\u00e9 approuv\u00e9e\n\n # Il faudra implementer par la suite votre gestion des webhook pour la validation ou tout autre traitement du paiement effectuer \u00e0 partir de la liste d'objet WebhookTransaction re\u00e7u.\n\n elif status == EventFutureStatus.TIMEOUT:\n # La v\u00e9rification manuelle du statut de la transaction se fait automatiquement si timeout donc si timeout est lev\u00e9 pas besoin de rev\u00e9rifier manuellement le status sur le coup.\n\n print(\"\\nLa transaction a expir\u00e9.\\n\")\n\n elif status == EventFutureStatus.CANCELLED:\n print(\"\\nTransaction annul\u00e9e par l'utilisateur\\n\")\n\n # Creation de l'instance Fedapay Connector\n fedapay = FedapayConnector(use_listen_server= True) \n\n # Configuration des callbacks\n fedapay.set_payment_callback_function(payment_callback) # executer a chaques appels reussi a fedapay_pay()\n fedapay.set_webhook_callback_function(webhook_callback) # executer \u00e0 la r\u00e9ception de webhooks fedapay valides\n fedapay.set_on_persited_listening_processes_loading_finished_callback(run_after_finalise) \n # \u00e9xectuer lors de la r\u00e9cup\u00e9ration des ecoutes d'event fedapay perduent lors d'un potentiel \n # redemarrage de l'app pendant que des ecoutes sont actives.\n\n # lancement de la restauration des processus d'\u00e9coute\n await fedapay.load_persisted_listening_processes()\n\n # D\u00e9marrage du listener interne\n fedapay.start_webhook_server()\n\n # Configuration paiement\n setup = PaiementSetup(\n pays=Pays.benin,\n method=MethodesPaiement.mtn_open\n )\n \n client = UserData(\n nom=\"Doe\",\n prenom=\"John\",\n email=\"john@example.com\",\n tel=\"0162626262\"\n )\n\n # Ex\u00e9cution paiement\n resp = await fedapay.fedapay_pay(\n setup=setup,\n client_infos=client,\n montant_paiement=1000,\n payment_contact=\"0162626262\"\n )\n\n # Attente r\u00e9sultat\n status, webhooks = await fedapay.fedapay_finalise(resp.id_transaction)\n\n await run_after_finalise(status, webhooks)\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n### 4. Mode API Existante (Int\u00e9gration FastAPI ou framework similaire)\n\nDans 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\u00e9tiers pour le traitement des transaction.\n\n```python\nfrom fastapi import FastAPI\nfrom contextlib import asynccontextmanager\nfrom fedapay_connector import FedapayConnector\n\n\n... code du fichier main.py ...\n\n@asynccontextmanager\nasync def lifespan(app: FastAPI):\n # Creation de l'instance Fedapay Connector\n fedapay = FedapayConnector(use_listen_server= False) \n\n # importer ou definissez prealablement les callabacks si voulu\n # Configuration des callbacks\n fedapay.set_payment_callback_function(payment_callback) # executer a chaques appels reussi a fedapay_pay()\n fedapay.set_webhook_callback_function(webhook_callback) # executer \u00e0 la r\u00e9ception de webhooks fedapay valides\n fedapay.set_on_persited_listening_processes_loading_finished_callback(run_after_finalise) # \u00e9xectuer lors de la r\u00e9cup\u00e9ration des ecoutes d'event fedapay perduent lors d'un potentiel redemarrage de l'app pendant que des ecoutes sont actives.\n\n # lancement de la restauration des processus d'\u00e9coute\n await fedapay.load_persisted_listening_processes()\n\n yield\n\n #permet un arret propre de fedapay connector\n await fedapay.shutdown_cleanup()\n\napp = FastAPI(lifespan=lifespan)\n\n@app.post(\"/webhooks/fedapay\")\nasync def fedapay_webhook(request: Request):\n payload = await request.body()\n headers = request.headers\n \n # Validation signature\n signature = headers.get(\"x-fedapay-signature\")\n FedapayConnector().verify_signature(payload, signature)\n \n # Traitement webhook\n event = await request.json()\n await FedapayConnector().fedapay_save_webhook_data(event)\n \n return {\"status\": \"success\"}\n \n... suite de votre code ...\n```\n\nSi les methodes de paiement que vous souhaiter utilis\u00e9s ne sont pas disponibles en paiement sans redirection vous devrez recup\u00e9rer le paiement link et le retourner au front end pour affichage dans une webview ou un element similaire pour finalisation par l'utilisateur.\nLe satut sera toutefois toujours capturer par le backend directement donc il n'est pas neccessaire de le recup\u00e9rer cot\u00e9 client. \n\n## Fonctionnalit\u00e9s Avanc\u00e9es\n\n### Gestion des Webhooks\n\n```python\n# 1. Serveur Int\u00e9gr\u00e9\nfedapay = FedapayConnector(\n use_listen_server=True,\n listen_server_port=3000,\n listen_server_endpoint_name=\"webhooks\"\n)\nfedapay.start_webhook_server()\n\n# 2. Int\u00e9gration API Existante\nfedapay = FedapayConnector(use_listen_server=False)\nawait fedapay.fedapay_save_webhook_data(webhook_data)\n```\n\n### Callbacks Personnalis\u00e9s\n\n```python\nasync def on_payment(payment: PaymentHistory):\n \"\"\"Appel\u00e9 apr\u00e8s chaque paiement\"\"\"\n print(f\"Nouveau paiement: {payment.id}\")\n \nasync def on_webhook(webhook: WebhookHistory):\n \"\"\"Appel\u00e9 pour chaque webhook\"\"\"\n print(f\"Webhook re\u00e7u: {webhook.name}\")\n\nasync def run_after_finalise(\n status: EventFutureStatus, data: list[WebhookHistory] | None\n):\n \"\"\"Appel\u00e9 apr\u00e8s la r\u00e9solution d'\u00e9coutes r\u00e9cup\u00e9r\u00e9es \"\"\"\n \n if status == EventFutureStatus.RESOLVED:\n print(\"\\nTransaction r\u00e9ussie\\n\")\n print(f\"\\nDonn\u00e9es finales : {data}\\n\")\n\n elif status == EventFutureStatus.TIMEOUT:\n print(\"\\nLa transaction a expir\u00e9.\\n\")\n\n elif status == EventFutureStatus.CANCELLED:\n print(\"\\nTransaction annul\u00e9e par l'utilisateur\\n\")\n\nfedapay.set_payment_callback_function(on_payment)\nfedapay.set_webhook_callback_function(on_webhook)\nfedapay.set_on_persited_listening_processes_loading_finished_callback(run_after_finalise)\n```\n\n### Persistence et Restauration\n\nLe module g\u00e8re automatiquement :\n- Sauvegarde des transactions en cours\n- Restauration apr\u00e8s red\u00e9marrage\n- Reprise des \u00e9couteurs interrompus\n- Synchronisation avec FedaPay\n\n## \ud83d\udd27 D\u00e9pannage\n\n### Probl\u00e8mes Courants\n\n1. **Les webhooks ne sont pas re\u00e7us**\n - V\u00e9rifier l'URL configur\u00e9e dans FedaPay\n - V\u00e9rifier la cl\u00e9 secr\u00e8te webhook\n\n2. **Erreurs de timeout**\n - Augmenter la valeur du timeout\n - V\u00e9rifier la connexion r\u00e9seau\n - Consulter les logs pour plus de d\u00e9tails\n\n## Contribution\n\nLes contributions sont les bienvenues!\n\n## Licence\n\nCe 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.\n\n## \ud83d\udd12 S\u00e9curit\u00e9\n\n- Ne jamais exposer les cl\u00e9s API\n- Toujours valider les signatures webhook\n- Utiliser HTTPS en production\n- Impl\u00e9menter des timeouts appropri\u00e9s",
"bugtrack_url": null,
"license": "AGPL-3.0-or-later",
"summary": "Un client asynchrone robuste pour l'API FedaPay, offrant une gestion automatis\u00e9e des paiements avec support complet des webhooks.",
"version": "1.3.2",
"project_urls": {
"Homepage": "https://github.com/Dasero197/fedapay_connector",
"Issues": "https://github.com/Dasero197/fedapay_connector/issues"
},
"split_keywords": [
"api",
" asynchrone",
" connector",
" fedapay",
" paiement"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "aaaccbcd71d74d47425dac341b066795ad372c533d1d95c9a2fbd1c0ee2c38ae",
"md5": "86bb3910047dcf047a1ed682dbfdf1c7",
"sha256": "4acf2117d001ee61a148222e7c71a0066f33935c5f4bce31eac936c0ac4dc6a9"
},
"downloads": -1,
"filename": "fedapay_connector-1.3.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "86bb3910047dcf047a1ed682dbfdf1c7",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 26474,
"upload_time": "2025-07-20T19:57:23",
"upload_time_iso_8601": "2025-07-20T19:57:23.966762Z",
"url": "https://files.pythonhosted.org/packages/aa/ac/cbcd71d74d47425dac341b066795ad372c533d1d95c9a2fbd1c0ee2c38ae/fedapay_connector-1.3.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "dfc1ac41b0e5de1f6b3351b6e703e615145c7aa4f88838bfab91b35418718df8",
"md5": "5c9ac06b1cf7dc0677c6893e8ed96757",
"sha256": "0851cea30894e50b7090ded80570b8707fa3c64ffbf97d22278dbddbc48e0f4f"
},
"downloads": -1,
"filename": "fedapay_connector-1.3.2.tar.gz",
"has_sig": false,
"md5_digest": "5c9ac06b1cf7dc0677c6893e8ed96757",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 23923,
"upload_time": "2025-07-20T19:57:32",
"upload_time_iso_8601": "2025-07-20T19:57:32.159579Z",
"url": "https://files.pythonhosted.org/packages/df/c1/ac41b0e5de1f6b3351b6e703e615145c7aa4f88838bfab91b35418718df8/fedapay_connector-1.3.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-20 19:57:32",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Dasero197",
"github_project": "fedapay_connector",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [
{
"name": "aiohappyeyeballs",
"specs": [
[
"==",
"2.6.1"
]
]
},
{
"name": "aiohttp",
"specs": [
[
"==",
"3.12.14"
]
]
},
{
"name": "aiosignal",
"specs": [
[
"==",
"1.3.2"
]
]
},
{
"name": "annotated-types",
"specs": [
[
"==",
"0.7.0"
]
]
},
{
"name": "anyio",
"specs": [
[
"==",
"4.9.0"
]
]
},
{
"name": "attrs",
"specs": [
[
"==",
"25.3.0"
]
]
},
{
"name": "build",
"specs": [
[
"==",
"1.2.2.post1"
]
]
},
{
"name": "certifi",
"specs": [
[
"==",
"2025.1.31"
]
]
},
{
"name": "cffi",
"specs": [
[
"==",
"1.17.1"
]
]
},
{
"name": "charset-normalizer",
"specs": [
[
"==",
"3.4.1"
]
]
},
{
"name": "click",
"specs": [
[
"==",
"8.1.8"
]
]
},
{
"name": "cryptography",
"specs": [
[
"==",
"44.0.2"
]
]
},
{
"name": "dnspython",
"specs": [
[
"==",
"2.7.0"
]
]
},
{
"name": "docutils",
"specs": [
[
"==",
"0.21.2"
]
]
},
{
"name": "email_validator",
"specs": [
[
"==",
"2.2.0"
]
]
},
{
"name": "fastapi",
"specs": [
[
"==",
"0.115.12"
]
]
},
{
"name": "frozenlist",
"specs": [
[
"==",
"1.5.0"
]
]
},
{
"name": "greenlet",
"specs": [
[
"==",
"3.2.3"
]
]
},
{
"name": "h11",
"specs": [
[
"==",
"0.16.0"
]
]
},
{
"name": "httpcore",
"specs": [
[
"==",
"1.0.8"
]
]
},
{
"name": "httpx",
"specs": [
[
"==",
"0.28.1"
]
]
},
{
"name": "id",
"specs": [
[
"==",
"1.5.0"
]
]
},
{
"name": "idna",
"specs": [
[
"==",
"3.10"
]
]
},
{
"name": "iniconfig",
"specs": [
[
"==",
"2.1.0"
]
]
},
{
"name": "jaraco.classes",
"specs": [
[
"==",
"3.4.0"
]
]
},
{
"name": "jaraco.context",
"specs": [
[
"==",
"6.0.1"
]
]
},
{
"name": "jaraco.functools",
"specs": [
[
"==",
"4.1.0"
]
]
},
{
"name": "jeepney",
"specs": [
[
"==",
"0.9.0"
]
]
},
{
"name": "keyring",
"specs": [
[
"==",
"25.6.0"
]
]
},
{
"name": "markdown-it-py",
"specs": [
[
"==",
"3.0.0"
]
]
},
{
"name": "mdurl",
"specs": [
[
"==",
"0.1.2"
]
]
},
{
"name": "more-itertools",
"specs": [
[
"==",
"10.6.0"
]
]
},
{
"name": "multidict",
"specs": [
[
"==",
"6.4.3"
]
]
},
{
"name": "nh3",
"specs": [
[
"==",
"0.2.21"
]
]
},
{
"name": "packaging",
"specs": [
[
"==",
"24.2"
]
]
},
{
"name": "pluggy",
"specs": [
[
"==",
"1.5.0"
]
]
},
{
"name": "propcache",
"specs": [
[
"==",
"0.3.1"
]
]
},
{
"name": "pycparser",
"specs": [
[
"==",
"2.22"
]
]
},
{
"name": "pydantic",
"specs": [
[
"==",
"2.11.3"
]
]
},
{
"name": "pydantic_core",
"specs": [
[
"==",
"2.33.1"
]
]
},
{
"name": "Pygments",
"specs": [
[
"==",
"2.19.1"
]
]
},
{
"name": "pyproject_hooks",
"specs": [
[
"==",
"1.2.0"
]
]
},
{
"name": "pytest",
"specs": [
[
"==",
"8.3.5"
]
]
},
{
"name": "pytest-asyncio",
"specs": [
[
"==",
"0.26.0"
]
]
},
{
"name": "readme_renderer",
"specs": [
[
"==",
"44.0"
]
]
},
{
"name": "requests",
"specs": [
[
"==",
"2.32.4"
]
]
},
{
"name": "requests-toolbelt",
"specs": [
[
"==",
"1.0.0"
]
]
},
{
"name": "rfc3986",
"specs": [
[
"==",
"2.0.0"
]
]
},
{
"name": "rich",
"specs": [
[
"==",
"14.0.0"
]
]
},
{
"name": "SecretStorage",
"specs": [
[
"==",
"3.3.3"
]
]
},
{
"name": "sniffio",
"specs": [
[
"==",
"1.3.1"
]
]
},
{
"name": "SQLAlchemy",
"specs": [
[
"==",
"2.0.41"
]
]
},
{
"name": "starlette",
"specs": [
[
"==",
"0.46.2"
]
]
},
{
"name": "twine",
"specs": [
[
"==",
"6.1.0"
]
]
},
{
"name": "typing-inspection",
"specs": [
[
"==",
"0.4.0"
]
]
},
{
"name": "typing_extensions",
"specs": [
[
"==",
"4.13.2"
]
]
},
{
"name": "urllib3",
"specs": [
[
"==",
"2.5.0"
]
]
},
{
"name": "uvicorn",
"specs": [
[
"==",
"0.34.2"
]
]
},
{
"name": "yarl",
"specs": [
[
"==",
"1.19.0"
]
]
}
],
"lcname": "fedapay-connector"
}
