Python

Compatibility

Python 3.6+. Only depends on requests; everything else is stdlib.

Install

pip install requests

Client Implementation

import time, hmac, hashlib, secrets, json
import requests

APP_ID = "cp_a1b2c3d4e5f6071829304a5b6c7d8e9f"
SECRET = "f1e2d3c4b5a69788776655443322110099aabbccddeeff00112233445566778899"
BASE = "https://api.coinepay.net"


def call(method: str, path: str, body_dict: dict | None = None,
         idempotency_key: str | None = None) -> dict:
    """Unified OpenAPI call"""
    body = json.dumps(body_dict or {}, separators=(",", ":")).encode("utf-8")
    body_hash = hashlib.sha256(body).hexdigest()
    ts = str(int(time.time()))
    nonce = secrets.token_hex(16)
    sign_input = f"{method}\n{path}\n\n{ts}\n{nonce}\n{body_hash}"
    sig = hmac.new(SECRET.encode(), sign_input.encode(), hashlib.sha256).hexdigest()

    headers = {
        "X-App-Id": APP_ID,
        "X-Timestamp": ts,
        "X-Nonce": nonce,
        "X-Signature": sig,
        "Content-Type": "application/json",
    }
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key

    return requests.request(method, BASE + path, headers=headers, data=body, timeout=30).json()


# Webhook signature verification
def verify_webhook(signature_header: str, timestamp_header: str,
                   raw_body: bytes, webhook_secret: str) -> bool:
    if not signature_header.startswith("v1,"):
        return False
    sig = signature_header[3:]
    expected = hmac.new(
        webhook_secret.encode(),
        f"{timestamp_header}.".encode() + raw_body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(sig, expected)

Usage

import uuid

# 1. List card headers
resp = call("POST", "/api/v1/openapi/card_headers/list", {"page": 1, "page_size": 20})
print(resp["data"]["list"][0]["header_id"])

# 2. List card configs
resp = call("POST", "/api/v1/openapi/card_configs/list", {"header_id": "hdr_5"})
package_id = resp["data"]["list"][0]["package_id"]

# 3. Apply virtual card (with idempotency key)
resp = call("POST", "/api/v1/openapi/card/apply",
            {"header_id": "hdr_5", "package_id": package_id,
             "first_name": "John", "last_name": "Doe"},
            idempotency_key=str(uuid.uuid4()))
card_id = resp["data"]["card_id"]
print(f"Created card: {card_id}, status={resp['data']['status']}")

# 4. Create recharge
resp = call("POST", "/api/v1/openapi/card/recharge",
            {"card_id": card_id, "amount": "100.00"},
            idempotency_key=str(uuid.uuid4()))
print(f"Transaction: {resp['data']['transaction_id']}")

Webhook Receiver (FastAPI)

from fastapi import FastAPI, Request, HTTPException
import os

app = FastAPI()
WEBHOOK_SECRET = os.environ["COINEPAY_WEBHOOK_SECRET"]

@app.post("/webhook")
async def webhook(request: Request):
    raw = await request.body()
    sig = request.headers.get("Webhook-Signature", "")
    ts  = request.headers.get("Webhook-Timestamp", "")
    event_id = request.headers.get("Webhook-Id", "")
    event_type = request.headers.get("Webhook-Type", "")

    if not verify_webhook(sig, ts, raw, WEBHOOK_SECRET):
        raise HTTPException(401, "invalid signature")

    # Timestamp ±5 min
    if abs(int(time.time()) - int(ts)) > 300:
        raise HTTPException(401, "timestamp expired")

    # Dedupe by event_id
    if already_processed(event_id):
        return {"ok": True}

    payload = json.loads(raw)
    handle_event(event_type, payload)
    mark_processed(event_id)

    return {"ok": True}

Defer heavy work

handle_event should only do "persist + enqueue". Don't do slow operations (DB queries, external calls) in the webhook handler — anything over 5 seconds will be re-delivered.

Errors & Retry

import time

class coinepayError(Exception): ...
class RateLimited(coinepayError): ...
class IdempotencyConflict(coinepayError): ...

def call_with_retry(method, path, body=None, idempotency_key=None,
                    max_attempts=4):
    for attempt in range(max_attempts):
        resp = call(method, path, body, idempotency_key)
        code = resp.get("code", 500)

        if code == 200:
            return resp["data"]
        if code == 429 and attempt < max_attempts - 1:
            time.sleep(min(2 ** attempt, 30))
            continue
        if code >= 500 and attempt < max_attempts - 1:
            time.sleep(2 ** attempt)
            continue
        if code == 409:
            raise IdempotencyConflict(resp.get("message"))
        raise coinepayError(f"{code}: {resp.get('message')}")

    raise coinepayError("max attempts exceeded")