greebo/youkassa-mock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: YooKassa mock v2

Browse Source
This commit is contained in:
adminko
2026-03-12 06:54:38 +00:00
commit c7db0a6dd1
7 changed files with 996 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
.gitignore vendored Normal file
View File

@@ -0,0 +1,21 @@
__pycache__/
*.py[cod]
*.pyo
*.pyd
*.so
.venv/
venv/
env/
.env
.env.*
.pytest_cache/
.mypy_cache/
.ruff_cache/
.coverage
htmlcov/
dist/
build/
*.egg-info/
.idea/
.vscode/
.DS_Store

16
Dockerfile Normal file
View File

@@ -0,0 +1,16 @@
FROM python:3.11-slim
WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
ENV MOCK_PORT=8081
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
EXPOSE 8081
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8081"]

418
README.md Normal file
View File

@@ -0,0 +1,418 @@
# YooKassa Mock v2
Легковесный mock-сервер для эмуляции базового сценария работы YooKassa.
## Что умеет
- создавать платеж
- возвращать `confirmation_url`
- показывать checkout-страницу
- завершать платеж как:
- `succeeded`
- `canceled`
- отправлять webhook
- повторять webhook при ошибке
- поддерживать `Idempotence-Key`
- отдавать статус платежа по `payment_id`
---
# Файлы
- `main.py` — FastAPI приложение
- `requirements.txt` — зависимости
- `Dockerfile` — образ
- `docker-compose.yml` — запуск контейнера
- `.env` — переменные окружения
---
# Запуск
## 1. Сборка и старт
~~~bash
docker compose up -d --build
~~~
## 2. Проверка здоровья
~~~bash
curl -s http://127.0.0.1:${YMK_PUBLIC_PORT}/health
~~~
Пример ответа:
~~~json
{"status":"ok","payments_total":0,"idempotence_keys_total":0}
~~~
---
# Переменные окружения
Пример `.env`:
~~~env
YMK_PUBLIC_PORT=8083
WEBHOOK_URL=http://192.168.149.101:8000/api/tickets/webhook/yookassa
MOCK_HOST=192.168.149.101
MOCK_PORT=8083
MOCK_REQUIRE_AUTH=0
MOCK_SHOP_ID=test_shop
MOCK_SECRET_KEY=test_secret
WEBHOOK_RETRY_COUNT=3
WEBHOOK_RETRY_DELAY_SEC=1
WEBHOOK_DELAY_SEC=0
~~~
## Описание
### `YMK_PUBLIC_PORT`
Внешний порт хоста, на который публикуется контейнер.
### `WEBHOOK_URL`
URL для отправки webhook после смены статуса платежа.
### `MOCK_HOST`
Хост, который будет подставляться в `confirmation_url`.
### `MOCK_PORT`
Порт, который будет подставляться в `confirmation_url`.
### `MOCK_REQUIRE_AUTH`
- `0` — Basic Auth не обязателен
- `1` — Basic Auth обязателен
### `MOCK_SHOP_ID`
Логин для Basic Auth, если auth включен.
### `MOCK_SECRET_KEY`
Пароль для Basic Auth, если auth включен.
### `WEBHOOK_RETRY_COUNT`
Количество попыток отправки webhook.
### `WEBHOOK_RETRY_DELAY_SEC`
Пауза между retry webhook.
### `WEBHOOK_DELAY_SEC`
Искусственная задержка перед первой отправкой webhook.
---
# API
## 1. POST `/v3/payments`
Создание платежа.
### Обязательное
- заголовок `Idempotence-Key`
- `confirmation.return_url`
- `amount.value`
### Пример запроса
~~~bash
curl -s -X POST http://127.0.0.1:8083/v3/payments \
-H 'Content-Type: application/json' \
-H 'Idempotence-Key: test-001' \
-d '{
"amount": {
"value": "100.00",
"currency": "RUB"
},
"confirmation": {
"type": "redirect",
"return_url": "http://192.168.149.101:3000/payment/return"
},
"capture": true,
"description": "Заказ №1"
}'
~~~
### Пример успешного ответа
~~~json
{
"id": "fdb332f8-bb1d-4def-bc1d-d5b00a7e287a",
"status": "pending",
"paid": false,
"amount": {
"value": "100.00",
"currency": "RUB"
},
"confirmation": {
"type": "redirect",
"confirmation_url": "http://192.168.149.101:8083/checkout?payment_id=fdb332f8-bb1d-4def-bc1d-d5b00a7e287a"
},
"created_at": "2026-03-12T06:35:47.512Z",
"description": "Заказ №1",
"metadata": {},
"capture": true,
"recipient": {
"account_id": "test_shop",
"gateway_id": "100700"
},
"refundable": false,
"test": true
}
~~~
### Идемпотентность
Если повторно вызвать `POST /v3/payments` с тем же `Idempotence-Key`, мок вернет тот же платеж, а не создаст новый.
---
## 2. GET `/v3/payments/{payment_id}`
Получение статуса платежа.
### Пример
~~~bash
curl -s http://127.0.0.1:8083/v3/payments/fdb332f8-bb1d-4def-bc1d-d5b00a7e287a
~~~
### Пример ответа
~~~json
{
"id": "fdb332f8-bb1d-4def-bc1d-d5b00a7e287a",
"status": "succeeded",
"paid": true,
"amount": {
"value": "100.00",
"currency": "RUB"
},
"confirmation": {
"type": "redirect",
"confirmation_url": "http://192.168.149.101:8083/checkout?payment_id=fdb332f8-bb1d-4def-bc1d-d5b00a7e287a"
},
"created_at": "2026-03-12T06:35:47.512Z",
"description": "Заказ №1",
"metadata": {},
"capture": true,
"recipient": {
"account_id": "test_shop",
"gateway_id": "100700"
},
"refundable": true,
"test": true
}
~~~
---
## 3. GET `/checkout?payment_id=<id>`
HTML-страница оплаты.
Что показывает:
- `payment_id`
- статус
- сценарий
- описание
- сумму
На странице есть три кнопки:
- `Оплатить успешно`
- `Отменить оплату`
- `Ошибка оплаты`
---
## 4. POST `/process/{payment_id}`
Меняет статус платежа, отправляет webhook, делает redirect на `return_url`.
### Входные варианты
Через form-data / x-www-form-urlencoded поле `action`:
- `success`
- `cancel`
- `fail`
### Пример success
~~~bash
curl -i -X POST http://127.0.0.1:8083/process/fdb332f8-bb1d-4def-bc1d-d5b00a7e287a \
-d 'action=success'
~~~
### Пример cancel
~~~bash
curl -i -X POST http://127.0.0.1:8083/process/294ec8cd-c1fb-44b4-91cc-04d669343390 \
-d 'action=cancel'
~~~
### Поведение
- `success` → статус `succeeded`, `paid=true`
- `cancel` → статус `canceled`, `paid=false`
- `fail` → статус `canceled`, `paid=false`
После этого сервис делает:
- webhook на `WEBHOOK_URL`
- `303 See Other` на `return_url`
---
## 5. GET `/health`
Проверка состояния сервиса.
### Пример
~~~bash
curl -s http://127.0.0.1:8083/health
~~~
### Ответ
~~~json
{
"status": "ok",
"payments_total": 2,
"idempotence_keys_total": 2
}
~~~
---
# Webhook
После `success`:
~~~json
{
"event": "payment.succeeded",
"type": "notification",
"object": {
"id": "payment_id",
"status": "succeeded",
"paid": true,
"amount": {
"value": "100.00",
"currency": "RUB"
},
"description": "Заказ №1",
"metadata": {}
}
}
~~~
После `cancel` / `fail`:
~~~json
{
"event": "payment.canceled",
"type": "notification",
"object": {
"id": "payment_id",
"status": "canceled",
"paid": false,
"amount": {
"value": "250.00",
"currency": "RUB"
},
"description": "Заказ №2",
"metadata": {
"mock_scenario": "cancel"
}
}
}
~~~
---
# Retry webhook
Если endpoint webhook отвечает ошибкой или недоступен:
- мок делает повторные попытки
- количество попыток задается `WEBHOOK_RETRY_COUNT`
- задержка между попытками задается `WEBHOOK_RETRY_DELAY_SEC`
Логи смотреть так:
~~~bash
docker logs --tail 100 yookassa-mock
~~~
---
# Сценарии через metadata
При создании платежа можно задать сценарий по умолчанию:
~~~json
"metadata": {
"mock_scenario": "cancel"
}
~~~
Допустимые значения:
- `success`
- `cancel`
- `fail`
Если в `POST /process/{payment_id}` не передать `action`, будет использован `mock_scenario`.
---
# Basic Auth
Если включить:
~~~env
MOCK_REQUIRE_AUTH=1
~~~
Тогда endpoints `/v3/payments` и `/v3/payments/{payment_id}` будут требовать Basic Auth.
Логин/пароль берутся из:
- `MOCK_SHOP_ID`
- `MOCK_SECRET_KEY`
---
# Ограничения
Это мок, не полная копия YooKassa.
Сейчас нет:
- persistent storage
- списка платежей
- ручного cancel endpoint в стиле реального API
- полной схемы YooKassa
- реальной проверки секретов YooKassa
- capture/authorize flow с отдельным подтверждением списания
- рефандов
---
# Текущее состояние проверки
Проверено вручную:
- создание платежа — OK
- `confirmation_url` — OK
- `GET /v3/payments/{id}` — OK
- идемпотентность — OK
- `pending -> succeeded` — OK
- `pending -> canceled` — OK
- redirect на `return_url` — OK
- retry webhook — OK
Проблема сейчас только одна:
`WEBHOOK_URL` отвечает `404 Not Found`, значит endpoint на основном backend отсутствует или указан неверно.

19
docker-compose.yml Normal file
View File

@@ -0,0 +1,19 @@
services:
yookassa-mock:
build: .
container_name: yookassa-mock
env_file:
- .env
ports:
- "${YMK_PUBLIC_PORT:-8083}:8081"
environment:
WEBHOOK_URL: "${WEBHOOK_URL}"
MOCK_HOST: "${MOCK_HOST}"
MOCK_PORT: "${MOCK_PORT:-8081}"
MOCK_REQUIRE_AUTH: "${MOCK_REQUIRE_AUTH:-0}"
MOCK_SHOP_ID: "${MOCK_SHOP_ID:-}"
MOCK_SECRET_KEY: "${MOCK_SECRET_KEY:-}"
WEBHOOK_RETRY_COUNT: "${WEBHOOK_RETRY_COUNT:-3}"
WEBHOOK_RETRY_DELAY_SEC: "${WEBHOOK_RETRY_DELAY_SEC:-1}"
WEBHOOK_DELAY_SEC: "${WEBHOOK_DELAY_SEC:-0}"
restart: unless-stopped

340
main.py Normal file
View File

@@ -0,0 +1,340 @@
import asyncio
import base64
import os
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, Optional
from uuid import uuid4
import httpx
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from jinja2 import Template
app = FastAPI(title="YooKassa Mock v2")
payments_db: Dict[str, Dict[str, Any]] = {}
idempotence_db: Dict[str, Dict[str, Any]] = {}
WEBHOOK_URL = os.getenv("WEBHOOK_URL", "").strip()
MOCK_HOST = os.getenv("MOCK_HOST", "").strip()
MOCK_PORT = os.getenv("MOCK_PORT", "8081").strip()
MOCK_REQUIRE_AUTH = os.getenv("MOCK_REQUIRE_AUTH", "0").strip().lower() in {"1", "true", "yes", "on"}
MOCK_SHOP_ID = os.getenv("MOCK_SHOP_ID", "").strip()
MOCK_SECRET_KEY = os.getenv("MOCK_SECRET_KEY", "").strip()
WEBHOOK_RETRY_COUNT = max(1, int(os.getenv("WEBHOOK_RETRY_COUNT", "3")))
WEBHOOK_RETRY_DELAY_SEC = float(os.getenv("WEBHOOK_RETRY_DELAY_SEC", "1"))
WEBHOOK_DELAY_SEC = float(os.getenv("WEBHOOK_DELAY_SEC", "0"))
CHECKOUT_TEMPLATE = Template("""
<!doctype html>
<html lang="ru">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>YooKassa Mock Checkout</title>
<style>
body {
font-family: Arial, sans-serif;
background: #f6f7fb;
margin: 0;
padding: 40px 20px;
color: #222;
}
.card {
max-width: 560px;
margin: 0 auto;
background: #fff;
border-radius: 12px;
padding: 32px;
box-shadow: 0 8px 24px rgba(0,0,0,0.08);
}
h1 { margin-top: 0; font-size: 24px; }
.amount { font-size: 28px; font-weight: 700; margin: 16px 0 24px; }
.row { margin: 8px 0; color: #555; }
.actions {
display: grid;
gap: 12px;
margin-top: 24px;
}
button {
width: 100%;
border: 0;
border-radius: 10px;
padding: 14px 18px;
font-size: 16px;
font-weight: 700;
cursor: pointer;
background: #006efc;
color: white;
}
button.secondary { background: #6b7280; }
button.danger { background: #b91c1c; }
code {
background: #f3f4f6;
padding: 2px 6px;
border-radius: 6px;
}
</style>
</head>
<body>
<div class="card">
<h1>Тестовая оплата</h1>
<div class="row">Платеж: <code>{{ payment_id }}</code></div>
<div class="row">Статус: <code>{{ status }}</code></div>
<div class="row">Сценарий по умолчанию: <code>{{ scenario }}</code></div>
<div class="row">Описание: <code>{{ description }}</code></div>
<div>Сумма к оплате:</div>
<div class="amount">{{ amount }} {{ currency }}</div>
<div class="actions">
<form method="post" action="/process/{{ payment_id }}">
<input type="hidden" name="action" value="success">
<button type="submit">Оплатить успешно</button>
</form>
<form method="post" action="/process/{{ payment_id }}">
<input type="hidden" name="action" value="cancel">
<button class="secondary" type="submit">Отменить оплату</button>
</form>
<form method="post" action="/process/{{ payment_id }}">
<input type="hidden" name="action" value="fail">
<button class="danger" type="submit">Ошибка оплаты</button>
</form>
</div>
</div>
</body>
</html>
""")
def utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
def parse_amount_value(value: Any) -> str:
try:
normalized = Decimal(str(value)).quantize(Decimal("0.01"))
except (InvalidOperation, ValueError):
raise HTTPException(status_code=400, detail="invalid amount.value")
if normalized <= 0:
raise HTTPException(status_code=400, detail="amount.value must be greater than 0")
return f"{normalized:.2f}"
def build_base_url(request: Request) -> str:
if MOCK_HOST:
return f"http://{MOCK_HOST}:{MOCK_PORT}"
return str(request.base_url).rstrip("/")
def build_payment_response(payment: Dict[str, Any], request: Request) -> Dict[str, Any]:
base_url = build_base_url(request)
payment_id = payment["id"]
return {
"id": payment_id,
"status": payment["status"],
"paid": payment["paid"],
"amount": {
"value": payment["amount"]["value"],
"currency": payment["amount"]["currency"],
},
"confirmation": {
"type": "redirect",
"confirmation_url": f"{base_url}/checkout?payment_id={payment_id}",
},
"created_at": payment["created_at"],
"description": payment["description"],
"metadata": payment["metadata"],
"capture": payment["capture"],
"recipient": {
"account_id": payment["recipient"]["account_id"],
"gateway_id": payment["recipient"]["gateway_id"],
},
"refundable": payment["status"] == "succeeded" and bool(payment["capture"]),
"test": True,
}
def require_basic_auth(authorization: Optional[str]) -> None:
if not MOCK_REQUIRE_AUTH:
return
if not authorization or not authorization.startswith("Basic "):
raise HTTPException(status_code=401, detail="basic authorization required")
try:
encoded = authorization.split(" ", 1)[1]
decoded = base64.b64decode(encoded).decode("utf-8")
username, password = decoded.split(":", 1)
except Exception:
raise HTTPException(status_code=401, detail="invalid basic authorization")
if username != MOCK_SHOP_ID or password != MOCK_SECRET_KEY:
raise HTTPException(status_code=401, detail="invalid shop credentials")
def resolve_event_name(status: str) -> str:
if status == "succeeded":
return "payment.succeeded"
return "payment.canceled"
async def send_webhook(payment: Dict[str, Any]) -> None:
if not WEBHOOK_URL:
return
if WEBHOOK_DELAY_SEC > 0:
await asyncio.sleep(WEBHOOK_DELAY_SEC)
payload = {
"event": resolve_event_name(payment["status"]),
"type": "notification",
"object": {
"id": payment["id"],
"status": payment["status"],
"paid": payment["paid"],
"amount": payment["amount"],
"description": payment["description"],
"metadata": payment["metadata"],
},
}
last_error = None
for attempt in range(1, WEBHOOK_RETRY_COUNT + 1):
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(WEBHOOK_URL, json=payload)
response.raise_for_status()
print(f"Webhook delivered: payment_id={payment['id']} attempt={attempt}")
return
except Exception as exc:
last_error = exc
print(f"Webhook attempt failed: payment_id={payment['id']} attempt={attempt} error={exc}")
if attempt < WEBHOOK_RETRY_COUNT:
await asyncio.sleep(WEBHOOK_RETRY_DELAY_SEC)
print(f"Webhook delivery failed: payment_id={payment['id']} error={last_error}")
@app.post("/v3/payments")
async def create_payment(
request: Request,
authorization: Optional[str] = Header(default=None),
idempotence_key: Optional[str] = Header(default=None, alias="Idempotence-Key"),
):
require_basic_auth(authorization)
if not idempotence_key:
raise HTTPException(status_code=400, detail="Idempotence-Key header is required")
if idempotence_key in idempotence_db:
existing_payment_id = idempotence_db[idempotence_key]["payment_id"]
payment = payments_db[existing_payment_id]
return JSONResponse(content=build_payment_response(payment, request))
payload = await request.json()
amount_data = payload.get("amount") or {}
confirmation_data = payload.get("confirmation") or {}
metadata = payload.get("metadata") or {}
return_url = confirmation_data.get("return_url")
if not return_url:
raise HTTPException(status_code=400, detail="confirmation.return_url is required")
amount_value = parse_amount_value(amount_data.get("value"))
currency = amount_data.get("currency", "RUB")
capture = bool(payload.get("capture", False))
description = str(payload.get("description") or "")[:128]
scenario = str(metadata.get("mock_scenario", "success")).strip().lower()
if scenario not in {"success", "cancel", "fail"}:
raise HTTPException(status_code=400, detail="metadata.mock_scenario must be success, cancel or fail")
payment_id = str(uuid4())
payment = {
"id": payment_id,
"status": "pending",
"paid": False,
"created_at": utc_now_iso(),
"description": description,
"metadata": metadata,
"capture": capture,
"return_url": return_url,
"scenario": scenario,
"amount": {
"value": amount_value,
"currency": currency,
},
"recipient": {
"account_id": MOCK_SHOP_ID or "100500",
"gateway_id": "100700",
},
}
payments_db[payment_id] = payment
idempotence_db[idempotence_key] = {"payment_id": payment_id}
return JSONResponse(content=build_payment_response(payment, request))
@app.get("/v3/payments/{payment_id}")
async def get_payment(payment_id: str, request: Request, authorization: Optional[str] = Header(default=None)):
require_basic_auth(authorization)
payment = payments_db.get(payment_id)
if not payment:
raise HTTPException(status_code=404, detail="payment not found")
return JSONResponse(content=build_payment_response(payment, request))
@app.get("/checkout", response_class=HTMLResponse)
async def checkout(payment_id: str):
payment = payments_db.get(payment_id)
if not payment:
raise HTTPException(status_code=404, detail="payment not found")
html = CHECKOUT_TEMPLATE.render(
payment_id=payment["id"],
status=payment["status"],
scenario=payment["scenario"],
description=payment["description"] or "-",
amount=payment["amount"]["value"],
currency=payment["amount"]["currency"],
)
return HTMLResponse(content=html)
@app.post("/process/{payment_id}")
async def process_payment(payment_id: str, request: Request):
payment = payments_db.get(payment_id)
if not payment:
raise HTTPException(status_code=404, detail="payment not found")
form = await request.form()
action = str(form.get("action") or payment["scenario"]).strip().lower()
if action == "success":
payment["status"] = "succeeded"
payment["paid"] = True
elif action in {"cancel", "fail"}:
payment["status"] = "canceled"
payment["paid"] = False
else:
raise HTTPException(status_code=400, detail="unsupported action")
await send_webhook(payment)
return RedirectResponse(url=payment["return_url"], status_code=303)
@app.get("/health")
async def health():
return {
"status": "ok",
"payments_total": len(payments_db),
"idempotence_keys_total": len(idempotence_db),
}

177
main.py.bak Normal file
View File

@@ -0,0 +1,177 @@
import os
from uuid import uuid4
from typing import Dict, Any
import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from jinja2 import Template
app = FastAPI(title="YooKassa Mock")
payments_db: Dict[str, Dict[str, Any]] = {}
WEBHOOK_URL = os.getenv("WEBHOOK_URL", "").strip()
MOCK_HOST = os.getenv("MOCK_HOST", "").strip()
MOCK_PORT = int(os.getenv("MOCK_PORT", "8081"))
CHECKOUT_TEMPLATE = Template("""
<!doctype html>
<html lang="ru">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Оплата заказа</title>
<style>
body {
font-family: Arial, sans-serif;
background: #f6f7fb;
margin: 0;
padding: 40px 20px;
color: #222;
}
.card {
max-width: 480px;
margin: 0 auto;
background: #fff;
border-radius: 12px;
padding: 32px;
box-shadow: 0 8px 24px rgba(0,0,0,0.08);
}
h1 {
margin-top: 0;
font-size: 24px;
}
.amount {
font-size: 28px;
font-weight: 700;
margin: 16px 0 24px;
}
button {
width: 100%;
border: 0;
border-radius: 10px;
padding: 14px 18px;
font-size: 16px;
font-weight: 700;
cursor: pointer;
background: #006efc;
color: white;
}
.meta {
margin-top: 20px;
font-size: 13px;
color: #666;
word-break: break-all;
}
</style>
</head>
<body>
<div class="card">
<h1>Тестовая оплата</h1>
<div>Сумма к оплате:</div>
<div class="amount">{{ amount }} {{ currency }}</div>
<form method="post" action="/process/{{ payment_id }}">
<button type="submit">Оплатить</button>
</form>
<div class="meta">
payment_id: {{ payment_id }}
</div>
</div>
</body>
</html>
""")
def build_base_url(request: Request) -> str:
if MOCK_HOST:
return f"http://{MOCK_HOST}:{MOCK_PORT}"
return str(request.base_url).rstrip("/")
@app.post("/v3/payments")
async def create_payment(request: Request):
payload = await request.json()
payment_id = str(uuid4())
amount_data = payload.get("amount") or {}
confirmation_data = payload.get("confirmation") or {}
value = amount_data.get("value", "0.00")
currency = amount_data.get("currency", "RUB")
return_url = confirmation_data.get("return_url")
if not return_url:
raise HTTPException(status_code=400, detail="confirmation.return_url is required")
payments_db[payment_id] = {
"return_url": return_url,
"amount": value,
"currency": currency,
}
base_url = build_base_url(request)
confirmation_url = f"{base_url}/checkout?payment_id={payment_id}"
response = {
"id": payment_id,
"status": "pending",
"paid": False,
"amount": {
"value": value,
"currency": currency,
},
"confirmation": {
"type": "redirect",
"confirmation_url": confirmation_url,
},
"test": True,
}
return JSONResponse(content=response)
@app.get("/checkout", response_class=HTMLResponse)
async def checkout(payment_id: str):
payment = payments_db.get(payment_id)
if not payment:
raise HTTPException(status_code=404, detail="payment not found")
html = CHECKOUT_TEMPLATE.render(
payment_id=payment_id,
amount=payment["amount"],
currency=payment["currency"],
)
return HTMLResponse(content=html)
@app.post("/process/{payment_id}")
async def process_payment(payment_id: str):
payment = payments_db.get(payment_id)
if not payment:
raise HTTPException(status_code=404, detail="payment not found")
if WEBHOOK_URL:
webhook_payload = {
"event": "payment.succeeded",
"type": "notification",
"object": {
"id": payment_id,
"status": "succeeded",
},
}
try:
async with httpx.AsyncClient(timeout=10.0) as client:
await client.post(WEBHOOK_URL, json=webhook_payload)
except Exception as e:
print(f"Webhook send failed: {e}")
return RedirectResponse(url=payment["return_url"], status_code=303)
@app.get("/health")
async def health():
return {"status": "ok"}

5
requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
fastapi==0.115.8
uvicorn[standard]==0.34.0
httpx==0.28.1
jinja2==3.1.5
python-multipart==0.0.20