Жишээнүүд

FastAPI callback урсгал

Доорх жишээ нь invoice үүсгэж, callback endpoint дээрээ payment_check хийж төлбөрийн мэдээлэл авах энгийн урсгалыг харуулна.

examples/quickstart.py

import asyncio
from decimal import Decimal

from fastapi import FastAPI, status

from qpay_client.v2 import AsyncQPayClient, QPaySettings
from qpay_client.v2.enums import ObjectType
from qpay_client.v2.schemas import InvoiceCreateSimpleRequest, Offset, PaymentCheckRequest

# QPay client settings
settings = QPaySettings.sandbox()

# Init async client
client = AsyncQPayClient(settings=settings)

# init FastAPI app
app = FastAPI()

# Just a dummy db
payment_database = {}


async def create_invoice():
    response = await client.invoice_create(
        InvoiceCreateSimpleRequest(
            sender_invoice_no="1234567",
            invoice_receiver_code="terminal",
            invoice_description="test",
            amount=Decimal(1500),
            callback_url="https://api.your-domain.mn/payments?payment_id=1234567",
        )
    )

    # keep the qpay invoice_id in database, used for checking payment later!
    payment_database["1234567"] = {
        "id": "1234567",
        "invoice_id": response.invoice_id,
        "amount": Decimal(1500),
    }

    # Showing QPay invoice to the user ...
    print(response.qPay_shortUrl)


# You define the uri and query/param of your callback
# Your callback API must return
#   Response(status_code=200, body="SUCCESS")
@app.get("/payments", status_code=status.HTTP_200_OK)
async def qpay_callback(payment_id: str):
    data = payment_database.get(payment_id)
    if not data:
        raise ValueError("Payment not found")
    invoice_id = str(data["invoice_id"])
    response = await client.payment_check(
        PaymentCheckRequest(
            object_type=ObjectType.invoice, object_id=invoice_id, offset=Offset(page_number=1, page_limit=100)
        )
    )

    # do something with payment ...

    print(response)

    # This is important !
    return "SUCCESS"


if __name__ == "__main__":
    asyncio.run(create_invoice())

Энэ жишээнд:

sandbox тохиргоогоор AsyncQPayClient үүсгэнэ
invoice үүсгэсний дараа invoice_id-г өөрийн storage-д хадгална
callback ирэх үед payment_check ашиглан тухайн invoice-ийн төлбөрийг шалгана

Context manager ашиглах

Синхрон клиент:

from qpay_client.v2 import QPayClient, QPaySettings

settings = QPaySettings.sandbox()

with QPayClient(settings=settings) as client:
    invoice = client.invoice_get("your-invoice-id")
    print(invoice.invoice_status)

Асинхрон клиент:

from qpay_client.v2 import AsyncQPayClient, QPaySettings

settings = QPaySettings.sandbox()

async with AsyncQPayClient(settings=settings) as client:
    invoice = await client.invoice_get("your-invoice-id")
    print(invoice.invoice_status)

Алдаа боловсруулах

QPay талын алдаа гарвал QPayError exception шидэгдэнэ.

from qpay_client.v2 import QPayError

try:
    invoice = await client.invoice_get("missing-id")
except QPayError as exc:
    print(exc)

Retry тохируулах

Network retry болон payment_check polling-ийг тохиргоогоор удирдаж болно.

from qpay_client.v2 import QPaySettings

settings = QPaySettings.sandbox(
    client_retries=2,
    client_delay=0.25,
    client_jitter=0.1,
    payment_check_retries=8,
    payment_check_delay=0.5,
    payment_check_jitter=0.2,
)