"""faymaco.py — minimal Faymaco API client (Python 3.8+, requires `requests`). Docs: https://docs.fayma.co - LLM guide: https://docs.fayma.co/llms.txt from faymaco import Faymaco, verify_webhook fmc = Faymaco(api_key=os.environ["FAYMACO_API_KEY"]) """ import hashlib import hmac import re import time import requests class FaymacoError(Exception): def __init__(self, message, code=None, status=None, details=None): super().__init__(message) self.code = code self.status = status self.details = details class Faymaco: def __init__(self, api_key, base_url="https://apifayko.peelo.chat/api/v1"): if not api_key: raise ValueError("Faymaco: api_key is required") self.api_key = api_key self.base_url = base_url.rstrip("/") def _request(self, method, path, body=None, params=None, idempotency_key=None): headers = {"Authorization": f"Bearer {self.api_key}"} if idempotency_key: headers["Idempotency-Key"] = idempotency_key resp = requests.request( method, self.base_url + path, json=body, params=params, headers=headers, timeout=30 ) try: data = resp.json() except ValueError: data = {} if not resp.ok or data.get("success") is False: err = data.get("error") or {} raise FaymacoError( err.get("message") or f"HTTP {resp.status_code}", code=err.get("code"), status=resp.status_code, details=err.get("details"), ) return data.get("data") def create_subscription( self, customer, amount, frequency, currency="XOF", start_date=None, start_next_month=None, webhooks=None, idempotency_key=None, ): """customer = {"name": ..., "phone": "+221..."}; frequency in monthly|quarterly|semi_annual|annual.""" body = {"customer": customer, "amount": amount, "frequency": frequency, "currency": currency} if start_date is not None: body["startDate"] = start_date if start_next_month is not None: body["startNextMonth"] = start_next_month if webhooks is not None: body["webhooks"] = webhooks return self._request("POST", "/subscriptions", body=body, idempotency_key=idempotency_key) def list_subscriptions(self, status=None, limit=None, cursor=None): params = {k: v for k, v in {"status": status, "limit": limit, "cursor": cursor}.items() if v is not None} return self._request("GET", "/subscriptions", params=params) def get_subscription(self, sub_id): return self._request("GET", f"/subscriptions/{sub_id}") def pause_subscription(self, sub_id): return self._request("POST", f"/subscriptions/{sub_id}/pause") def resume_subscription(self, sub_id): return self._request("POST", f"/subscriptions/{sub_id}/resume") def cancel_subscription(self, sub_id): return self._request("POST", f"/subscriptions/{sub_id}/cancel") def create_payment_request( self, customer, amount, currency="XOF", due_date=None, reminders=None, webhooks=None, external_ref=None, expires_in=None, expires_at=None, replace_existing=None, idempotency_key=None, ): """One-off payment request. customer = {"name": ..., "phone": "+221..."}; reminders = list of ISO dates, each strictly after due_date and before expiry; expires_in = checkout validity in seconds from due_date (60s .. 90 days); replace_existing = cancel the open request of that number instead of raising 409.""" body = {"customer": customer, "amount": amount, "currency": currency} if due_date is not None: body["dueDate"] = due_date if reminders is not None: body["reminders"] = reminders if webhooks is not None: body["webhooks"] = webhooks if external_ref is not None: body["externalRef"] = external_ref if expires_in is not None: body["expiresIn"] = expires_in if expires_at is not None: body["expiresAt"] = expires_at if replace_existing is not None: body["replaceExisting"] = replace_existing return self._request("POST", "/payment-requests", body=body, idempotency_key=idempotency_key) def create_checkout(self, customer, amount, expires_in=1800, **kwargs): """Pay-once checkout: one WhatsApp message, no reminder, and the request closes itself (status "expired" + payment_request.expired webhook) if unpaid within expires_in seconds. An abandoned checkout never lingers in the customer's list nor blocks the next one.""" kwargs.pop("reminders", None) # a checkout never reminds kwargs.setdefault("replace_existing", True) return self.create_payment_request(customer, amount, expires_in=expires_in, **kwargs) def list_payment_requests(self, status=None, source=None, limit=None, cursor=None): params = { k: v for k, v in {"status": status, "source": source, "limit": limit, "cursor": cursor}.items() if v is not None } return self._request("GET", "/payment-requests", params=params) def get_payment_request(self, request_id): return self._request("GET", f"/payment-requests/{request_id}") def cancel_payment_request(self, request_id): return self._request("POST", f"/payment-requests/{request_id}/cancel") def verify_webhook(raw_body, signature_header, secret, tolerance_sec=300): """Verify a Faymaco webhook signature. raw_body: the RAW (unparsed) request body, str or bytes, exactly as received. signature_header: value of the `X-Faymaco-Signature` header. secret: your account webhook secret (Developers page). """ m = re.search(r"t=(\d+),v1=([0-9a-f]+)", signature_header or "") if not m: return False ts, v1 = m.group(1), m.group(2) if abs(time.time() - int(ts)) > tolerance_sec: return False if isinstance(raw_body, bytes): raw_body = raw_body.decode("utf-8") expected = hmac.new(secret.encode(), f"{ts}.{raw_body}".encode(), hashlib.sha256).hexdigest() return hmac.compare_digest(expected, v1) # ─── Example: create a subscription ────────────────────────────────────── # import os # fmc = Faymaco(api_key=os.environ["FAYMACO_API_KEY"]) # data = fmc.create_subscription( # customer={"name": "Awa Diop", "phone": "+221770000000"}, # amount=5000, # frequency="monthly", # webhooks={"onSuccess": "https://your-app.com/webhooks/faymaco", # "onExpired": "https://your-app.com/webhooks/faymaco"}, # idempotency_key="order-12345", # ) # print(data["subscription"]["id"], data["subscription"]["status"]) # ─── Example: Flask webhook endpoint ───────────────────────────────────── # from flask import Flask, request # app = Flask(__name__) # # @app.post("/webhooks/faymaco") # def faymaco_webhook(): # raw = request.get_data() # raw bytes — needed for signature # if not verify_webhook(raw, request.headers.get("X-Faymaco-Signature"), # os.environ["FAYMACO_WEBHOOK_SECRET"]): # return "bad signature", 400 # evt = request.get_json() # if evt["event"] == "subscription.payment.succeeded": # pass # grant access for evt["data"]["subscriptionId"] # elif evt["event"] == "subscription.payment.failed": # pass # revoke / flag overdue # elif evt["event"] == "payment_request.succeeded": # pass # mark the order paid — match it with evt["data"]["externalRef"] # elif evt["event"] == "payment_request.expired": # pass # checkout abandoned — release the order you reserved # return "", 200 # ─── Example: one-off payment request ──────────────────────────────────── # data = fmc.create_payment_request( # customer={"name": "Awa Diop", "phone": "+221770000000"}, # amount=25000, # due_date="2026-08-20T09:00:00Z", # reminders=["2026-08-22T09:00:00Z", "2026-08-25T09:00:00Z"], # webhooks={"onSuccess": "https://your-app.com/webhooks/faymaco"}, # external_ref="order-4821", # idempotency_key="order-4821", # ) # print(data["paymentRequest"]["id"], data["paymentRequest"]["status"]) # ─── Example: pay-once checkout (no reminder, auto-expiry) ─────────────── # One WhatsApp message. Unpaid after 30 min → status "expired" and a # payment_request.expired webhook; nothing lingers in the customer's list. # data = fmc.create_checkout( # customer={"name": "Awa Diop", "phone": "+221770000000"}, # amount=25000, # expires_in=1800, # webhooks={"onSuccess": "https://your-app.com/webhooks/faymaco", # "onExpired": "https://your-app.com/webhooks/faymaco"}, # external_ref="order-4821", # idempotency_key="order-4821", # ) # print(data["paymentRequest"]["id"], data["paymentRequest"]["expiresAt"])