Past raw HTTP calls.
The generated resource methods cover every endpoint, but a few things needed more than a flat method signature. This page covers all of them.
Pagination
Nomba's list endpoints are cursor-paginated server-side — they
return a cursor you feed back in for the next page.
paginate / apaginate drive that loop for
you; no second pagination scheme is introduced.
from nomba import Nomba, paginate
nomba = Nomba(...)
for account in paginate(nomba.accounts.list_all_accounts, limit=50):
print(account["accountRef"])
from nomba import AsyncNomba, apaginate
async with AsyncNomba(...) as nomba:
async for txn in apaginate(nomba.transactions.fetch_transactions_on_the_parent_account, limit=50):
print(txn)
Confirmed paginated endpoints: accounts.list_all_accounts, accounts.fetch_terminals_assigned_to_an_account, accounts.fetch_terminals_assigned_to_the_parent_account, virtual_accounts.filter_virtual_accounts, and all six transactions.* list/filter methods.
Guided card-payment flow
Card checkout is a multi-step sequence: submit card details, then
maybe an OTP, maybe a 3D Secure redirect, then confirm.
CardPaymentFlow wraps the sequence and decodes Nomba's
response codes into plain booleans, instead of you needing to know
what "T0" or "S0" mean.
order = nomba.checkout.create_an_online_checkout_order(
order={"orderReference": "order-001", "amount": "1000", ...}
)
flow = nomba.card_payment(order_reference="order-001")
step = flow.submit_card(card_details="...", key="")
if step.requires_otp:
step = flow.submit_otp("123456")
elif step.requires_3ds:
# redirect the user using step.secure_authentication_data
...
if step.completed:
result = flow.confirm()
step is a CardPaymentStep with .completed, .requires_otp, .requires_3ds, .transaction_id, .message. nomba.card_payment(...) returns an AsyncCardPaymentFlow off AsyncNomba too.
Webhook signature verification
Implements Nomba's documented HMAC-SHA256 scheme
(nomba-signature / nomba-timestamp
headers, base64-encoded signature) plus a replay-window check on the
timestamp — a valid signature alone doesn't prove a webhook wasn't
captured and resent later.
from nomba import verify_webhook_request, NombaValidationError
@app.post("/webhooks/nomba")
def handle_webhook():
try:
payload = verify_webhook_request(
signature_key="...",
body=request.get_data(),
headers=request.headers,
max_age_seconds=300, # default; reject webhooks older than 5 min
)
except NombaValidationError:
return "invalid signature", 401
if payload["event_type"] == "payment_success":
...
return "", 200
Pass max_age_seconds=None to disable the replay check if you have your own.
Local request validation
Flat fields like order_reference= are enforced by
Python itself. But nested objects — like order={...}
in checkout order creation — have their own required fields a flat
signature can't check. Every write call is validated against Nomba's
bundled OpenAPI spec before any network call.
from nomba import NombaValidationError
try:
nomba.checkout.create_an_online_checkout_order(order={"orderReference": "x"})
except NombaValidationError as e:
print(e)
# Missing required field(s) in request body: order.amount, order.currency, ...
Reliability: locking + retry/backoff
NombaClient / AsyncNombaClient guard token
fetching with a lock, so concurrent requests never race to re-fetch
a token — only one fetch happens, the rest reuse it. Requests that
hit a 429 or transient 5xx retry
automatically with exponential backoff, respecting
Retry-After when Nomba sends one.
nomba = Nomba(
client_id="...", client_secret="...", account_id="...",
max_retries=3, # default
backoff_factor=0.5, # default; delay ~= backoff_factor * 2^attempt
)
Bounded concurrency for fan-out calls
asyncio.gather has no built-in concurrency limit.
Firing off many calls at once can trigger a retry storm if several
start failing together, or just trip Nomba's rate limit by bursting
too many requests in one window. gather_limited runs
the same calls with a cap.
from nomba import AsyncNomba, gather_limited
async with AsyncNomba(...) as nomba:
calls = [
(lambda ref=ref: nomba.virtual_accounts.fetch_a_virtual_account(ref))
for ref in account_refs
]
results = await gather_limited(calls, limit=5)