Модуль 7 · Продвинутые техники

Урок 7.1 — Dependency Injection в aiosend

Изучаем механизм внедрения зависимостей (DI) в aiosend: передача произвольных kwargs в CryptoPay, чтение значений в хендлерах и передача параметров через invoice.poll().

Dependency Injection (DI) — это паттерн, при котором зависимости не создаются внутри объекта, а передаются извне. В aiosend DI реализован через механизм произвольных аргументов: ты можешь передать любые **kwargs в конструктор CryptoPay() и затем читать их в хендлерах или передавать через invoice.poll(). Это мощный инструмент для организации кода в больших проектах.

💉

Что такое Dependency Injection

В традиционном подходе объект сам создаёт свои зависимости. Например, хендлер сам создаёт подключение к базе данных. При DI зависимости передаются в объект извне — это делает код более гибким, тестируемым и слабо связанным.

В aiosend DI работает через словарь cp["key"] = value. Ты можешь сохранить в клиенте любые данные: пул соединений к БД, экземпляр бота, конфигурацию, сервисы — и затем получить к ним доступ в любом хендлере через cp["key"].

💡 Основная идея

Вместо того чтобы передавать зависимости через глобальные переменные или создавать их внутри хендлеров, ты регистрируешь всё нужное в клиенте при старте приложения. Хендлеры просто обращаются к cp["db"], cp["bot"] и т.д. Это делает код чище и упрощает тестирование.

❌ Без DI (плохо)

db = Database()

@cp.invoice_paid()
async def handler(inv):
    # зависимость создаётся вне
    # или через глобальную переменную
    await db.save_payment(inv)

✅ С DI (хорошо)

cp["db"] = Database()

@cp.invoice_paid()
async def handler(inv, cp):
    await cp["db"].save_payment(inv)

🔧

Передача произвольных kwargs в CryptoPay()

Конструктор CryptoPay() принимает не только документированные параметры, но и любые дополнительные **kwargs. Все они сохраняются во внутреннем словаре и становятся доступными через cp["key"].

Python · Передача зависимостей через конструктор

from aiosend import CryptoPay

# Допустим, у нас есть сервисы
class Database:
    async def save_payment(self, invoice):
        print(f"Сохранён платёж #{invoice.invoice_id}")

class MailService:
    async def send_receipt(self, invoice):
        print(f"Отправлен чек для #{invoice.invoice_id}")

# Создаём клиент и передаём зависимости
cp = CryptoPay(
    token="1234:TOKEN",
    db=Database(),          # произвольный kwargs
    mail=MailService(),     # ещё один
    config={"lang": "ru"},  # можно и простые значения
)

# Все переданные kwargs доступны через cp["key"]
print(cp["db"])      # 
print(cp["mail"])    # 
print(cp["config"])  # {"lang": "ru"}

🔍 Как это работает внутри?

Класс CryptoPay наследуется от миксина DIMixin, который реализует методы __setitem__, __getitem__ и __contains__ для работы в стиле словаря. При передаче **kwargs в конструктор, все они сохраняются в self._di_store.

Ты также можешь добавлять и изменять зависимости после создания клиента:

Python · Добавление зависимостей после создания

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

# Добавляем зависимости после создания
cp["db"] = Database()
cp["mail"] = MailService()
cp["started_at"] = "2026-01-15T10:00:00"

# Проверяем наличие ключа
if "db" in cp:
    print("База данных зарегистрирована")

# Удаляем зависимость
del cp["mail"]

⚠️ Важно: конфликты имён

Не используй ключи, совпадающие с именами атрибутов CryptoPay (например, session, network, polling_config). Они уже заняты внутренними механизмами. Используй уникальные имена: "my_db", "user_service", "app_config".

🎣

Доступ к зависимостям в хендлерах

Самая мощная особенность DI в aiosend — это автоматическая передача клиента cp в хендлеры. Когда ты регистрируешь хендлер через декоратор @cp.invoice_paid(), aiosend может передать в него объект клиента, если хендлер принимает параметр cp.

Python · Хендлер с доступом к DI

import asyncio
from aiosend import CryptoPay

# Сервисы
class PaymentService:
    async def process(self, invoice):
        print(f"Обработка платежа #{invoice.invoice_id}")

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

async def main():
    cp = CryptoPay(
        token="1234:TOKEN",
        payment_service=PaymentService(),
        logger=Logger(),
    )

    @cp.invoice_paid()
    async def on_paid(invoice, cp):  # cp передаётся автоматически
        logger = cp["logger"]
        svc = cp["payment_service"]

        await logger.log(f"Получен платёж #{invoice.invoice_id}")
        await svc.process(invoice)

    await cp.start_polling()

asyncio.run(main())

💡 Как aiosend передаёт cp в хендлер?

Когда хендлер вызывается, aiosend проверяет его сигнатуру с помощью inspect.signature(). Если среди параметров есть cp, aiosend автоматически подставляет объект клиента. Если параметра cp нет — он просто не передаётся. Это безопасно и не ломает существующие хендлеры.

Проверка наличия ключа

Всегда проверяй наличие ключа перед использованием, чтобы избежать KeyError:

Python · Безопасный доступ к DI

@cp.invoice_paid()
async def safe_handler(invoice, cp):
    # Проверяем наличие ключа
    if "db" not in cp:
        print("База данных не подключена!")
        return

    db = cp["db"]

    # Используем .get() со значением по умолчанию
    config = cp.get("config", {})
    lang = config.get("lang", "en")

    await db.save_payment(invoice)
    print(f"Сохранено. Язык: {lang}")

📨

Передача параметров через invoice.poll()

Метод invoice.poll() позволяет ожидать оплату конкретного инвойса. Через него тоже можно передать произвольные kwargs, которые будут доступны в хендлере on_paid.

Это особенно полезно, когда нужно привязать к платежу дополнительные данные:

user_id — кому зачислить товар
product_id — какой товар куплен
chat_id — куда отправить уведомление
callback — функцию для вызова после оплаты

Python · Передача user_id через invoice.poll()

import asyncio
from aiosend import CryptoPay

async def main():
    cp = CryptoPay(token="1234:TOKEN")

    # Создаём инвойс для конкретного пользователя
    invoice = await cp.create_invoice(
        amount=10,
        asset="USDT",
        payload="product_123",
    )

    # Ожидаем оплату, передавая user_id
    paid_invoice = await invoice.poll(
        on_paid=on_paid,           # хендлер
        user_id=12345,             # произвольный kwargs
        product_name="Премиум",
    )

async def on_paid(invoice, **kwargs):
    user_id = kwargs.get("user_id")
    product = kwargs.get("product_name")
    print(f"Пользователь {user_id} купил {product}!")
    print(f"Инвойс #{invoice.invoice_id} оплачен")

asyncio.run(main())

Передача callback-функции

Ты можешь передать не только данные, но и функции обратного вызова:

Python · Динамический callback через kwargs

async def send_goods(invoice, user_id):
    print(f"Отправляем товар пользователю {user_id}")
    print(f"Товар по инвойсу #{invoice.invoice_id}")

async def main():
    cp = CryptoPay(token="1234:TOKEN")

    invoice = await cp.create_invoice(amount=5, asset="USDT")

    # Передаём callback через kwargs
    paid = await invoice.poll(
        on_paid=on_paid,
        user_id=42,
        callback=send_goods,
        chat_id=-1001234567,
    )

async def on_paid(invoice, **kwargs):
    callback = kwargs.get("callback")
    user_id = kwargs.get("user_id")

    if callback:
        await callback(invoice, user_id)

    print(f"Чат для уведомления: {kwargs.get('chat_id')}")

🚀

Практический пример: магазин с DI

Давай соберём всё вместе и создадим полноценный пример магазина с использованием DI. У нас будут сервисы для работы с БД, отправки уведомлений и обработки платежей — все они зарегистрированы через DI.

Python · Полный пример с DI

import asyncio
import random
from aiosend import CryptoPay

class Database:
    """Сервис базы данных."""

    async def create_order(self, user_id, product, amount):
        order_id = random.randint(1000, 9999)
        print(f"[DB] Заказ #{order_id}: user={user_id}, product={product}, amount={amount}")
        return order_id

    async def confirm_payment(self, invoice_id, order_id):
        print(f"[DB] Платёж #{invoice_id} подтверждён для заказа #{order_id}")

class Notifier:
    """Сервис уведомлений."""

    async def notify_user(self, user_id, message):
        print(f"[NOTIFY] user={user_id}: {message}")

    async def notify_admin(self, message):
        print(f"[ADMIN] {message}")

class PaymentHandler:
    """Сервис обработки платежей."""

    async def on_successful_payment(self, invoice, db, notifier, user_id):
        await db.confirm_payment(invoice.invoice_id, invoice.payload)
        await notifier.notify_user(
            user_id,
            f"Оплата #{invoice.invoice_id} на {invoice.amount} {invoice.asset} прошла!",
        )

async def main():
    # Регистрируем все зависимости в клиенте
    cp = CryptoPay(
        token="1234:YOUR_TOKEN",
        db=Database(),
        notifier=Notifier(),
        payment_handler=PaymentHandler(),
    )

    # Хендлер с доступом к DI
    @cp.invoice_paid()
    async def on_paid(invoice, cp):
        db = cp["db"]
        notifier = cp["notifier"]
        handler = cp["payment_handler"]

        # Получаем user_id из payload
        user_id = invoice.payload or "unknown"

        await handler.on_successful_payment(invoice, db, notifier, user_id)
        await notifier.notify_admin(
            f"Поступила оплата #{invoice.invoice_id}: {invoice.amount} {invoice.asset}"
        )

    # Создаём тестовый инвойс
    invoice = await cp.create_invoice(
        amount=25,
        asset="USDT",
        description="Премиум-подписка",
        payload="user_42",
    )
    print(f"Инвойс создан: {invoice.bot_invoice_url}")

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

asyncio.run(main())

🔍 Что здесь происходит?

Мы создали три сервиса и зарегистрировали их в клиенте через kwargs. Хендлер on_paid получает cp и через него обращается к сервисам. Все сервисы легко заменить на моки для тестирования — достаточно передать другие объекты при создании клиента.

🔄

DI при использовании polling

При использовании cp.start_polling() все зарегистрированные зависимости доступны в хендлерах точно так же. Это работает и для обработчиков, назначенных через @cp.invoice_paid(), @cp.check_paid() и других декораторов.

Python · Polling с DI

import asyncio
from aiosend import CryptoPay

async def main():
    cp = CryptoPay(
        token="1234:TOKEN",
        my_config={"admin_id": 12345, "lang": "ru"},
    )

    @cp.invoice_paid()
    async def on_paid(invoice, cp):
        config = cp["my_config"]
        print(f"Уведомить админа {config['admin_id']}")
        print(f"Язык уведомления: {config['lang']}")

    @cp.check_paid()
    async def on_check_paid(check, cp):
        config = cp["my_config"]
        print(f"Чек {check.check_id} оплачен!")
        print(f"Админ для уведомления: {config['admin_id']}")

    await cp.start_polling()

asyncio.run(main())

💡 Совет: типизация зависимостей

Для удобства можно создать датакласс с типизированными полями и передавать его как единую зависимость. Это даст автодополнение в IDE:

Python · Типизированная конфигурация

from dataclasses import dataclass

@dataclass
class AppServices:
    db: Database
    notifier: Notifier
    config: dict

services = AppServices(db=Database(), notifier=Notifier(), config={})
cp = CryptoPay(token="TOKEN", services=services)

# В хендлере:
@cp.invoice_paid()
async def handler(invoice, cp):
    svc: AppServices = cp["services"]
    await svc.db.save_payment(invoice)

📌

Что важно запомнить

1️⃣

CryptoPay() принимает любые kwargs. Все они сохраняются во внутреннем хранилище и доступны через cp["key"].

2️⃣

Хендлеры получают cp. Если хендлер принимает параметр cp, aiosend автоматически передаёт объект клиента.

3️⃣

invoice.poll() передаёт kwargs. Все дополнительные аргументы попадают в хендлер как **kwargs.

4️⃣

Избегай конфликтов имён. Не используй ключи, совпадающие с атрибутами CryptoPay.

5️⃣

DI упрощает тестирование. Ты можешь подменять зависимости на моки простой заменой значения в словаре.

🎯

Практическое задание

Задание: Реализуй систему DI для магазина

Создай скрипт, который:

Определяет три сервиса: Database, TelegramNotifier, Analytics
Регистрирует их в CryptoPay через kwargs
Создаёт хендлер @cp.invoice_paid(), который через DI получает все три сервиса
При получении оплаты: сохраняет в БД, отправляет уведомление в Telegram, записывает аналитику
Использует cp.get() для безопасного доступа

Подсказка:

Python · Шаблон решения

import asyncio
from aiosend import CryptoPay

class Database:
    async def save(self, invoice):
        print(f"Сохранено: #{invoice.invoice_id}")

class TelegramNotifier:
    async def send(self, user_id, text):
        print(f"TG -> {user_id}: {text}")

class Analytics:
    async def track(self, event, data):
        print(f"Analytics: {event} {data}")

async def main():
    cp = CryptoPay(
        token="YOUR_TOKEN",
        db=Database(),
        tg=TelegramNotifier(),
        analytics=Analytics(),
    )

    @cp.invoice_paid()
    async def on_paid(invoice, cp):
        # Используй cp.get() для безопасного доступа
        db = cp.get("db")
        tg = cp.get("tg")
        analytics = cp.get("analytics")

        if db:
            await db.save(invoice)
        if tg:
            await tg.send(
                invoice.payload or "unknown",
                f"Оплачено: {invoice.amount} {invoice.asset}",
            )
        if analytics:
            await analytics.track("payment", {
                "id": invoice.invoice_id,
                "amount": invoice.amount,
            })

    await cp.start_polling()

asyncio.run(main())

Урок 7.1: Dependency Injection

5 вопросов