This dependency-free example follows the same authenticated preflight, single-task creation, task polling, result, and payment-status flow.
import json
import os
import time
from urllib.error import HTTPError
from urllib.parse import quote, urlencode
from urllib.request import Request, urlopen
BASE = os.environ.get("BASE_URL", "").rstrip("/")
CDK = os.environ.get("CDK", "")
METHOD = os.environ.get("PAYMENT_METHOD", "gcash").lower()
TOKEN = os.environ.get("ACCESS_TOKEN", "")
METHODS = {
"upi": {"country": "IN", "currency": "INR"},
"ideal": {"country": "NL", "currency": "EUR"},
"kakao": {"country": "KR", "currency": "KRW"},
"gcash": {"country": "PH", "currency": "PHP"},
}
def describe(body):
detail = body.get("detail", body.get("error")) if isinstance(body, dict) else None
if isinstance(detail, list):
return "; ".join(
f"{'.'.join(map(str, item.get('loc', [])))}: {item.get('msg', item)}"
for item in detail
)
if isinstance(detail, str):
return detail
if detail:
return json.dumps(detail, ensure_ascii=False)
return body.get("message", "unknown_error") if isinstance(body, dict) else "unknown_error"
def api(path, method="GET", payload=None):
data = json.dumps(payload).encode() if payload is not None else None
request = Request(
BASE + path,
data=data,
method=method,
headers={"Accept": "application/json", "Content-Type": "application/json"},
)
try:
with urlopen(request, timeout=300) as response:
raw = response.read().decode()
return json.loads(raw) if raw else {}
except HTTPError as error:
raw = error.read().decode(errors="replace")
try:
body = json.loads(raw) if raw else {}
except json.JSONDecodeError:
body = {"detail": raw}
retry_after = error.headers.get("Retry-After")
suffix = f" (retry after {retry_after}s)" if retry_after else ""
raise RuntimeError(f"HTTP {error.code}: {describe(body)}{suffix}") from None
def preflight():
if METHOD == "gcash":
path = "/api/paylinks/eligibility/gcash"
payload = {"cdk": CDK, "access_token": TOKEN}
else:
path = "/api/paylinks/eligibility/check"
payload = {"cdk": CDK, "access_token": TOKEN, "payment_method": METHOD}
result = api(path, "POST", payload)
outcome = str(result.get("outcome") or "").strip().lower()
if outcome == "disabled":
if METHOD != "gcash":
return ""
raise RuntimeError("GCash eligibility check is disabled; it cannot be skipped")
checks = {
"valid": result.get("valid") is True,
"payment_method_available": result.get("payment_method_available") is True,
"checkout_amount_is_zero": result.get("checkout_amount_is_zero") is True,
"eligible": result.get("eligible") is True,
"outcome": outcome == "eligible",
}
failed = [name for name, passed in checks.items() if not passed]
if failed:
reason = result.get("reason") or outcome or "unknown"
raise RuntimeError(
f"Not eligible (failed checks: {', '.join(failed)}): "
f"{reason} - {result.get('message', '')}"
)
proof = str(result.get("eligibility_proof") or "").strip()
try:
proof_expires_at = float(result.get("proof_expires_at") or 0)
except (TypeError, ValueError):
proof_expires_at = 0
if not proof or proof_expires_at <= time.time():
raise RuntimeError("Eligibility proof is missing or expired")
return proof
def payment_status(task_id):
encoded = quote(task_id, safe="")
while True:
state = api(f"/api/paylinks/tasks/{encoded}/payment-status")
print(f"payment_state={state.get('payment_state', 'unknown')} paid={bool(state.get('paid'))}")
if state.get("terminal") or not state.get("retryable"):
return state
time.sleep(max(1, int(state.get("retry_after_seconds") or 30)))
def main():
if not BASE.startswith("https://") or not CDK or not TOKEN or METHOD not in METHODS:
raise RuntimeError("Set HTTPS BASE_URL, CDK, ACCESS_TOKEN, and a supported PAYMENT_METHOD")
verified = api(
"/api/card-key/verify",
"POST",
{"cdk": CDK, "payment_method": METHOD},
)
if not verified.get("ok") or METHOD not in verified.get("allowed_payment_methods", []):
raise RuntimeError(f"CDK rejected: {verified.get('code', 'method_not_allowed')}")
summary = api("/api/paylinks/tasks/summary?" + urlencode({"cdk": CDK}))
if int(summary.get("max_submit_now") or 0) < 1:
raise RuntimeError("CDK has no capacity now")
proof = preflight()
created = api(
"/api/paylinks/tasks",
"POST",
{
"payment_method": METHOD,
"cdk": CDK,
"access_token": TOKEN,
"country": METHODS[METHOD]["country"],
"currency": METHODS[METHOD]["currency"],
"eligibility_proof": proof,
},
)
task_id = created.get("task_id")
if not task_id:
raise RuntimeError("Create response did not include task_id")
task = None
while task is None:
query = urlencode({"task_ids": task_id})
snapshot = api("/api/paylinks/tasks/statuses?" + query)
if task_id in snapshot.get("missing", []):
raise RuntimeError("Task is missing")
current = next(
(item for item in snapshot.get("items", []) if item.get("task_id") == task_id),
None,
)
if current is None or current.get("status") in {"queued", "running"}:
time.sleep(3)
continue
task = current
if task.get("status") != "succeeded":
failure = task.get("failure") or {}
raise RuntimeError(
f"Task {task.get('status')}: {failure.get('code', '')} {failure.get('summary', '')}"
)
print("payment_url=" + str((task.get("result") or {}).get("payment_url", "")))
if METHOD != "ideal":
payment_status(task_id)
if __name__ == "__main__":
main()
