Python SDK

Қысқаша

Python SDK — бір ғана модуль: qutpay.py. Python 3.8 және одан жоғары нұсқада жұмыс істейді, тек стандартты кітапхананы пайдаланады (urllib, hmac, hashlib) — requests те, басқа тәуелділік те керек емес.

Модульден үш нәрсе экспортталады:

Архив: https://api.qut.kz/downloads/qutpay-sdk-python.zip

Орнату

Архивті ашып, ішінде pip арқылы орнатыңыз:

unzip qutpay-sdk-python.zip -d qutpay-sdk
pip install ./qutpay-sdk

Немесе qutpay.py файлын жобаңызға көшіріп қойыңыз — сыртқы тәуелділігі болмағандықтан, ол солай да жұмыс істейді. Дистрибутив атауы — qutpay, импорттайтын модуль де qutpay.

Инициализация

import os
from qutpay import QutPay, QutPayError, verify_webhook

qp = QutPay(
    api_key=os.environ["QUTPAY_API_KEY"],   # qp_live_… немесе qp_test_…
    base_url="https://api.qut.kz",          # әдепкі мән
    timeout=20.0,                            # секунд
)

api_key бос болса ValueError шығады. Кілт орта айнымалысында тұруы керек — кодқа да, репозиторийге де жазбаңыз.

Счёт жасау

Барлық аргументтер snake_case түрінде беріледі, SDK оларды API күткен camelCase-ке өзі аударады.

inv = qp.create_invoice(
    amount=12500,
    kind="qr",                               # "qr" (әдепкі) немесе "phone"
    description="Тапсырыс №4471",
    external_order_id="4471",
    customer={"name": "Айгүл", "phone": "77011234567", "email": "a@b.kz"},
    success_url="https://site.kz/ok",
    fail_url="https://site.kz/fail",
    metadata={"branch": "almaty-abay"},
    idempotency_key="order-4471",            # Idempotency-Key тақырыбына кетеді
)

print(inv["id"], inv["status"], inv["payUrl"], inv["expiresAt"])

amount — бірінші позициялық аргумент, қалғаны тек атауымен беріледі. Жауап — қарапайым dict: id, status, payUrl, qrUrl, deepLink, qrImageUrl, expiresAt. Жауап өрістері camelCase күйінде қалады.

Қалған әдістер

fresh = qp.get_invoice(inv["id"], live=True)
if fresh["status"] == "paid":
    mark_order_paid("4471")

qp.refund_invoice(inv["id"], 5000, "Тауардың бір бөлігі қайтарылды")

SDK-да жазылым, bulk және форма-хук әдістері жоқ. Оларды тікелей HTTP арқылы шақырыңыз:

import json, urllib.request

req = urllib.request.Request(
    "https://api.qut.kz/api/v1/invoices/bulk",
    data=json.dumps({"invoices": [{"amount": 1000, "externalOrderId": "a-1"}]}).encode(),
    headers={"X-API-Key": os.environ["QUTPAY_API_KEY"], "Content-Type": "application/json"},
    method="POST",
)
with urllib.request.urlopen(req) as resp:
    result = json.loads(resp.read())   # HTTP 207, әр элемент бойынша нәтиже

Webhook қолтаңбасын тексеру

verify_webhook(secret, raw_body, headers, tolerance_sec=300)  # → True / False

raw_body — өзгертілмеген байттар немесе жол. request.json арқылы алынған объект жарамайды: қолтаңба дәл сол байттар бойынша есептеледі. Функция X-Webhook-Signature пен X-Webhook-Timestamp тақырыптарын регистрге қарамай оқиды, sha256= префиксін тексереді, 5 минуттан ескі жеткізілімді қабылдамайды және hmac.compare_digest арқылы салыстырады.

Flask мысалы

import json, os
from flask import Flask, request, abort, redirect
from qutpay import QutPay, QutPayError, verify_webhook

app = Flask(__name__)
qp = QutPay(api_key=os.environ["QUTPAY_API_KEY"])
SECRET = os.environ["QUTPAY_WEBHOOK_SECRET"]


@app.post("/buy")
def buy():
    order = create_order(request.form)
    try:
        inv = qp.create_invoice(
            amount=order.total,
            description=f"Тапсырыс №{order.id}",
            external_order_id=str(order.id),
            success_url=f"https://site.kz/orders/{order.id}",
            idempotency_key=f"order-{order.id}",
        )
    except QutPayError as e:
        app.logger.error("qutpay %s %s %s", e.status, e.code, e)
        abort(502)
    save_invoice_id(order.id, inv["id"])
    return redirect(inv["payUrl"], code=303)


@app.post("/qutpay-webhook")
def hook():
    raw = request.get_data()                 # bytes, өзгертілмеген
    if not verify_webhook(SECRET, raw, request.headers):
        abort(401)

    e = json.loads(raw)
    invoice = e["invoice"]

    # Идемпотенттілік: бір оқиға екі рет келуі мүмкін
    if not already_handled(invoice["id"], invoice["status"]):
        if e["event"] == "invoice.paid":
            mark_order_paid(invoice["externalOrderId"], invoice)
        elif e["event"] == "invoice.refunded":
            mark_order_refunded(invoice["externalOrderId"])
        remember_handled(invoice["id"], invoice["status"])

    return "", 200                           # 2xx бермесеңіз 11 рет қайталанады

Django мысалы

Django-да дене request.body өрісінде тұрады, тақырыптар — request.headers. Маршрутты CSRF тексеруінен босату міндетті.

# views.py
import json, os
from django.http import HttpResponse, HttpResponseForbidden
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from qutpay import verify_webhook

SECRET = os.environ["QUTPAY_WEBHOOK_SECRET"]


@csrf_exempt
@require_POST
def qutpay_webhook(request):
    raw = request.body                       # bytes, өзгертілмеген
    if not verify_webhook(SECRET, raw, request.headers):
        return HttpResponseForbidden(status=401)

    e = json.loads(raw)
    invoice = e["invoice"]
    handle_event.delay(e["event"], invoice)  # ауыр жұмысты Celery-ге беріңіз
    return HttpResponse(status=200)

# urls.py
from django.urls import path
from .views import qutpay_webhook

urlpatterns = [path("qutpay-webhook", qutpay_webhook)]

ATOMIC_REQUESTS қосулы болса, ұзақ жұмысты транзакция ішінде жасамаңыз: webhook 8 секундтан ұзақ жауап берсе, жеткізілім сәтсіз деп есептеліп, қайталанады.

Қате өңдеу

Кез келген сәтсіз жауапта SDK QutPayError лақтырады:

try:
    inv = qp.create_invoice(amount=amount, external_order_id=order_id)
except QutPayError as e:
    if e.code == "kaspi_session_expired":
        notify_admin("Кассир байланысы үзілді")
    elif e.code in ("tariff_limit_reached", "tariff_daily_burst"):
        notify_admin("Лимит")                # қайталамаңыз
    elif e.code in ("request_rate_limited", "timeout", "network_error"):
        retry_later()                         # 1, 2, 4, 8 секунд бэкоффпен
    else:
        raise

Логиканы әрқашан .code мәніне қарап жазыңыз: мәтін өзгеруі мүмкін, код өзгермейді. Барлық кодтар: Қателер каталогы.

Жиі қойылатын сұрақтар

PyPI-дан орнатуға бола ма? Жоқ, пакет жария тізілімде емес. Архивті жүктеп, pip install ./qutpay-sdk жасаңыз немесе файлды көшіріп алыңыз.

async нұсқасы бар ма? Жоқ, клиент синхронды. FastAPI-де run_in_threadpool арқылы шақырыңыз немесе өз httpx клиентіңізді жазыңыз — verify_webhook функциясын бөлек пайдалана бересіз.

FastAPI-де raw body қалай аламын? raw = await request.body(), тақырыптар — request.headers.

Жауап өрістері неге snake_case емес? Кіріс аргументтер ыңғай үшін snake_case, ал жауап API-ден келген күйінде қайтады: payUrl, externalOrderId, expiresAt.

Сынауды қалай бастаған дұрыс? Sandbox кілтімен счёт жасап, qp.simulate_invoice(inv["id"], "paid") шақырыңыз — webhook шынайы жіберіледі: Sandbox-та төлемді симуляциялау.