Урок 7.4 — Роутеры

BaseRouter — это базовый класс для всех роутеров. Он предоставляет те же декораторы для регистрации хендлеров, что и CryptoPay: @router.invoice_paid(), @router.check_paid() и другие.

from aiosend import CryptoPay, BaseRouter

# Создаём роутер
router = BaseRouter()

# Регистрируем хендлеры на роутере
@router.invoice_paid()
async def on_invoice_paid(invoice):
    print(f"Инвойс #{invoice.invoice_id} оплачен!")

@router.check_paid()
async def on_check_paid(check):
    print(f"Чек #{check.check_id} активирован!")

@router.invoice_expired()
async def on_invoice_expired(invoice):
    print(f"Инвойс #{invoice.invoice_id} истёк")

# Подключаем роутер к клиенту
cp = CryptoPay(token="1234:TOKEN")
cp.include_router(router)

# После include_router все хендлеры роутера активны
await cp.start_polling()

Доступные декораторы BaseRouter:

PollingRouter расширяет BaseRouter и добавляет функциональность для работы с polling. По сути, это тот же BaseRouter, но с дополнительными методами для запуска/остановки polling, если ты хочешь сделать роутер самодостаточным.

from aiosend import CryptoPay, PollingRouter

# PollingRouter — наследник BaseRouter
router = PollingRouter()

@router.invoice_paid()
async def on_paid(invoice):
    print(f"Polling: инвойс #{invoice.invoice_id} оплачен!")

cp = CryptoPay(token="1234:TOKEN")
cp.include_router(router)

# Запускаем polling (хендлеры из роутера будут работать)
await cp.start_polling()

# PollingRouter можно использовать там же, где и BaseRouter.
# Разница только в том, что у PollingRouter есть доп. методы
# для самостоятельного управления polling.

WebhookRouter — роутер, предназначенный для работы с вебхуками. Он также расширяет BaseRouter.

from aiosend import CryptoPay, WebhookRouter
from aiosend.webhook import FastAPIManager
from fastapi import FastAPI

app = FastAPI()
manager = FastAPIManager(app, path="/webhook")

# Создаём роутер для вебхуков
router = WebhookRouter()

@router.invoice_paid()
async def on_paid(invoice):
    print(f"Webhook: инвойс #{invoice.invoice_id} оплачен!")

cp = CryptoPay(
    token="1234:TOKEN",
    webhook_manager=manager,
)
cp.include_router(router)

# Все хендлеры из роутера будут обрабатывать вебхуки
# FastAPI сам запускается через uvicorn

Методы include_router() и include_routers() подключают роутеры к клиенту. При подключении все хендлеры из роутера копируются в клиент.

include_router()

Подключает один роутер:

from aiosend import CryptoPay, BaseRouter

router = BaseRouter()

@router.invoice_paid()
async def handler(invoice):
    print(f"Оплачен: #{invoice.invoice_id}")

cp = CryptoPay(token="1234:TOKEN")
cp.include_router(router)  # подключаем один роутер

include_routers()

Подключает несколько роутеров за раз:

from aiosend import CryptoPay, BaseRouter

router1 = BaseRouter()
router2 = BaseRouter()
router3 = BaseRouter()

# Регистрируем хендлеры на каждом роутере
@router1.invoice_paid()
async def handler1(invoice):
    print(f"Хендлер 1: инвойс #{invoice.invoice_id}")

@router2.invoice_paid()
async def handler2(invoice):
    print(f"Хендлер 2: инвойс #{invoice.invoice_id}")

@router3.invoice_paid()
async def handler3(invoice):
    print(f"Хендлер 3: инвойс #{invoice.invoice_id}")

cp = CryptoPay(token="1234:TOKEN")

# Подключаем все три роутера одной строкой
cp.include_routers(router1, router2, router3)

# Все хендлеры из всех роутеров будут вызваны
# при оплате инвойса (один за другим)
await cp.start_polling()

Роутеры позволяют организовать код в многофайловую структуру. Каждый роутер — это отдельный файл (модуль) со своими хендлерами. Основной файл только подключает роутеры и запускает клиент.

Пример структуры проекта:

project/
├── main.py              # Точка входа
├── routers/
│   ├── __init__.py
│   ├── payments.py      # Роутер для платежей
│   ├── checks.py        # Роутер для чеков
│   └── admin.py         # Роутер для админ-событий
└── services/
    ├── __init__.py
    ├── database.py
    └── notifier.py

routers/payments.py

from aiosend import BaseRouter

router = BaseRouter()

@router.invoice_paid()
async def on_invoice_paid(invoice, cp):
    """Обработка оплаты инвойса."""
    db = cp.get("db")
    if db:
        await db.save_payment(invoice)
    print(f"Платёж #{invoice.invoice_id} обработан")

@router.invoice_expired()
async def on_invoice_expired(invoice):
    """Обработка просроченного инвойса."""
    print(f"Инвойс #{invoice.invoice_id} просрочен")

routers/checks.py

from aiosend import BaseRouter

router = BaseRouter()

@router.check_paid()
async def on_check_paid(check, cp):
    """Обработка активации чека."""
    notifier = cp.get("notifier")
    if notifier:
        await notifier.send_notification(
            f"Чек #{check.check_id} активирован!"
        )
    print(f"Чек #{check.check_id} активирован на {check.amount}")

main.py

import asyncio
from aiosend import CryptoPay
from routers.payments import router as payments_router
from routers.checks import router as checks_router

async def main():
    # Создаём клиент с DI
    cp = CryptoPay(
        token="1234:TOKEN",
        db=Database(),
        notifier=Notifier(),
    )

    # Подключаем все роутеры
    cp.include_routers(
        payments_router,
        checks_router,
    )

    # Запускаем polling
    await cp.start_polling()

asyncio.run(main())

Роутеры можно вкладывать друг в друга. Это полезно, когда у тебя есть группа хендлеров, которые логически belong к одному роутеру, но сами делятся на подгруппы.

from aiosend import BaseRouter

# Создаём «подроутеры»
payment_router = BaseRouter()
check_router = BaseRouter()

@payment_router.invoice_paid()
async def on_paid(invoice):
    print(f"Оплачен инвойс #{invoice.invoice_id}")

@check_router.check_paid()
async def on_check(check):
    print(f"Активирован чек #{check.check_id}")

# Создаём главный роутер и вкладываем подроутеры
main_router = BaseRouter()
main_router.include_router(payment_router)
main_router.include_router(check_router)

# Подключаем главный роутер к клиенту
cp = CryptoPay(token="1234:TOKEN")
cp.include_router(main_router)
# Все хендлеры из payment_router и check_router будут активны

# ============= routers/__init__.py =============
from .payments import router as payments_router
from .checks import router as checks_router
from .admin import router as admin_router

# ============= routers/payments.py =============
from aiosend import BaseRouter

router = BaseRouter()

@router.invoice_paid()
async def payment_handler(invoice, cp):
    db = cp["db"]
    await db.save({
        "type": "invoice",
        "id": invoice.invoice_id,
        "amount": invoice.amount,
        "asset": invoice.asset,
    })

@router.invoice_expired()
async def expiry_handler(invoice):
    print(f"Просрочен: #{invoice.invoice_id}")

# ============= routers/checks.py =============
from aiosend import BaseRouter

router = BaseRouter()

@router.check_paid()
async def check_handler(check, cp):
    notifier = cp["notifier"]
    await notifier.send(f"Чек #{check.check_id} активирован")

# ============= routers/admin.py =============
from aiosend import BaseRouter

router = BaseRouter()

@router.invoice_paid()
@router.check_paid()
async def admin_logger(event, cp):
    """Логирует все события."""
    logger = cp["logger"]
    event_type = "invoice" if hasattr(event, "invoice_id") else "check"
    event_id = getattr(event, "invoice_id", getattr(event, "check_id", None))
    await logger.log(f"{event_type} #{event_id}")

# ============= main.py =============
import asyncio
from aiosend import CryptoPay
from routers import (
    payments_router,
    checks_router,
    admin_router,
)

class Database:
    async def save(self, data):
        print(f"[DB] Сохранено: {data}")

class Notifier:
    async def send(self, msg):
        print(f"[NOTIFY] {msg}")

class Logger:
    async def log(self, msg):
        print(f"[LOG] {msg}")

async def main():
    cp = CryptoPay(
        token="1234:TOKEN",
        db=Database(),
        notifier=Notifier(),
        logger=Logger(),
    )

    cp.include_routers(
        payments_router,
        checks_router,
        admin_router,
    )

    print("Запуск polling...")
    await cp.start_polling()

asyncio.run(main())