Payment Docs
Примеры кода

Python

Минимальный сервер на FastAPI, который создаёт заказы и проверяет webhook-уведомления

Минимальный сервер на FastAPI с двумя маршрутами: один создаёт заказ и перенаправляет плательщика, другой принимает webhook-уведомления и проверяет заголовок X-Signature.

pip install fastapi uvicorn requests
main.py
import hashlib
import hmac
import os

import requests
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse, RedirectResponse

API_URL = "https://api.riopay.online/v1"
API_TOKEN = os.environ["API_TOKEN"]  # выдаётся вашим менеджером

app = FastAPI(title="Payments example")


# 1. Создаём заказ и перенаправляем плательщика на платёжную страницу
@app.post("/create-order")
def create_order(amount: float, order_id: str, user_id: str | None = None):
    payload = {
        "amount": str(amount),
        "externalId": order_id,
        "externalUserId": user_id,
        "purpose": f"Payment for order {order_id}",
        "successUrl": "https://your-site.com/payment-success",
        "failUrl": "https://your-site.com/payment-failed",
        "callbackUrl": "https://your-site.com/webhooks/payments",
    }
    headers = {"X-Api-Token": API_TOKEN, "Content-Type": "application/json"}

    response = requests.post(f"{API_URL}/orders", json=payload, headers=headers, timeout=30)
    if response.status_code != 200:
        raise HTTPException(status_code=response.status_code, detail=response.text)

    order = response.json()
    # Сохраните order["id"] рядом со своим order_id, чтобы потом сопоставлять webhook-уведомления
    return RedirectResponse(url=order["paymentLink"])


# 2. Принимаем webhook-уведомления
@app.post("/webhooks/payments")
async def webhook(request: Request):
    raw_body = await request.body()
    signature = request.headers.get("x-signature", "")

    expected = hmac.new(API_TOKEN.encode(), raw_body, hashlib.sha512).hexdigest()
    if not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=403, detail="Invalid signature")

    data = await request.json()
    status, external_id = data.get("status"), data.get("externalId")

    # Webhook-уведомления могут приходить повторно: обработка должна быть идемпотентной
    if status == "COMPLETED":
        print(f"Заказ {external_id} оплачен")
        # отметьте заказ как оплаченный в своей базе данных
    elif status in {"REFUND", "CHARGEBACK"}:
        print(f"Заказ {external_id} отменён: {status}")
        # деньги вернулись плательщику: отметьте заказ как возвращённый
    elif status in {"FAILED", "CANCELED", "EXPIRED", "BLOCKED"}:
        print(f"Заказ {external_id} завершён со статусом {status}")
        # позже может стать COMPLETED, если банк подтвердит платёж

    return JSONResponse(content={"message": "OK"})

Запуск:

uvicorn main:app --port 3000

Что адаптировать

  • Замените жёстко заданные URL на свои адреса успеха, ошибки и webhook.
  • Сохраняйте id из ответа на создание заказа вместе со своим externalId.
  • Перенесите обработку статусов в свою модель заказа; только COMPLETED означает, что деньги поступили.

На этой странице