Интеграция платежей на Python: пример и проверка

Интеграция платежей на Python занимает примерно сотню строк: один вызов POST /api/v1/payments через httpx и один эндпоинт, который принимает колбэк и проверяет HMAC. Сложность не в объёме кода, а в трёх местах, где Python ведёт себя не так, как ожидает разработчик: таймаут HTTP-клиента, тип суммы и момент, когда фреймворк уже успел распарсить тело запроса.

Разберём рабочий пример на FastAPI, а в конце — что менять, если у вас Flask.

HTTP-клиент: таймаут задаётся явно

requests без параметра timeout ждёт ответа буквально бесконечно — это поведение по умолчанию, и оно однажды подвесит вам весь пул воркеров. У httpx таймаут есть из коробки (5 секунд), но для платежей его лучше задать руками и переиспользовать один клиент, а не создавать соединение на каждый заказ.

import os
import uuid
from decimal import Decimal

import httpx

BASE = os.environ["ROLLYPAY_API_URL"]
API_KEY = os.environ["ROLLYPAY_API_KEY"]

client = httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0))


async def create_payment(order_id: str, amount: Decimal, title: str) -> dict:
    payload = {
        "amount": f"{amount:.2f}",
        "payment_currency": "RUB",
        "order_id": order_id,
        "description": title,
        "payment_method": "sbp",
        "metadata": {"source": "web"},
    }
    resp = await client.post(
        f"{BASE}/api/v1/payments",
        json=payload,
        headers={
            "X-API-Key": API_KEY,
            "X-Nonce": str(uuid.uuid4()),
        },
    )
    if resp.status_code >= 400:
        raise RuntimeError(f"{resp.status_code}: {resp.text}")
    return resp.json()

X-Nonce — новый uuid.uuid4() на каждый HTTP-вызов. Значение живёт 10 минут, повтор вернёт 401 nonce already used. Частая ошибка — вынести UUID в константу модуля: первый платёж пройдёт, второй упадёт, и связь между причиной и следствием найдётся не сразу.

Decimal вместо float: где исчезают копейки

0.1 + 0.2 в Python даёт 0.30000000000000004. На одном заказе это незаметно, на сверке за месяц — расхождение в несколько рублей и полдня разбирательств. Сумма должна быть Decimal от строки или целым числом копеек, и ни на одном шаге не превращаться в float.

from decimal import Decimal, ROUND_HALF_UP

price = Decimal("1490.00")
discount = price * Decimal("0.15")
total = (price - discount).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)

assert f"{total:.2f}" == "1266.50"   # уходит в API строкой

В PostgreSQL держите сумму в numeric(12, 2) или bigint в копейках. Тип double precision вернёт вам ту же ошибку округления с другой стороны.

Колбэк на FastAPI: сырое тело до Pydantic

Подпись считается от строки X-Timestamp + "." + неизменённые байты тела. Если объявить аргумент обработчика как Pydantic-модель, FastAPI распарсит JSON раньше вас, и восстановить исходные байты через json.dumps() не получится: порядок ключей и пробелы будут другими, HMAC не совпадёт. Берите await request.body() и разбирайте JSON вручную — уже после проверки.

import hashlib
import hmac
import time

from fastapi import BackgroundTasks, Request, Response

SECRET = os.environ["ROLLYPAY_SIGNING_SECRET"].encode()


@app.post("/webhooks/rollypay")
async def rollypay_callback(request: Request, tasks: BackgroundTasks):
    raw = await request.body()
    ts = request.headers.get("X-Timestamp", "")
    got = request.headers.get("X-Signature", "")

    expected = hmac.new(SECRET, ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, got):
        return Response(status_code=401)

    if not ts.isdigit() or abs(time.time() - int(ts)) > 300:
        return Response(status_code=401)

    event = json.loads(raw)
    stored = await store_event(event)      # уникальный индекс по payment_id
    if stored:
        tasks.add_task(grant_access, event["order_id"])
    return Response(status_code=200)

hmac.compare_digest обязателен вместо ==: обычное сравнение строк обрывается на первом несовпавшем символе, и по времени ответа подпись подбирается посимвольно. Подробный разбор — в тексте про проверку подписи webhook.

Фон против синхронной выдачи

Выдача доступа почти всегда медленнее, чем приём колбэка: письмо, приглашение в закрытый канал, генерация лицензии. Если делать это прямо в обработчике, вы упрётесь в таймаут отправителя и получите повтор события, пока первая обработка ещё идёт.

Работающая схема из двух шагов: синхронно сохраните событие в таблицу payment_events с уникальным индексом по payment_id, а долгую часть отдайте в BackgroundTasks, Celery или ARQ. Ответ 200 отдавайте после INSERT, а не после письма — тогда падение почтового сервиса не заставит платёжную систему слать колбэк по кругу.

Школа английского на 200 учеников приходила с обратной схемой: обработчик ждал ответа от CRM, CRM отвечала за 40 секунд, колбэк повторялся, и ученик получал три письма с доступом. Перенос CRM-вызова в фоновую задачу и уникальный индекс по payment_id закрыли обе проблемы за один вечер. Механику повторов разбирает статья об идемпотентности платёжных запросов.

Если у вас Flask, а не FastAPI

Логика та же, отличается только способ добраться до байтов. В Flask это request.get_data() — важно вызвать его без аргумента as_text=True, иначе вы получите строку с перекодировкой, а не исходные байты.

@app.post("/webhooks/rollypay")
def rollypay_callback():
    raw = request.get_data()                    # bytes, не str
    ts = request.headers.get("X-Timestamp", "")
    expected = hmac.new(SECRET, ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, request.headers.get("X-Signature", "")):
        return "", 401
    ...

Не обращайтесь к request.json до проверки: Flask закеширует разобранный объект, а вам всё равно нужны байты. И не ставьте перед маршрутом middleware, который читает поток запроса, — второй раз тело уже не прочитается.

Пять ошибок, которые видно в логах

Налоговая сторона остаётся на продавце: статус, чек и возвраты — ваша зона, платёжный сервис отвечает за техническое подключение. Самозанятый формирует чек в «Мой налог» и следит за лимитом 2,4 млн ₽ в год. Какие методы оплаты будут доступны вашему проекту, решает модерация после проверки категории. Если ещё выбираете формат подключения, посмотрите обзор API приёма платежей и раздел платежей для B2B-SaaS.