Python SDK

python -m pip install "waffo==0.3.0b0"
# 或安装最新预发布版本
python -m pip install --pre waffo

from waffo import Environment, Waffo, WaffoConfig

config = WaffoConfig(
    api_key="your-api-key",
    private_key="your-base64-encoded-private-key",
    waffo_public_key="waffo-public-key",
    merchant_id="your-merchant-id",
    environment=Environment.SANDBOX,
)

waffo = Waffo(config)

from uuid import uuid4

payment_request_id = uuid4().hex

response = waffo.order().create(
    {
        "paymentRequestId": payment_request_id,
        "merchantOrderId": f"ORDER_{payment_request_id}",
        "orderCurrency": "HKD",
        "orderAmount": "100.00",
        "orderDescription": "Test Product",
        "notifyUrl": "https://your-site.com/webhook",
        "userInfo": {
            "userId": "user_123",
            "userEmail": "user@example.com",
            "userTerminal": "WEB",
        },
        "paymentInfo": {"productName": "ONE_TIME_PAYMENT"},
        "goodsInfo": {"goodsUrl": "https://your-site.com/product/001"},
    }
)

if response.is_success():
    data = response.get_data()
    print("Redirect action:", data.order_action if data else None)
    print("Acquiring Order ID:", data.acquiring_order_id if data else None)

response = waffo.order().inquiry({"paymentRequestId": payment_request_id})

response = waffo.subscription().create({ ... })
inquiry = waffo.subscription().inquiry({"subscriptionId": "..."})
waffo.subscription().cancel({"subscriptionId": "..."})

response = waffo.order().refund(
    {
        "refundRequestId": "ref_...",
        "acquiringOrderId": "ACQ...",
        "refundAmount": "50.00",
        "refundReason": "Customer requested",
    }
)

inquiry = waffo.refund().inquiry({"refundRequestId": "ref_..."})

from fastapi import FastAPI, Header, Request, Response
from waffo import Waffo

app = FastAPI()
waffo = Waffo.from_env()

waffo.webhook().on_payment(
    lambda notification: print(notification.get("result", {}).get("acquiringOrderId"))
)

@app.post("/webhook")
async def webhook(request: Request, x_signature: str | None = Header(default=None)) -> Response:
    body = (await request.body()).decode("utf-8")
    result = waffo.webhook().handle_webhook(body, x_signature)

    return Response(
        content=result.response_body,
        media_type="application/json",
        headers={"X-SIGNATURE": result.response_signature},
        status_code=200 if result.success else 400,
    )

from waffo import WaffoError, WaffoUnknownStatusError

try:
    response = waffo.order().create({ ... })
except WaffoUnknownStatusError:
    # 重要：支付可能已成功！使用查询接口确认
    inquiry = waffo.order().inquiry({"paymentRequestId": payment_request_id})
except WaffoError:
    # 客户端错误（配置、签名等），修复后可重试
    raise