Code examples

Code Examples

Production-ready code examples in multiple programming languages.

‍

Configuration

Python

PYTHON

import requests
from typing import Optional, Dict, Any, List

class PreciumClient:
    def __init__(self, api_key: str, brand_id: str):
        self.api_key = api_key
        self.brand_id = brand_id
        self.base_url = "https://gate.reviopay.com/api/v1"
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })

    def _request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
        url = f"{self.base_url}{endpoint}"
        response = self.session.request(method, url, json=data)
        response.raise_for_status()
        return response.json() if response.text else {}

# Initialize
precium = PreciumClient("your_api_key", "your_brand_id")

‍

Node.js

JSX

const axios = require('axios');

class PreciumClient {
  constructor(apiKey, brandId) {
    this.brandId = brandId;
    this.client = axios.create({
      baseURL: 'https://gate.reviopay.com/api/v1',
      headers: {
        'Authorization': `Bearer ${apiKey}`,
        'Content-Type': 'application/json'
      }
    });
  }

  async request(method, endpoint, data = null) {
    const response = await this.client.request({ method, url: endpoint, data });
    return response.data;
  }
}

const precium = new PreciumClient('your_api_key', 'your_brand_id');
module.exports = precium;

‍

PHP

PHP

<?php
class PreciumClient {
    private string $apiKey;
    private string $brandId;
    private string $baseUrl = 'https://gate.reviopay.com/api/v1';

    public function __construct(string $apiKey, string $brandId) {
        $this->apiKey = $apiKey;
        $this->brandId = $brandId;
    }

    public function request(string $method, string $endpoint, ?array $data = null): array {
        $ch = curl_init();
        curl_setopt_array($ch, [
            CURLOPT_URL => $this->baseUrl . $endpoint,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_CUSTOMREQUEST => strtoupper($method),
            CURLOPT_HTTPHEADER => [
                'Authorization: Bearer ' . $this->apiKey,
                'Content-Type: application/json'
            ]
        ]);
        if ($data !== null) {
            curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($data));
        }
        $response = curl_exec($ch);
        curl_close($ch);
        return json_decode($response, true) ?? [];
    }
}

$precium = new PreciumClient('your_api_key', 'your_brand_id');

‍

Payment Link

Create Payment Link

‍

Python:

PYTHON

def create_payment_link(email: str, amount_cents: int, product_name: str, reference: str = None) -> Dict:
    data = {
        "client": {"email": email},
        "purchase": {
            "currency": "ZAR",
            "products": [{"name": product_name, "price": amount_cents}]
        },
        "brand_id": precium.brand_id
    }
    if reference:
        data["reference"] = reference
    return precium._request("POST", "/purchases/", data)

# Usage
payment = create_payment_link(
    email='customer@example.com',
    amount_cents=150000,
    product_name='Invoice #12345'
)
print(f"Payment link: {payment['checkout_url']}")

‍

Node.js:

JSX

async function createPaymentLink(email, amountCents, productName, reference = null) {
  const data = {
    client: { email },
    purchase: {
      currency: 'ZAR',
      products: [{ name: productName, price: amountCents }]
    },
    brand_id: precium.brandId
  };
  if (reference) data.reference = reference;
  return await precium.request('POST', '/purchases/', data);
}

// Usage
const payment = await createPaymentLink('customer@example.com', 150000, 'Invoice #12345');
console.log(`Payment link: ${payment.checkout_url}`);

‍

E-commerce Checkout

Create Checkout with Redirects

‍

Python:

PYTHON

def create_checkout(
    client_id: str,
    products: List[Dict],
    order_id: str,
    success_url: str,
    failure_url: str,
    payment_methods: List[str] = None
) -> Dict:
    data = {
        "client_id": client_id,
        "purchase": {"currency": "ZAR", "products": products},
        "brand_id": precium.brand_id,
        "reference": order_id,
        "success_redirect": success_url,
        "failure_redirect": failure_url
    }
    if payment_methods:
        data["payment_method_whitelist"] = payment_methods
    return precium._request("POST", "/purchases/", data)

# Usage
checkout = create_checkout(
    client_id='client-uuid',
    products=[
        {"name": "Widget", "price": 5000, "quantity": "2"},
        {"name": "Shipping", "price": 5000}
    ],
    order_id='ORD-2026-0001',
    success_url='https://shop.com/success',
    failure_url='https://shop.com/failure',
    payment_methods=["visa", "mastercard", "ozow"]
)

‍

Node.js:

JSX

async function createCheckout(clientId, products, orderId, successUrl, failureUrl, paymentMethods = null) {
  const data = {
    client_id: clientId,
    purchase: { currency: 'ZAR', products },
    brand_id: precium.brandId,
    reference: orderId,
    success_redirect: successUrl,
    failure_redirect: failureUrl
  };
  if (paymentMethods) data.payment_method_whitelist = paymentMethods;
  return await precium.request('POST', '/purchases/', data);
}

‍

Recurring Payments

Tokenize Card (First Payment)

‍

Python:

PYTHON

def create_tokenization_purchase(
    client_id: str,
    amount_cents: int,
    product_name: str,
    success_url: str,
    failure_url: str
) -> Dict:
    return precium._request("POST", "/purchases/", {
        "client_id": client_id,
        "purchase": {
            "currency": "ZAR",
            "products": [{"name": product_name, "price": amount_cents}]
        },
        "brand_id": precium.brand_id,
        "force_recurring": True,
        "success_redirect": success_url,
        "failure_redirect": failure_url
    })

# Store the returned purchase ID as your recurring token

‍

Charge with Token

‍

Python:

PYTHON

def charge_recurring(
    client_id: str,
    recurring_token: str,
    amount_cents: int,
    product_name: str,
    reference: str
) -> Dict:
    # Create purchase
    purchase = precium._request("POST", "/purchases/", {
        "client_id": client_id,
        "purchase": {
            "currency": "ZAR",
            "products": [{"name": product_name, "price": amount_cents}]
        },
        "brand_id": precium.brand_id,
        "reference": reference
    })

    # Charge using token
    return precium._request(
        "POST",
        f"/purchases/{purchase['id']}/charge/",
        {"recurring_token": recurring_token}
    )

‍

Refunds

Python:

PYTHON

def refund_full(purchase_id: str) -> Dict:
    return precium._request("POST", f"/purchases/{purchase_id}/refund/", {})

def refund_partial(purchase_id: str, amount_cents: int) -> Dict:
    return precium._request("POST", f"/purchases/{purchase_id}/refund/", {"amount": amount_cents})

‍

Node.js:

JSX

async function refund(purchaseId, amountCents = null) {
  const data = amountCents ? { amount: amountCents } : {};
  return await precium.request('POST', `/purchases/${purchaseId}/refund/`, data);
}

‍

Webhooks

Express.js Handler

JSX

const crypto = require('crypto');

// Callbacks are signed with ASYMMETRIC RSA — there is no shared secret.
// X-Signature = base64( RSA PKCS#1 v1.5 over the SHA-256 digest of the raw body )
// Key source: public_key on the webhook object, or GET /public_key/ for a
// per-purchase success_callback. Store the PEM, do not fetch it per request.
const PUBLIC_KEY_PEM = process.env.PRECIUM_WEBHOOK_PUBLIC_KEY;

function verifySignature(rawBody, signatureB64) {
  const verifier = crypto.createVerify('RSA-SHA256');
  verifier.update(rawBody);            // raw Buffer, not a re-serialised object
  verifier.end();
  return verifier.verify(PUBLIC_KEY_PEM, Buffer.from(signatureB64, 'base64'));
}

// express.raw is essential: JSON.parse then re-stringify breaks the signature.
app.post('/webhooks/precium', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.headers['x-signature'];

  if (!signature || !verifySignature(req.body, signature)) {
    return res.status(401).send('Invalid signature');
  }

  const event = JSON.parse(req.body.toString());

  // Deliveries can repeat even after a 200, so make this idempotent.
  switch (event.event_type) {
    case 'purchase.paid':
      handlePaymentSuccess(event.data);
      break;
    case 'purchase.payment_failure':
      handlePaymentFailure(event.data);
      break;
    case 'payment.refunded':
      handleRefund(event.data);
      break;
  }

  res.status(200).send('OK');   // respond within 5 seconds; queue the work
});

‍

Python Flask Handler

PYTHON

import base64
import os
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.exceptions import InvalidSignature
from flask import Flask, request

app = Flask(__name__)

# Callbacks are signed with ASYMMETRIC RSA — there is no shared secret.
# X-Signature = base64( RSA PKCS#1 v1.5 over the SHA-256 digest of the raw body )
# Key source: public_key on the webhook object, or GET /public_key/ for a
# per-purchase success_callback.
PUBLIC_KEY = serialization.load_pem_public_key(
    os.environ["PRECIUM_WEBHOOK_PUBLIC_KEY"].encode()
)

def verify_signature(raw_body: bytes, signature_b64: str) -> bool:
    try:
        PUBLIC_KEY.verify(
            base64.b64decode(signature_b64),
            raw_body,                 # raw bytes, never request.get_json()
            padding.PKCS1v15(),
            hashes.SHA256(),
        )
        return True
    except (InvalidSignature, ValueError):
        return False

@app.route('/webhooks/precium', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Signature', '')
    raw_body = request.get_data()          # bytes, before any parsing

    if not verify_signature(raw_body, signature):
        return 'Invalid signature', 401

    event = request.get_json()

    # Deliveries can repeat even after a 200, so make this idempotent.
    if event['event_type'] == 'purchase.paid':
        handle_payment_success(event['data'])
    elif event['event_type'] == 'purchase.payment_failure':
        handle_payment_failure(event['data'])
    elif event['event_type'] == 'payment.refunded':
        handle_refund(event['data'])

    return 'OK', 200                       # respond within 5 seconds

‍

Direct Post Form

HTML

<form method='POST' action='{{ direct_post_url }}'>
  <input type='hidden' name='pm' value='visa'>

  <input type='text' name='cardholder_name' maxlength='30' required
         pattern='[A-Za-z\s\.\-']+'>

  <input type='text' name='card_number' maxlength='19' required
         pattern='\d{13,19}'>

  <input type='text' name='expires' placeholder='MM/YY' maxlength='5' required
         pattern='\d{2}/\d{2}'>

  <input type='text' name='cvc' maxlength='4' required
         pattern='\d{3,4}'>

  <label>
    <input type='checkbox' name='remember_card' value='on'>
    Save for future
  </label>

  <button type='submit'>Pay Now</button>
</form>

‍

Subscriptions

Create Plan

PYTHON

def create_subscription_plan(
    title: str,
    price_cents: int,
    period: int = 1,
    period_units: str = "months"
) -> Dict:
    return precium._request("POST", "/billing_templates/", {
        "title": title,
        "purchase": {
            "currency": "ZAR",
            "products": [{"name": title, "price": price_cents}]
        },
        "brand_id": precium.brand_id,
        "is_subscription": True,
        "subscription_active": True,
        "subscription_period": period,
        "subscription_period_units": period_units,
        "force_recurring": True
    })

‍

Add Subscriber

PYTHON

def add_subscriber(template_id: str, client_id: str, reference: str = None) -> Dict:
    data = {"client_id": client_id, "send_receipt": True}
    if reference:
        data["invoice_reference"] = reference
    return precium._request(
        "POST",
        f"/billing_templates/{template_id}/add_subscriber/",
        data
    )

‍

Error Handling

PYTHON

def safe_api_call(func, *args, **kwargs):
    try:
        return func(*args, **kwargs)
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 401:
            print("Authentication failed - check API key")
        elif e.response.status_code == 400:
            errors = e.response.json()
            print(f"Validation error: {errors}")
        elif e.response.status_code == 429:
            print("Rate limited - implement backoff")
        else:
            print(f"API error: {e}")
        return None