Урок 6.5 — PollingConfig и PayloadData

PollingConfig — это dataclass, который определяет параметры polling-цикла. Он передаётся в конструктор CryptoPay или в метод start_polling().

from dataclasses import dataclass

@dataclass
class PollingConfig:
    timeout: int = 300   # максимальное время ожидания события (сек)
    delay: float = 2.0  # задержка между запросами (сек)
    
    # Можно добавить свои поля (при наследовании)
    ...

PollingConfig можно передать двумя способами:

from aiosend import CryptoPay
from aiosend.polling import PollingConfig

# Способ 1: в конструкторе CryptoPay (используется по умолчанию)
cp1 = CryptoPay(
    token="YOUR_TOKEN",
    polling_config=PollingConfig(
        timeout=600,   # ждём 10 минут
        delay=3,       # проверяем каждые 3 секунды
    ),
)

# Способ 2: в start_polling (переопределяет на время запуска)
@cp1.invoice_paid()
async def on_paid(invoice, **kwargs):
    print(f"Оплачено: {invoice.invoice_id}")

async def main():
    # Переопределяем конфиг для этого запуска
    await cp1.start_polling(
        polling_config=PollingConfig(timeout=120, delay=1),
    )

PollingManager — это класс, который непосредственно управляет циклом polling. Он создаётся внутри CryptoPay при вызове start_polling().

Ключевые методы PollingManager:

Внутренний цикл PollingManager выглядит так:

class PollingManager:
    def __init__(self, client: CryptoPay, config: PollingConfig):
        self.client = client
        self.config = config
        self._is_running = False
    
    async def run(self):
        self._is_running = True
        last_invoice_id = 0
        last_check_id = 0
        
        try:
            while self._is_running:
                # 1. Проверяем новые оплаченные инвойсы
                paid = await self.client.get_invoices(
                    status="paid",
                    offset=last_invoice_id,
                )
                for invoice in paid:
                    await self._process_invoice_paid(invoice)
                    last_invoice_id = max(last_invoice_id, invoice.invoice_id)
                
                # 2. Проверяем новые просроченные инвойсы
                expired = await self.client.get_invoices(
                    status="expired",
                    offset=last_invoice_id,
                )
                for invoice in expired:
                    await self._process_invoice_expired(invoice)
                    last_invoice_id = max(last_invoice_id, invoice.invoice_id)
                
                # 3. Проверяем новые чеки (если есть обработчики)
                activated = await self.client.get_checks(
                    status="activated",
                    offset=last_check_id,
                )
                for check in activated:
                    await self._process_check_activated(check)
                    last_check_id = max(last_check_id, check.check_id)
                
                # 4. Задержка между циклами
                await asyncio.sleep(self.config.delay)
                
        except asyncio.CancelledError:
            pass
        finally:
            self._is_running = False
    
    async def _process_invoice_paid(self, invoice):
        # Применяет фильтры и вызывает обработчики
        for handler in self.client._paid_handlers:
            if await self._check_filters(handler.filters, invoice):
                await handler.func(invoice, **handler.kwargs)
    
    async def stop(self):
        self._is_running = False

PollingManager использует offset для отслеживания, какие инвойсы/чеки уже обработаны. Это гарантирует, что каждое событие будет обработано ровно один раз.

PayloadData — это Pydantic-модель для создания структурированных payload'ов. Она решает проблему «сырых» строковых payload'ов, позволяя упаковывать данные в формат prefix:sep:data и легко извлекать их обратно.

from pydantic import BaseModel

class PayloadData(BaseModel):
    """Базовый класс для типизированных payload."""
    
    prefix: str = ""     # префикс для идентификации типа payload
    sep: str = ":"       # разделитель между частями
    
    def pack(self) -> str:
        """Упаковывает данные в строку payload."""
        ...
    
    @classmethod
    def unpack(cls, payload: str) -> "PayloadData | None":
        """Распаковывает строку payload в объект PayloadData."""
        ...
    
    @classmethod
    def filter(cls, condition) -> "PayloadData":
        """Создаёт MagicFilter для фильтрации по payload."""
        ...

Основные возможности PayloadData:

• pack() — упаковывает поля модели в строку вида prefix:field1:field2:...
• unpack() — распаковывает строку обратно в экземпляр модели
• filter() — создаёт MagicFilter для поиска инвойсов/чеков с определённым payload

from aiosend.payload import PayloadData

# Создаём данные и упаковываем
data = PayloadData(prefix="order", sep=":")
payload_str = data.pack()
print(payload_str)  # "order::"

# Можно передать кастомные поля (Pydantic это допускает)
# Но лучше создать подкласс (см. ниже)

# Распаковываем обратно
parsed = PayloadData.unpack("order::")
print(parsed.prefix)  # "order"

# Создаём фильтр для поиска инвойсов с этим payload
filter_ = PayloadData.filter(prefix="order")
# filter_ теперь можно использовать в @cp.invoice_paid(filter_)

Настоящая сила PayloadData раскрывается при создании своих подклассов с типизированными полями.

from aiosend.payload import PayloadData
from typing import Optional

class OrderPayload(PayloadData):
    """Payload для заказов."""
    prefix: str = "order"
    user_id: int = 0
    product_id: str = ""
    quantity: int = 1
    
    def description(self) -> str:
        return f"Заказ {self.product_id} x{self.quantity} для user #{self.user_id}"

class BonusPayload(PayloadData):
    """Payload для бонусов."""
    prefix: str = "bonus"
    user_id: int = 0
    reason: str = ""
    
    def description(self) -> str:
        return f"Бонус для user #{self.user_id}: {self.reason}"

class SubscriptionPayload(PayloadData):
    """Payload для подписок."""
    prefix: str = "sub"
    user_id: int = 0
    plan: str = "monthly"
    auto_renew: bool = False
    sep: str = "|"  # кастомный разделитель

Теперь используем эти модели на практике:

import asyncio
from aiosend import CryptoPay
from aiosend.payload import PayloadData

cp = CryptoPay(token="YOUR_TOKEN")

# ─── Создание инвойса с PayloadData ─────────────────────

# Упаковываем данные заказа
order = OrderPayload(
    user_id=12345,
    product_id="premium_month",
    quantity=1,
)
payload_str = order.pack()
print(f"Payload: {payload_str}")  # "order:12345:premium_month:1"

# Создаём инвойс с этим payload
invoice = await cp.create_invoice(
    amount=25,
    asset="USDT",
    payload=payload_str,
)

# ─── Обработка оплаты с распаковкой ─────────────────────

@cp.invoice_paid()
async def on_paid(invoice, **kwargs):
    # Распаковываем payload обратно в модель
    parsed = OrderPayload.unpack(invoice.payload)
    if parsed:
        print(f"✅ {parsed.description()}")
        print(f"   Сумма: {invoice.amount} {invoice.asset}")
        # Здесь бизнес-логика по заказу
        await process_order(parsed)
    else:
        print(f"⚠️ Неизвестный формат payload: {invoice.payload}")

async def process_order(order: OrderPayload):
    await asyncio.sleep(0.3)
    print(f"   Заказ обработан: {order.product_id} x{order.quantity}")

async def main():
    await cp.start_polling()

asyncio.run(main())

Метод PayloadData.filter() создаёт MagicFilter, который можно передать в декоратор для фильтрации событий по содержимому payload. Это работает в паре с методом pack() — фильтр проверяет, начинается ли строка payload с определённого префикса.

# Создаём фильтр для каждого типа payload
order_filter = OrderPayload.filter()  # проверяет prefix="order"
bonus_filter = BonusPayload.filter()  # проверяет prefix="bonus"
sub_filter = SubscriptionPayload.filter()  # проверяет prefix="sub"

# Используем фильтры в декораторах
@cp.invoice_paid(order_filter)
async def on_order_paid(invoice, **kwargs):
    parsed = OrderPayload.unpack(invoice.payload)
    print(f"📦 Оплачен заказ: {parsed.description()}")

@cp.invoice_paid(bonus_filter)
async def on_bonus_paid(invoice, **kwargs):
    parsed = BonusPayload.unpack(invoice.payload)
    print(f"🎁 Бонус оплачен: {parsed.description()}")

@cp.invoice_paid(sub_filter)
async def on_sub_paid(invoice, **kwargs):
    parsed = SubscriptionPayload.unpack(invoice.payload)
    print(f"💳 Подписка {parsed.plan} оплачена (auto_renew={parsed.auto_renew})")

# Можно комбинировать с другими фильтрами
@cp.invoice_paid(order_filter, F.amount > 100)
async def on_large_order(invoice, **kwargs):
    parsed = OrderPayload.unpack(invoice.payload)
    print(f"💰 Крупный заказ: {parsed.description()} на {invoice.amount} USDT")

Вы также можете создать фильтр с произвольным префиксом:

# Фильтр для заказов с конкретным product_id
order_filter_custom = OrderPayload.filter(product_id="premium_month")

# Фильтр для бонусов с определённой причиной
bonus_filter_custom = BonusPayload.filter(reason="welcome")

# Фильтр для подписок с auto_renew=True
sub_filter_renew = SubscriptionPayload.filter(auto_renew=True)

# Использование:
@cp.invoice_paid(order_filter_custom)
async def on_premium_order(invoice, **kwargs):
    print(f"🌟 Премиум-заказ оплачен!")
    # parsed = OrderPayload.unpack(invoice.payload)
    # гарантированно product_id="premium_month"

Обратите внимание: filter() с кастомными параметрами проверяет только префикс и разделитель. Он не проверяет содержимое полей (это сделает unpack() в обработчике). Фильтр лишь гарантирует, что payload имеет правильный формат.

Рассмотрим более сложные и специализированные сценарии использования PayloadData в реальных проектах.

Пример: Вложенные PayloadData модели

Вы можете использовать вложенные Pydantic-модели внутри PayloadData для сложных структур данных:

from pydantic import BaseModel
from aiosend.payload import PayloadData
from typing import Optional, List

class Address(BaseModel):
    """Адрес доставки."""
    city: str = ""
    street: str = ""
    house: str = ""
    apartment: str = ""

class CartItem(BaseModel):
    """Товар в корзине."""
    product_id: str = ""
    name: str = ""
    price: float = 0.0
    quantity: int = 1

class OrderPayload(PayloadData):
    """Сложный заказ с вложенными моделями."""
    prefix: str = "order"
    user_id: int = 0
    address: Optional[Address] = None
    items: List[CartItem] = []
    promo_code: str = ""
    
    def total(self) -> float:
        return sum(item.price * item.quantity for item in self.items)
    
    def __str__(self) -> str:
        return f"Заказ #{self.user_id}: {len(self.items)} товаров на ${self.total():.2f}"

# Использование:
order = OrderPayload(
    user_id=12345,
    address=Address(city="Москва", street="Тверская", house="1"),
    items=[
        CartItem(product_id="p1", name="Футболка", price=29.99, quantity=2),
        CartItem(product_id="p2", name="Кепка", price=19.99, quantity=1),
    ],
    promo_code="WELCOME10",
)

payload_str = order.pack()
print(f"Payload: {payload_str}")
print(f"Описание: {order}")
print(f"Сумма: ${order.total():.2f}")

# Распаковка
parsed = OrderPayload.unpack(payload_str)
if parsed:
    for item in parsed.items:
        print(f"  - {item.name} x{item.quantity} = ${item.price * item.quantity:.2f}")

Пример: PayloadData с сериализацией datetime

Можно использовать поля Pydantic с кастомными типами:

from datetime import datetime
from aiosend.payload import PayloadData

class EventPayload(PayloadData):
    """Payload для мероприятий."""
    prefix: str = "event"
    user_id: int = 0
    event_id: str = ""
    booked_at: str = ""  # ISO формат даты
    
    @classmethod
    def create(cls, user_id: int, event_id: str) -> "EventPayload":
        return cls(
            user_id=user_id,
            event_id=event_id,
            booked_at=datetime.utcnow().isoformat(),
        )
    
    @property
    def booking_time(self) -> datetime | None:
        try:
            return datetime.fromisoformat(self.booked_at)
        except (ValueError, TypeError):
            return None

class PromoPayload(PayloadData):
    """Payload для промо-акций."""
    prefix: str = "promo"
    user_id: int = 0
    promo_id: str = ""
    expires_at: str = ""
    
    @property
    def is_expired(self) -> bool:
        if not self.expires_at:
            return False
        try:
            exp = datetime.fromisoformat(self.expires_at)
            return datetime.utcnow() > exp
        except (ValueError, TypeError):
            return False

# Использование:
event = EventPayload.create(user_id=123, event_id="conf_2024")
print(f"Бронирование: {event.booking_time}")

promo = PromoPayload(
    user_id=456,
    promo_id="SUMMER24",
    expires_at="2025-01-01T00:00:00",
)
print(f"Промо истекло: {promo.is_expired}")

# Использование с инвойсом
invoice = await cp.create_invoice(
    amount=100,
    asset="USDT",
    payload=promo.pack(),
)

# В обработчике:
@cp.invoice_paid(PromoPayload.filter())
async def on_promo_paid(invoice, **kwargs):
    parsed = PromoPayload.unpack(invoice.payload)
    if parsed and not parsed.is_expired:
        print(f"🎉 Промо {parsed.promo_id} активировано!")
    elif parsed and parsed.is_expired:
        print(f"⏰ Промо {parsed.promo_id} истекло")

Пример: PayloadData с валидацией

Pydantic-валидация работает и в PayloadData:

from pydantic import field_validator, Field
from aiosend.payload import PayloadData

class PaymentPayload(PayloadData):
    """Payload с валидацией полей."""
    prefix: str = "pay"
    user_id: int = Field(..., ge=1)
    amount: float = Field(..., gt=0)
    currency: str = "USDT"
    
    @field_validator("currency")
    @classmethod
    def validate_currency(cls, v: str) -> str:
        allowed = ["USDT", "TON", "BTC", "ETH", "USDC"]
        if v not in allowed:
            raise ValueError(f"Валюта {v} не поддерживается")
        return v
    
    @field_validator("amount")
    @classmethod
    def validate_amount(cls, v: float) -> float:
        if v > 100000:
            raise ValueError("Сумма превышает лимит")
        return v

# При попытке создать с неверными данными — ошибка валидации
try:
    bad = PaymentPayload(
        prefix="pay",
        user_id=0,  # ошибка: ge=1
        amount=-10,  # ошибка: gt=0
        currency="INVALID",
    )
except Exception as e:
    print(f"Ошибка валидации: {e}")

# Правильное использование
good = PaymentPayload(
    prefix="pay",
    user_id=123,
    amount=99.99,
    currency="USDT",
)
print(f"Payload: {good.pack()}")
print(f"Валидация пройдена!")

Пример: Миграция PayloadData между версиями

Если структура данных меняется со временем, можно поддерживать несколько версий:

from aiosend.payload import PayloadData

class UserPayloadV1(PayloadData):
    """Версия 1: базовые поля."""
    prefix: str = "user"
    user_id: int = 0
    name: str = ""

class UserPayloadV2(PayloadData):
    """Версия 2: добавлен email."""
    prefix: str = "user"
    user_id: int = 0
    name: str = ""
    email: str = ""

# Функция для обработки payload любой версии
def process_user_payload(payload_str: str) -> dict | None:
    """Пытается распарсить payload любой версии."""
    # Сначала пробуем V2
    v2 = UserPayloadV2.unpack(payload_str)
    if v2:
        return {
            "user_id": v2.user_id,
            "name": v2.name,
            "email": v2.email,
        }
    
    # Затем V1
    v1 = UserPayloadV1.unpack(payload_str)
    if v1:
        return {
            "user_id": v1.user_id,
            "name": v1.name,
            "email": "unknown@example.com",  # значение по умолчанию
        }
    
    return None

@cp.invoice_paid()
async def on_paid(invoice, **kwargs):
    data = process_user_payload(invoice.payload)
    if data:
        print(f"Пользователь {data['name']} (id={data['user_id']}) "
              f"email={data['email']} оплатил {invoice.amount} USDT")

Создадим систему для интернет-магазина, которая использует PayloadData для идентификации заказов и PollingConfig для настройки поведения polling.

import asyncio
import logging
from dataclasses import dataclass, field
from typing import Optional
from magic_filter import F
from aiosend import CryptoPay
from aiosend.polling import PollingConfig
from aiosend.payload import PayloadData

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ─── Payload модели ──────────────────────────────────────

class ProductPayload(PayloadData):
    prefix: str = "product"
    product_id: str = ""
    user_id: int = 0
    
    def __str__(self):
        return f"Товар #{self.product_id} для user #{self.user_id}"

class SubscriptionPayload(PayloadData):
    prefix: str = "sub"
    user_id: int = 0
    months: int = 1
    sep: str = "|"
    
    def __str__(self):
        return f"Подписка на {self.months} мес. для user #{self.user_id}"

# ─── Магазин ──────────────────────────────────────────────

@dataclass
class Store:
    cp: CryptoPay
    orders: dict = field(default_factory=dict)
    
    def register_handlers(self):
        # Обработчик оплаты товара
        @self.cp.invoice_paid(ProductPayload.filter())
        async def on_product_paid(invoice, **kwargs):
            parsed = ProductPayload.unpack(invoice.payload)
            if parsed:
                logger.info(f"🛒 {parsed} — {invoice.amount} USDT")
                self.orders[parsed.product_id] = "paid"
                await self.deliver_product(parsed)
        
        # Обработчик оплаты подписки
        @self.cp.invoice_paid(SubscriptionPayload.filter())
        async def on_sub_paid(invoice, **kwargs):
            parsed = SubscriptionPayload.unpack(invoice.payload)
            if parsed:
                logger.info(f"💳 {parsed} — {invoice.amount} USDT")
                await self.activate_subscription(parsed)
        
        # Обработчик просрочки
        @self.cp.invoice_expired()
        async def on_expired(invoice, **kwargs):
            logger.warning(f"⏰ Инвойс #{invoice.invoice_id} просрочен")
    
    async def deliver_product(self, payload: ProductPayload):
        await asyncio.sleep(0.5)
        logger.info(f"   📦 Товар #{payload.product_id} доставлен user #{payload.user_id}")
    
    async def activate_subscription(self, payload: SubscriptionPayload):
        await asyncio.sleep(0.3)
        logger.info(f"   ✅ Подписка активирована на {payload.months} мес.")
    
    async def create_product_order(self, product_id: str, user_id: int,
                                    price: float):
        payload = ProductPayload(product_id=product_id, user_id=user_id)
        invoice = await self.cp.create_invoice(
            amount=price,
            asset="USDT",
            payload=payload.pack(),
            description=str(payload),
        )
        logger.info(f"📄 Счёт #{invoice.invoice_id}: {payload}")
        return invoice
    
    async def create_subscription(self, user_id: int, months: int, price: float):
        payload = SubscriptionPayload(user_id=user_id, months=months)
        invoice = await self.cp.create_invoice(
            amount=price,
            asset="USDT",
            payload=payload.pack(),
            description=str(payload),
            expires_in=3600,
        )
        logger.info(f"📄 Счёт #{invoice.invoice_id}: {payload}")
        return invoice

async def main():
    cp = CryptoPay(
        token="YOUR_TOKEN",
        polling_config=PollingConfig(timeout=600, delay=2),
    )
    
    store = Store(cp)
    store.register_handlers()
    
    # Создаём тестовые заказы
    await store.create_product_order("game_pass", 101, 15)
    await store.create_product_order("vip_badge", 102, 50)
    await store.create_subscription(103, 3, 75)
    
    logger.info("🚀 Запуск polling...")
    try:
        await cp.start_polling()
    except KeyboardInterrupt:
        logger.info("Магазин закрыт")

if __name__ == "__main__":
    asyncio.run(main())

В реальном проекте PollingConfig и PayloadData используются вместе. PollingConfig настраивает механизм доставки событий, а PayloadData структурирует данные внутри этих событий.

import os
from aiosend import CryptoPay
from aiosend.polling import PollingConfig
from aiosend.payload import PayloadData

class PaymentService:
    """Сервис для управления платежами с поллинтом."""
    
    def __init__(self, token: str | None = None):
        self.cp = CryptoPay(
            token=token or os.getenv("CRYPTOPAY_TOKEN"),
            polling_config=PollingConfig(
                timeout=int(os.getenv("POLLING_TIMEOUT", "300")),
                delay=float(os.getenv("POLLING_DELAY", "2.0")),
            ),
        )
        self._register_handlers()
    
    def _register_handlers(self):
        @self.cp.invoice_paid()
        async def handle_paid(invoice, **kwargs):
            # Распаковываем payload в любую известную модель
            for model in [OrderPayload, BonusPayload, SubPayload]:
                parsed = model.unpack(invoice.payload)
                if parsed:
                    await self._process(invoice, parsed)
                    break
            else:
                # Неизвестный формат
                print(f"⚠️ Неизвестный payload: {invoice.payload}")
        
        @self.cp.invoice_expired()
        async def handle_expired(invoice, **kwargs):
            print(f"⏰ Просрочен инвойс #{invoice.invoice_id}")
    
    async def _process(self, invoice, payload):
        print(f"Обработка: {payload}")
    
    async def start(self):
        await self.cp.start_polling()

Такой сервис можно использовать как основу для любого проекта. Он гибко настраивается через переменные окружения, поддерживает любые типы PayloadData и корректно обрабатывает все события.