Как подключить Payme к Telegram боту на Python / forpes.ru

Главная
Как подключить Payme к Telegram боту на Python

Как подключить Payme к Telegram боту на Python -1

02.06.2026 06:00

pywave 0 3800 Источник

Всем привет! В этой статье разберём как подключить Payme — одну из самых популярных платёжных систем Узбекистана — к Telegram боту на Python. Для этого мы будем использовать библиотеку aiopayme — async-first решение с роутерами и dependency injection как в aiogram и FastAPI. В конце статьи вы получите полностью рабочую интеграцию: бот принимает команду /pay, пользователь оплачивает через Payme, бот получает уведомление об успешной оплате.

Установка

pip install aiopayme

Что такое aiopayme? aiopayme — это асинхронная Python библиотека для интеграции платёжной системы Payme. Главная особенность — роутерная архитектура, знакомая каждому кто работал с aiogram или FastAPI:

from aiopayme import Router

router = Router()

@router.check_perform_transaction()
async def check_perform(ctx: CheckPerformTransactionCtx, db: AsyncSession): ...

Единственная зависимость — httpx. Библиотека не навязывает ORM, фреймворк или структуру проекта.

Модели

Создадим две модели — заказ и транзакцию Payme.

# models/order.py

import enum
from sqlalchemy import Column, BigInteger, Integer, String
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class 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.py

from sqlalchemy import Column, Integer, String, BigInteger
from models.order import Base


class 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.py

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from aiopayme.exceptions import Errors
from aiopayme.utils import time_to_payme
from aiopayme.types import (
    CheckPerformTransactionCtx,
    CreateTransactionCtx,
    PerformTransactionCtx,
    CancelTransactionCtx,
    CheckTransactionCtx,
    GetStatementCtx,
)

from models import OrderStatus, Order, PaymeTransaction


class 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.py

from aiopayme import Router
from aiopayme.types import *
from sqlalchemy.ext.asyncio import AsyncSession

from services.payme import PaymeService

from deps import bot

router = 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_sessionmaker
from aiopayme import Payme
from aiogram import Bot

engine = 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.py

from fastapi import APIRouter
from sqlalchemy import insert
from pydantic import BaseModel

from deps import SessionLocal, payme
from models import Order

router = 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.py

from fastapi import APIRouter, Request
from deps import payme

router = 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.py

import asyncio
import httpx

from aiogram import Dispatcher
from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.filters import Command

from deps import bot

dp = 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.py

from sqladmin import ModelView
from models import Order, PaymeTransaction

class 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.py

import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
from sqladmin import Admin
import uvicorn

from aiopayme import Dispatcher

from models import Base
from deps import engine, SessionLocal, payme
from routers.order import router as order_router
from routers.webhook import router as webhook_router
from handlers.payme import router as payme_router
from admin.admin import OrderAdmin, PaymeTransactionAdmin

dp = Dispatcher()
dp.include_router(payme_router)

payme.setup(dp)
payme.provide(AsyncSession, SessionLocal)

@asynccontextmanager
async 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 bot
python 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
)

Спасибо за внимание! Удачи с интеграцией! ?