Всем привет! В этой статье разберём как подключить Payme — одну из самых популярных платёжных систем Узбекистана — к Telegram боту на Python. Для этого мы будем использовать библиотеку aiopayme — async-first решение с роутерами и dependency injection как в aiogram и FastAPI. В конце статьи вы получите полностью рабочую интеграцию: бот принимает команду /pay, пользователь оплачивает через Payme, бот получает уведомление об успешной оплате.
Установка
pip install aiopayme
Что такое aiopayme? aiopayme — это асинхронная Python библиотека для интеграции платёжной системы Payme. Главная особенность — роутерная архитектура, знакомая каждому кто работал с aiogram или FastAPI:
from aiopayme import Routerrouter = Router()@router.check_perform_transaction()async def check_perform(ctx: CheckPerformTransactionCtx, db: AsyncSession): ...
Единственная зависимость — httpx. Библиотека не навязывает ORM, фреймворк или структуру проекта.
Модели
Создадим две модели — заказ и транзакцию Payme.
# models/order.pyimport enumfrom sqlalchemy import Column, BigInteger, Integer, Stringfrom sqlalchemy.orm import DeclarativeBaseclass Base(DeclarativeBase): passclass OrderStatus(enum.Enum): PENDING = "pending" PAID = "paid" CANCELED = "canceled"class Order(Base): __tablename__ = "orders" id = Column(Integer, primary_key=True) telegram_id = Column(BigInteger, nullable=False) amount = Column(BigInteger, nullable=False) status = Column(String, default=OrderStatus.PENDING.value) payme_transaction_id = Column(String, nullable=True)
# models/payme.pyfrom sqlalchemy import Column, Integer, String, BigIntegerfrom models.order import Baseclass PaymeTransaction(Base): __tablename__ = "payme_transactions" id = Column(Integer, primary_key=True) payme_id = Column(String, unique=True, nullable=False) order_id = Column(Integer, nullable=True) state = Column(Integer, default=1) amount = Column(BigInteger, nullable=False) create_time = Column(BigInteger, nullable=False) perform_time = Column(BigInteger, default=0) cancel_time = Column(BigInteger, default=0) reason = Column(Integer, nullable=True)
PaymeService
Вынесем всю бизнес-логику в отдельный сервис. Это сделает хендлеры чистыми и логику легко тестируемой. Создаём services/payme.py:
# services/payme.pyfrom sqlalchemy import select, updatefrom sqlalchemy.ext.asyncio import AsyncSessionfrom aiopayme.exceptions import Errorsfrom aiopayme.utils import time_to_paymefrom aiopayme.types import ( CheckPerformTransactionCtx, CreateTransactionCtx, PerformTransactionCtx, CancelTransactionCtx, CheckTransactionCtx, GetStatementCtx,)from models import OrderStatus, Order, PaymeTransactionclass PaymeService: def __init__(self, db: AsyncSession): self.db = db async def get_order(self, order_id) -> Order | None: return await self.db.scalar( select(Order).where(Order.id == int(order_id)) ) async def get_transaction(self, payme_id: str) -> PaymeTransaction | None: return await self.db.scalar( select(PaymeTransaction).where(PaymeTransaction.payme_id == payme_id) ) async def get_active_transaction(self, order_id: int) -> PaymeTransaction | None: return await self.db.scalar( select(PaymeTransaction).where( PaymeTransaction.order_id == order_id, PaymeTransaction.state == 1, ) ) async def get_transactions(self, from_time: int, to_time: int): return (await self.db.scalars( select(PaymeTransaction).where( PaymeTransaction.create_time >= from_time, PaymeTransaction.create_time <= to_time, ) )).all() async def check_perform(self, ctx: CheckPerformTransactionCtx): order = await self.get_order(ctx.account.order_id) if not order: raise Errors.invalid_account() if order.amount * 100 != ctx.amount: raise Errors.invalid_amount() return ctx.ok(allow=True) async def create_transaction(self, ctx: CreateTransactionCtx): tx = await self.get_transaction(ctx.payme_id) order = await self.get_order(ctx.account.order_id) if not order: raise Errors.invalid_account() if order.amount * 100 != ctx.amount: raise Errors.invalid_amount() if tx: if tx.state == -1: raise Errors.unable_to_perform() return ctx.ok(transaction_id=tx.payme_id, create_time=tx.create_time) if order.status == OrderStatus.PAID: raise Errors.invalid_account() existing_tx = await self.get_active_transaction(order.id) if existing_tx: rejected = PaymeTransaction( payme_id=ctx.payme_id, order_id=order.id, amount=ctx.amount, create_time=ctx.time, state=-1, cancel_time=time_to_payme(), reason=3, ) self.db.add(rejected) await self.db.commit() raise Errors.unable_to_perform() tx = PaymeTransaction( payme_id=ctx.payme_id, order_id=order.id, amount=ctx.amount, create_time=ctx.time, state=1, ) self.db.add(tx) await self.db.execute( update(Order) .where(Order.id == order.id) .values(payme_transaction_id=ctx.payme_id) ) await self.db.commit() return ctx.ok(transaction_id=tx.payme_id, create_time=tx.create_time) async def perform_transaction(self, ctx: PerformTransactionCtx): tx = await self.get_transaction(ctx.transaction_id) if not tx: raise Errors.transaction_not_found() if tx.state == 2: return ctx.ok(transaction_id=tx.payme_id, perform_time=tx.perform_time, state=2) tx.state = 2 tx.perform_time = time_to_payme() await self.db.execute( update(Order) .where(Order.id == tx.order_id) .values(status=OrderStatus.PAID.value) ) await self.db.commit() return ctx.ok(transaction_id=tx.payme_id, perform_time=tx.perform_time, state=2) async def cancel_transaction(self, ctx: CancelTransactionCtx): tx = await self.get_transaction(ctx.transaction_id) if not tx: raise Errors.transaction_not_found() if tx.state in (-1, -2): return ctx.ok( transaction=tx.payme_id, cancel_time=tx.cancel_time, state=tx.state, reason=tx.reason, ) tx.state = -2 if tx.state == 2 else -1 tx.cancel_time = time_to_payme() tx.reason = ctx.reason await self.db.execute( update(Order) .where(Order.id == tx.order_id) .values(status=OrderStatus.CANCELLED.value) ) await self.db.commit() return ctx.ok( transaction=tx.payme_id, state=tx.state, cancel_time=tx.cancel_time, reason=tx.reason, ) async def check_transaction(self, ctx: CheckTransactionCtx): tx = await self.get_transaction(ctx.transaction_id) if not tx: raise Errors.transaction_not_found() return ctx.ok( state=tx.state, create_time=tx.create_time, perform_time=tx.perform_time, cancel_time=tx.cancel_time, reason=tx.reason, ) async def get_statement(self, ctx: GetStatementCtx): from_time = ctx.from_time to_time = ctx.to_time if from_time > to_time: from_time, to_time = to_time, from_time txs = await self.get_transactions(from_time, to_time) return ctx.ok(transactions=[ { "id": tx.payme_id, "time": tx.create_time, "amount": tx.amount, "account": {"order_id": tx.order_id}, "state": tx.state, "create_time": tx.create_time, "perform_time": tx.perform_time or 0, "cancel_time": tx.cancel_time or 0, "reason": tx.reason, } for tx in txs ])
Обратите внимание — сумма в Payme передаётся в тийинах (1 сум = 100 тийин), поэтому при сравнении умножаем order.amount * 100.
Хендлеры
Теперь создаём роутер и подключаем сервис. Каждый метод Payme — это отдельный декоратор. Dependency injection работает автоматически — просто добавляем нужные зависимости как аргументы функции:
# handlers/payme.pyfrom aiopayme import Routerfrom aiopayme.types import *from sqlalchemy.ext.asyncio import AsyncSessionfrom services.payme import PaymeServicefrom deps import botrouter = Router()@router.check_perform_transaction()async def check_perform(ctx: CheckPerformTransactionCtx, db: AsyncSession): return await PaymeService(db).check_perform(ctx)@router.create_transaction()async def create_transaction(ctx: CreateTransactionCtx, db: AsyncSession): return await PaymeService(db).create_transaction(ctx)@router.perform_transaction()async def perform_transaction(ctx: PerformTransactionCtx, db: AsyncSession): result = await PaymeService(db).perform_transaction(ctx) tx = await PaymeService(db).get_transaction(ctx.transaction_id) order = await PaymeService(db).get_order(tx.order_id) await bot.send_message( chat_id=order.telegram_id, text="✅ Оплата прошла успешно." ) return result@router.cancel_transaction()async def cancel_transaction(ctx: CancelTransactionCtx, db: AsyncSession): return await PaymeService(db).cancel_transaction(ctx)@router.check_transaction()async def check_transaction(ctx: CheckTransactionCtx, db: AsyncSession): return await PaymeService(db).check_transaction(ctx)@router.get_statement()async def get_statement(ctx: GetStatementCtx, db: AsyncSession): return await PaymeService(db).get_statement(ctx)
deps.py
Здесь инициализируем все зависимости — базу данных, бота и Payme:
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmakerfrom aiopayme import Paymefrom aiogram import Botengine = create_async_engine("sqlite+aiosqlite:///./db.sqlite3")SessionLocal = async_sessionmaker(engine, expire_on_commit=False)payme = Payme( merchant_id="your_merchant_id", secret_key=your_secret_test_key, sandbox=True, echo=True,)bot = Bot(token="your_bot_token")
Роутеры
Создаём два роутера — для вебхука Payme и для создания заказа:
# routers/order.pyfrom fastapi import APIRouterfrom sqlalchemy import insertfrom pydantic import BaseModelfrom deps import SessionLocal, paymefrom models import Orderrouter = APIRouter()class OrderCreate(BaseModel): telegram_id: int amount: int@router.post("/order/create")async def create_order(data: OrderCreate): async with SessionLocal() as db: result = await db.execute( insert(Order).values( telegram_id=data.telegram_id, amount=data.amount ).returning(Order.id) ) order_id = result.scalar() await db.commit() pay_link = payme.generate_pay_link( amount=data.amount, account={ "order_id": order_id, "telegram_id": data.telegram_id } ) return {"order_id": order_id, "pay_link": pay_link}
# routers/webhook.pyfrom fastapi import APIRouter, Requestfrom deps import paymerouter = APIRouter()@router.post("/payme")async def payme_webhook(request: Request): data = await request.json() result = await payme.handle( data=data, headers=dict(request.headers) ) return result
Payme отправляет все запросы на один эндпоинт POST /payme. Метод handle сам аутентифицирует запрос и направляет его в нужный хендлер.
Telegram бот
Бот принимает команду /pay, создаёт заказ через API и отправляет пользователю ссылку на оплату:
# bot/main.pyimport asyncioimport httpxfrom aiogram import Dispatcherfrom aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButtonfrom aiogram.filters import Commandfrom deps import botdp = Dispatcher()def get_pay_link_btn(url: str) -> InlineKeyboardMarkup: return InlineKeyboardMarkup( inline_keyboard=[ [InlineKeyboardButton(text="✅ Оплатить", url=url)] ] )@dp.message(Command("pay"))async def cmd_pay(message: Message): async with httpx.AsyncClient() as client: response = await client.post( "http://localhost:8000/order/create", json={ "telegram_id": message.from_user.id, "amount": 1000 } ) data = response.json() await message.answer( "Оплатить можно по кнопке ниже", reply_markup=get_pay_link_btn(data["pay_link"]) )async def main(): await dp.start_polling(bot)if __name__ == '__main__': asyncio.run(main())
После успешной оплаты Payme вызывает perform_transaction — и бот автоматически отправляет пользователю уведомление.
Админка
Для удобного управления заказами и транзакциями подключим sqladmin:
# admin/admin.pyfrom sqladmin import ModelViewfrom models import Order, PaymeTransactionclass OrderAdmin(ModelView, model=Order): name = "Order" name_plural = "Orders" column_list = "__all__"class PaymeTransactionAdmin(ModelView, model=PaymeTransaction): name = "Transaction" name_plural = "Transactions" column_list = "__all__"
Админка будет доступна по адресу http://localhost:8000/admin.
main.py
Собираем всё вместе — FastAPI, Payme, админка:
# main.pyimport asynciofrom contextlib import asynccontextmanagerfrom fastapi import FastAPIfrom sqlalchemy.ext.asyncio import AsyncSessionfrom sqladmin import Adminimport uvicornfrom aiopayme import Dispatcherfrom models import Basefrom deps import engine, SessionLocal, paymefrom routers.order import router as order_routerfrom routers.webhook import router as webhook_routerfrom handlers.payme import router as payme_routerfrom admin.admin import OrderAdmin, PaymeTransactionAdmindp = Dispatcher()dp.include_router(payme_router)payme.setup(dp)payme.provide(AsyncSession, SessionLocal)@asynccontextmanagerasync def lifespan(app: FastAPI): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield await engine.dispose()app = FastAPI(lifespan=lifespan)app.include_router(order_router)app.include_router(webhook_router)admin = Admin(app, engine)admin.add_view(OrderAdmin)admin.add_view(PaymeTransactionAdmin)async def main(): config = uvicorn.Config(app, host="0.0.0.0", port=8000) server = uvicorn.Server(config) await server.serve()if __name__ == '__main__': asyncio.run(main())
Бот запускается отдельным процессом:
python bot/main.py # Telegram botpython main.py # backend
Локальная разработка
туннель Payme должен достучаться до вашего локального сервера. Для этого используем туннель — ngrok или cloudflared.
cloudflared tunnel --url http://0.0.0.0/8000
Активация кассы
Перед боевым использованием необходимо пройти проверку в песочнице Payme. Все методы (CheckPerformTransaction, CreateTransaction, PerformTransaction и т.д.) должны корректно отвечать на тестовые запросы.
Заключение
Интеграция готова! Перед выходом в продакшн не забудьте: — Заменить secret_key на боевой
payme = Payme( merchant_id="your_merchant_id", secret_key="your_secret_key", # prod sandbox=False, echo=False)
Спасибо за внимание! Удачи с интеграцией! 🚀
ссылка на оригинал статьи https://habr.com/ru/articles/1042466/