Python Examples
Complete Python example client for integrating with the 123hub API usingrequests and hashlib.
This is a reference implementation, not an official SDK package. You can copy
it into your project and adapt it as needed.
Installation
pip install requests
Configuration
Store your credentials in environment variables:export PAYMENT_API_APP_ID="1"
export PAYMENT_API_VERSION="2"
export PAYMENT_API_SECRET_KEY="your_secret_key"
PaymentApiClient Class
A complete, reusable client class for all API operations:import hashlib
import hmac
import json
import os
import time
import uuid
import requests
from typing import Optional, Dict, Any
class PaymentApiClient:
"""123hub API Client for Python"""
def __init__(
self,
app_id: str = None,
api_version: int = None,
secret_key: str = None,
base_url: str = "https://api.bafanglaicai88.com",
timeout=(3.05, 30),
):
self.app_id = app_id or os.environ.get("PAYMENT_API_APP_ID")
self.api_version = api_version or int(os.environ.get("PAYMENT_API_VERSION", "1"))
self.secret_key = secret_key or os.environ.get("PAYMENT_API_SECRET_KEY")
self.base_url = base_url
self.timeout = timeout
if not self.app_id or not self.secret_key:
raise ValueError("app_id and secret_key are required")
if self.api_version not in (1, 2):
raise ValueError("api_version must be 1 or 2")
def _send(self, body: dict) -> dict:
"""Sign and send a request to the 123hub API."""
body_str = json.dumps(body, separators=(",", ":"))
path = f"/public/api/multihub/v{self.api_version}"
headers = {
"Content-Type": "application/json",
"X-Data-Application-Id": self.app_id,
}
if self.api_version == 1:
headers["X-Data-Hash"] = hashlib.sha512(
(body_str + self.secret_key).encode()
).hexdigest()
else:
timestamp = str(int(time.time()))
nonce = os.urandom(24).hex()
canonical = "\n".join([
"quadpay-multihub-request-v2",
self.app_id,
"POST",
"/public/api/multihub/v2",
timestamp,
nonce,
hashlib.sha256(body_str.encode()).hexdigest(),
])
headers.update({
"X-Data-Timestamp": timestamp,
"X-Data-Nonce": nonce,
"X-Data-Signature": hmac.new(
self.secret_key.encode(), canonical.encode(), hashlib.sha512
).hexdigest(),
})
response = requests.post(
self.base_url + path,
data=body_str,
headers=headers,
timeout=self.timeout,
)
response_text = response.text
parsed = json.loads(response_text)
if self.api_version == 2:
request_id = response.headers.get("X-Request-ID") or parsed.get("request_id")
canonical_response = "\n".join([
"quadpay-multihub-response-v2",
self.app_id,
nonce,
request_id or "",
str(response.status_code),
hashlib.sha256(response_text.encode()).hexdigest(),
])
expected = hmac.new(
self.secret_key.encode(), canonical_response.encode(), hashlib.sha512
).hexdigest()
received = response.headers.get("X-Data-Signature", "")
if not request_id or not hmac.compare_digest(received, expected):
raise ValueError("Merchant API v2 response signature verification failed")
return parsed
def ping(self) -> dict:
"""Health check / connectivity test."""
return self._send({"method": "gateway.ping", "params": {}})
def create_deposit(
self,
service_id: int,
amount: int,
currency: str,
c_id: str,
**kwargs,
) -> dict:
"""Create a deposit (pay-in) payment."""
return self._send({
"method": "payment.in",
"service_id": service_id,
"params": {
"payment": {
"identifiers": {"c_id": c_id},
"amount": {"value": amount, "currency": currency},
**kwargs,
}
},
})
def create_withdrawal(
self,
service_id: int,
amount: int,
currency: str,
c_id: str,
receiver: dict,
**kwargs,
) -> dict:
"""Create a withdrawal (payout) payment."""
return self._send({
"method": "payment.out",
"service_id": service_id,
"params": {
"payment": {
"identifiers": {"c_id": c_id},
"amount": {"value": amount, "currency": currency},
"receiver": receiver,
**kwargs,
}
},
})
def get_status(
self, c_id: str = None, h_id: str = None
) -> dict:
"""Check payment status by identifiers."""
identifiers = {}
if c_id:
identifiers["c_id"] = c_id
if h_id:
identifiers["h_id"] = h_id
return self._send({
"method": "payment.status",
"params": {"payment": {"identifiers": identifiers}},
})
def get_balance(self) -> dict:
"""Retrieve account balances."""
return self._send({"method": "balance.get", "params": {}})
Usage Examples
Ping (Health Check)
client = PaymentApiClient(app_id="1", api_version=2, secret_key="your_secret_key")
result = client.ping()
print(result)
# {"success": true, "result": {"message": "pong"}, ...}
Create a Deposit Payment
The examples below use INR (India), but the same client works for all supported currencies (MXN, ARS, TRY, AUD, LKR, UYU) — just change theservice_id, currency, and country values.
client = PaymentApiClient(app_id="1", api_version=2, secret_key="your_secret_key")
result = client.create_deposit(
service_id=14701,
amount=10000,
currency="INR",
c_id="12345",
description="Order #12345",
payer={
"email": "customer@example.com",
"person": {
"first_name": "John",
"last_name": "Doe",
},
},
)
if result["success"]:
payment = result["result"]["payment"]
print(f"Payment created successfully")
print(f"Request ID: {result['request_id']}")
else:
print(f"Error: {result['error']['code']} - {result['error']['message']}")
Create a Withdrawal (Payout)
result = client.create_withdrawal(
service_id=14701,
amount=5000,
currency="INR",
c_id="67890",
receiver={
"bank": {
"account": {"id": "1234567890"},
"ifsc": "SBIN0001234",
},
"person": {
"first_name": "Jane",
"last_name": "Doe",
},
},
description="Payout #67890",
)
if result["success"]:
print("Withdrawal created successfully")
else:
print(f"Error: {result['error']['code']} - {result['error']['message']}")
Check Payment Status
# Look up by client-side ID
result = client.get_status(c_id="12345")
if result["success"]:
status = result["result"]["payment"]["status"]["status"]
print(f"Payment status: {status}")
else:
print(f"Error: {result['error']['code']} - {result['error']['message']}")
# Look up by hub-side ID
result = client.get_status(h_id="999001")
if result["success"]:
status = result["result"]["payment"]["status"]["status"]
print(f"Payment status: {status}")
Get Balance
result = client.get_balance()
if result["success"]:
for amount in result["result"]["balance"]["amounts"]:
print(f"{amount['currency']}: {amount['value']}")
else:
print(f"Error: {result['error']['code']} - {result['error']['message']}")
Error Handling
Successful responses return HTTP 200, errors return HTTP 400. Always use thesuccess field to determine outcome:
client = PaymentApiClient(app_id="1", api_version=2, secret_key="your_secret_key")
result = client.create_deposit(
service_id=14701,
amount=10000,
currency="INR",
c_id="12345",
)
if result["success"]:
# Operation succeeded
payment = result["result"]
print(f"Request ID: {result['request_id']}")
print(f"Processing time: {result['processing_time']}ms")
else:
# Operation failed
error = result["error"]
print(f"Error code: {error['code']}")
print(f"Error message: {error['message']}")
# Handle specific error codes
if error["code"] == 3000:
print("Authentication failed - check your secret key")
elif error["code"] == 1005 or error["code"] == 9000:
print("Invalid parameters - check request body")
elif error["code"] == 7002:
print("Service temporarily unavailable - retry later")
Handling Network Errors
Network-level errors can still occur independently of API responses:import requests
client = PaymentApiClient(app_id="1", api_version=2, secret_key="your_secret_key")
try:
result = client.ping()
if result["success"]:
print("API is reachable")
else:
print(f"API error: {result['error']['message']}")
except requests.ConnectionError:
print("Cannot reach the API - check network connectivity")
except requests.Timeout:
print("Request timed out - try again")
except requests.RequestException as e:
print(f"Request failed: {e}")
except (KeyError, ValueError) as e:
print(f"Unexpected response format: {e}")
Complete Example
#!/usr/bin/env python3
"""
123hub API Integration Example
Demonstrates the complete flow: ping, check balance, create a deposit,
and poll for status.
"""
import os
import time
import hashlib
import hmac
import json
import requests
class PaymentApiClient:
def __init__(
self,
app_id,
api_version,
secret_key,
base_url="https://api.bafanglaicai88.com",
timeout=(3.05, 30),
):
if api_version not in (1, 2):
raise ValueError("api_version must be 1 or 2")
self.app_id = app_id
self.api_version = api_version
self.secret_key = secret_key
self.base_url = base_url
self.timeout = timeout
def _send(self, body):
body_str = json.dumps(body, separators=(",", ":"))
path = f"/public/api/multihub/v{self.api_version}"
headers = {
"Content-Type": "application/json",
"X-Data-Application-Id": self.app_id,
}
if self.api_version == 1:
headers["X-Data-Hash"] = hashlib.sha512(
(body_str + self.secret_key).encode()
).hexdigest()
else:
timestamp = str(int(time.time()))
nonce = os.urandom(24).hex()
canonical = "\n".join([
"quadpay-multihub-request-v2",
self.app_id,
"POST",
"/public/api/multihub/v2",
timestamp,
nonce,
hashlib.sha256(body_str.encode()).hexdigest(),
])
headers.update({
"X-Data-Timestamp": timestamp,
"X-Data-Nonce": nonce,
"X-Data-Signature": hmac.new(
self.secret_key.encode(), canonical.encode(), hashlib.sha512
).hexdigest(),
})
response = requests.post(
self.base_url + path,
data=body_str,
headers=headers,
timeout=self.timeout,
)
return response.json()
def ping(self):
return self._send({"method": "gateway.ping", "params": {}})
def get_balance(self):
return self._send({"method": "balance.get", "params": {}})
def create_deposit(self, service_id, amount, currency, c_id, **kwargs):
return self._send({
"method": "payment.in",
"service_id": service_id,
"params": {
"payment": {
"identifiers": {"c_id": c_id},
"amount": {"value": amount, "currency": currency},
**kwargs,
}
},
})
def get_status(self, c_id=None, h_id=None):
identifiers = {}
if c_id:
identifiers["c_id"] = c_id
if h_id:
identifiers["h_id"] = h_id
return self._send({
"method": "payment.status",
"params": {"payment": {"identifiers": identifiers}},
})
def main():
app_id = os.environ.get("PAYMENT_API_APP_ID", "1")
api_version = int(os.environ.get("PAYMENT_API_VERSION", "2"))
secret_key = os.environ.get("PAYMENT_API_SECRET_KEY", "your_secret_key")
client = PaymentApiClient(
app_id=app_id,
api_version=api_version,
secret_key=secret_key,
)
# 1. Ping the gateway
ping_result = client.ping()
if not ping_result["success"]:
print(f"Gateway unreachable: {ping_result['error']['message']}")
return
print("Gateway is up")
# 2. Check balance
balance_result = client.get_balance()
if balance_result["success"]:
for b in balance_result["result"]["balance"]["amounts"]:
print(f" {b['currency']}: {b['value']}")
else:
print(f"Balance check failed: {balance_result['error']['message']}")
# 3. Create a deposit
order_id = str(int(time.time()))
deposit_result = client.create_deposit(
service_id=14701,
amount=10000,
currency="INR",
c_id=order_id,
description=f"Order #{order_id}",
payer={
"email": "customer@example.com",
"person": {"first_name": "John", "last_name": "Doe"},
},
)
if deposit_result["success"]:
print(f"Deposit created (request_id={deposit_result['request_id']})")
else:
print(f"Deposit failed: {deposit_result['error']['message']}")
return
# 4. Poll for status
for _ in range(5):
time.sleep(3)
status_result = client.get_status(c_id=order_id)
if status_result["success"]:
status = status_result["result"]["payment"]["status"]["status"]
print(f"Payment status: {status}")
if status in ("success", "error", "canceled", "declined", "refunded", "partially_refunded"):
break
else:
print(f"Status check failed: {status_result['error']['message']}")
break
if __name__ == "__main__":
main()
