Примеры кода
Python
Минимальный сервер на FastAPI, который создаёт заказы и проверяет webhook-уведомления
Минимальный сервер на FastAPI с двумя маршрутами: один создаёт заказ и перенаправляет плательщика, другой принимает webhook-уведомления и проверяет заголовок X-Signature.
pip install fastapi uvicorn requestsimport 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означает, что деньги поступили.