Приймання криптоплатежів у FastAPI

Покроково

1. 1.Отримайте ключ API і секрет webhook

   У кабінеті Moonvex відкрийте «Інтеграція» → «Створити ключ» → тип «Секретний» (sk_live_…). Ключ показується один раз — збережіть його на сервері в MOONVEX_SECRET_KEY. Там само додайте webhook на https-адресу вашого сервера і збережіть секрет підпису в MOONVEX_WEBHOOK_SECRET.

   pip install fastapi uvicorn httpx asyncpg
   
   export MOONVEX_SECRET_KEY=sk_live_...
   export MOONVEX_WEBHOOK_SECRET=whsec_...
   export DATABASE_URL=postgresql://user:pass@localhost:5432/shop

2. 2.Створіть платіж

   Клієнт httpx створюється один раз у lifespan з базовою адресою і заголовком Authorization. Для кожного платежу передається Idempotency-Key, а сума — str(Decimal) з numeric-колонки.

   # main.py
   import os
   from contextlib import asynccontextmanager
   
   import asyncpg
   import httpx
   from fastapi import FastAPI, HTTPException, Request
   from fastapi.responses import RedirectResponse
   
   
   @asynccontextmanager
   async def lifespan(app: FastAPI):
       app.state.db = await asyncpg.create_pool(os.environ["DATABASE_URL"])
       app.state.moonvex = httpx.AsyncClient(
           base_url="https://api.moonvex.org",
           headers={"Authorization": "Bearer " + os.environ["MOONVEX_SECRET_KEY"]},
           timeout=15,
       )
       yield
       await app.state.moonvex.aclose()
       await app.state.db.close()
   
   
   app = FastAPI(lifespan=lifespan)
   
   
   @app.post("/orders/{order_id}/pay")
   async def pay(order_id: int, request: Request):
       order = await request.app.state.db.fetchrow("SELECT id, total FROM orders WHERE id = $1", order_id)
       if order is None:
           raise HTTPException(404)
       r = await request.app.state.moonvex.post(
           "/v1/payments",
           headers={"Idempotency-Key": f"order-{order_id}-1"},
           # numeric приходить як Decimal → str(): "25.00"
           json={"orderId": str(order_id), "amount": str(order["total"]), "currency": "USDT", "network": "TRON"},
       )
       payment = r.json()
       if r.status_code >= 400:
           raise HTTPException(502, payment["error"]["code"])
       return RedirectResponse(payment["paymentUrl"], status_code=303)

3. 3.Надішліть клієнта на paymentUrl

   paymentUrl — готова сторінка оплати: точна сума з комісією, адреса, QR-код, офіційний контракт токена й таймер, статус оновлюється сам. Відкривайте її редиректом або в новій вкладці — вбудувати у фрейм не можна. Повернення клієнта на ваш сайт ще не оплата: замовлення позначається оплаченим лише за webhook або статусом із GET /v1/payments/:id. RedirectResponse(..., status_code=303) після POST відкриє сторінку оплати.

4. 4.Прийміть webhook і перевірте підпис

   Moonvex надсилає POST із заголовком Moonvex-Signature: t=<unix-час>,v1=<підпис>. Підпис — HMAC-SHA256 секретом webhook від рядка «t.сире_тіло» у hex. Рахуйте його за байтами тіла до розбору JSON, порівнюйте за сталий час, відхиляйте події, старші за 5 хвилин. Після зміни секрету 24 години в заголовку два v1 — підходить будь-який. У FastAPI не оголошуйте Pydantic-модель для тіла webhook: візьміть Request і await request.body() — це початкові байти.

   # main.py (продовження)
   import hashlib
   import hmac
   import json
   import time
   
   
   def verify_moonvex(raw: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
       parts = [p.split("=", 1) for p in (header or "").split(",") if "=" in p]
       t = next((v for k, v in parts if k == "t"), None)
       if t is None or not t.isdigit() or abs(time.time() - int(t)) > tolerance:
           return False
       expected = hmac.new(secret.encode(), t.encode() + b"." + raw, hashlib.sha256).hexdigest()
       return any(hmac.compare_digest(expected, v) for k, v in parts if k == "v1")
   
   
   @app.post("/moonvex/webhook")
   async def moonvex_webhook(request: Request):
       # Request.body(), а не Pydantic-модель: потрібні початкові байти
       raw = await request.body()
       if not verify_moonvex(raw, request.headers.get("moonvex-signature", ""), os.environ["MOONVEX_WEBHOOK_SECRET"]):
           raise HTTPException(400, "bad signature")
       event = json.loads(raw)
       if event["type"] in ("payment.confirmed", "payment.overpaid"):
           await mark_paid_once(request.app.state.db, event["id"], event["data"]["orderId"])
       return {"received": True}

5. 5.Позначте замовлення оплаченим рівно один раз

   Під час повторної доставки id події той самий, тому запишіть його в таблицю з унікальним ключем в одній транзакції з оновленням замовлення: дубль нічого не змінить. Видавайте товар за payment.confirmed і payment.overpaid, а не за payment.detected. В asyncpg: INSERT … ON CONFLICT DO NOTHING RETURNING усередині conn.transaction(); None означає повторну доставку.

   # main.py (продовження)
   # CREATE TABLE processed_events (event_id text PRIMARY KEY, created_at timestamptz NOT NULL DEFAULT now());
   
   
   async def mark_paid_once(db: asyncpg.Pool, event_id: str, order_id: str) -> None:
       async with db.acquire() as conn, conn.transaction():
           fresh = await conn.fetchval(
               "INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING RETURNING event_id",
               event_id,
           )
           if fresh is None:
               return  # подію вже оброблено
           await conn.execute("UPDATE orders SET status = 'paid' WHERE id = $1 AND status <> 'paid'", int(order_id))

6. 6.Перевірте інтеграцію

   Тестової мережі в Moonvex немає — усе працює в основних мережах. Кнопка «Надіслати тест» у розділі «Інтеграція» надішле подію webhook.test — так перевіряється підпис. Потім оплатіть невеликий рахунок, наприклад 1 USDT у TRON: це звичайний платіж, гроші надійдуть на баланс. Журнал доставок із кодами відповідей і кнопкою «Повторити» — там само.

Підводні камені

• Pydantic-модель у параметрі ендпоінта розбирає й валідує тіло раніше за вас — підпис за нею не перевірити. Потрібен сирий Request.
• Створюйте httpx.AsyncClient один раз (lifespan), а не на кожен запит: так повторно використовуються з’єднання.
• Не використовуйте синхронний requests усередині async-ендпоінтів — він блокує цикл подій.
• Для довгих задач після webhook підійдуть BackgroundTasks або черга — відповідь має піти за 10 секунд.