Урок 2.1 — Создание клиента и работа с сетями

Класс CryptoPay — это главный класс библиотеки. Он наследуется от нескольких миксинов, которые добавляют ему методы API, инструменты (tools), обработку вебхуков и polling. Конструктор принимает следующие параметры:

from aiosend import CryptoPay

# Минимальный вариант — только токен
cp = CryptoPay(token="1234:ABCdefGHIjklmNOPqrstUVwxyz")

# Всё остальное — по умолчанию:
# network = MAINNET
# session = AiohttpSession
# timeout = 300 секунд
# polling_config = PollingConfig(timeout=300, delay=2)
# webhook_manager = None

Важно понимать: session передаётся как класс (не объект), потому что aiosend внутри создаёт экземпляр сессии, передавая ей network и timeout. Это сделано для гибкости — вы можете написать свою сессию на httpx, requests или любой другой библиотеке.

from aiosend import CryptoPay, MAINNET
from aiosend.client.session import AiohttpSession
from aiosend.polling import PollingConfig

cp = CryptoPay(
    token="1234:ABCdefGHIjklmNOPqrstUVwxyz",
    network=MAINNET,
    session=AiohttpSession,
    timeout=120,            # таймаут 2 минуты
    polling_config=PollingConfig(
        timeout=600,        # ждём оплату 10 минут
        delay=3,            # проверяем каждые 3 секунды
    ),
)

Crypto Pay API предоставляет две среды для работы: основную (MAINNET) и тестовую (TESTNET). Они полностью изолированы: токены, балансы, инвойсы, чеки — всё разделено.

from aiosend import CryptoPay, MAINNET

# Можно явно указать MAINNET
cp_main = CryptoPay(
    token="1234:MAINNET_TOKEN",
    network=MAINNET,
)
# Это эквивалентно:
cp_main2 = CryptoPay(token="1234:MAINNET_TOKEN")  # MAINNET по умолчанию

from aiosend import CryptoPay, TESTNET

cp_test = CryptoPay(
    token="1234:TESTNET_TOKEN",
    network=TESTNET,
)

Если вы попытаетесь использовать MAINNET-токен с TESTNET-сетью (или наоборот), библиотека выбросит исключение WrongNetworkError. Это защита от случайного использования не той сети.

from aiosend import CryptoPay, TESTNET

# MAINNET-токен с TESTNET-сетью → WrongNetworkError
cp = CryptoPay(
    token="1234:MAINNET_TOKEN",  # получен в @CryptoBot
    network=TESTNET,              # а сеть — тестовая!
)
# → aiosend.exceptions.WrongNetworkError:
# "Authorization failed. Token is served by the MAINNET,
#  you are using TESTNET"

Какой сетью пользоваться? Вот простая логика:

• TESTNET — для разработки и отладки. Тестовые токены, тестовые балансы, актив JET для тестов. Всё бесплатно.
• MAINNET — для продакшена. Реальные деньги, реальные пользователи, реальные транзакции.
• Начинайте с TESTNET, отладьте всё, и только потом переходите на MAINNET.

За сетью в aiosend стоит простой dataclass Network, который содержит два поля: name и base. MAINNET и TESTNET — это просто два предсозданных экземпляра этого класса.

from dataclasses import dataclass

@dataclass(frozen=True)
class Network:
    name: str   # имя сети: "MAINNET" или "TESTNET"
    base: str   # base URL с {method} в качестве плейсхолдера

    def url(self, method) -> str:
        return self.base.format(method=method.__method__)

# Предопределённые экземпляры:
MAINNET = Network(
    name="MAINNET",
    base="https://pay.crypt.bot/api/{method}",
)
TESTNET = Network(
    name="TESTNET",
    base="https://testnet-pay.crypt.bot/api/{method}",
)

Что это даёт? Вы можете создать свою собственную сеть с кастомным URL. Например, если вы используете прокси, зеркало API или свой эндпоинт:

from aiosend import CryptoPay
from aiosend.client.network import Network

# Кастомная сеть через прокси-сервер
PROXY_NET = Network(
    name="PROXY_MAINNET",
    base="https://my-proxy.example.com/cryptobot/api/{method}",
)

cp = CryptoPay(
    token="1234:TOKEN",
    network=PROXY_NET,
)

Класс Network также предоставляет полезные методы:

from aiosend import CryptoPay

cp = CryptoPay(token="TOKEN")

# Получаем QR-код для инвойса
invoice = await cp.create_invoice(amount=10, asset="USDT")
qr_url = cp.session.network.get_qr(invoice.bot_invoice_url)
print(qr_url)
# → https://qr.crypt.bot/?url=https://pay.crypt.bot/...

# Получаем URL сети
print(cp.session.network.name)  # "MAINNET" или "TESTNET"
print(cp.session.network.base)  # "https://pay.crypt.bot/api/{method}"

Параметр session определяет, как именно aiosend будет отправлять HTTP-запросы к API. По умолчанию используется AiohttpSession — обёртка над aiohttp.ClientSession. Но вы можете передать свой класс сессии.

AiohttpSession — стандартная сессия

AiohttpSession — это встроенная сессия на базе популярной библиотеки aiohttp. Она используется по умолчанию и подходит для 99% задач. Вот как она работает внутри:

import ssl
import certifi
from aiohttp import ClientSession, ClientTimeout, TCPConnector

class AiohttpSession(BaseSession):
    def __init__(self, network: Network, timeout: float = 300):
        super().__init__(network, timeout)
        self._session: ClientSession | None = None

    async def request(self, token, client, method):
        ssl_context = ssl.create_default_context(cafile=certifi.where())
        self._session = ClientSession(
            timeout=ClientTimeout(self.timeout),
            connector=TCPConnector(ssl_context=ssl_context),
        )
        async with self._session as session:
            resp = await session.post(
                url=self.network.url(method),
                data=method.model_dump_json(exclude_none=True),
                headers={
                    "Crypto-Pay-API-Token": token,
                    "Content-Type": "application/json",
                    "User-Agent": f"Python/3.x aiohttp/x aiosend/3.x",
                },
            )
            response = self._check_response(client, method, await resp.text())
        return response.result

BaseSession — абстрактный базовый класс

Если вы хотите написать свою сессию (например, на httpx или requests), вам нужно унаследоваться от BaseSession и реализовать метод request. Вот минимальный шаблон:

from aiosend.client.session.base import BaseSession
from aiosend.client.network import Network
import httpx

class HttpxSession(BaseSession):
    def __init__(self, network: Network, timeout: float = 300):
        super().__init__(network, timeout)
        self._client = httpx.AsyncClient(timeout=timeout)

    async def request(self, token, client, method):
        url = self.network.url(method)
        headers = {
            "Crypto-Pay-API-Token": token,
            "Content-Type": "application/json",
        }
        data = method.model_dump_json(exclude_none=True)
        resp = await self._client.post(url, content=data, headers=headers)
        return self._check_response(client, method, resp.text)

# Использование:
cp = CryptoPay(
    token="TOKEN",
    session=HttpxSession,  # передаём КЛАСС, не объект!
    timeout=60,
)

Параметр timeout

Таймаут — это максимальное время ожидания ответа от сервера в секундах. По умолчанию — 300 секунд (5 минут). Если сервер не отвечает за это время, выбрасывается исключение APITimeoutError.

# Таймаут 10 секунд — для быстрых запросов
cp_fast = CryptoPay(token="TOKEN", timeout=10)

# Таймаут 600 секунд — для длительных операций
cp_slow = CryptoPay(token="TOKEN", timeout=600)

# Таймаут 30 секунд — хороший баланс
cp_balanced = CryptoPay(token="TOKEN", timeout=30)

Два дополнительных параметра конструктора отвечают за обработку событий. Мы подробно разберём их в Модуле 6, но кратко познакомимся уже сейчас.

PollingConfig

PollingConfig — это dataclass с двумя полями, который определяет, как часто и как долго aiosend будет опрашивать API на предмет обновлений (например, оплаты инвойса).

from aiosend import CryptoPay
from aiosend.polling import PollingConfig

cp = CryptoPay(
    token="TOKEN",
    polling_config=PollingConfig(
        timeout=600,  # ждать оплату 10 минут
        delay=3,      # проверять каждые 3 секунды
    ),
)

WebhookManager

WebhookManager — это класс для приёма вебхуков (уведомлений об оплате) от Crypto Pay API. aiosend поддерживает несколько менеджеров из коробки:

from aiohttp import web
from aiosend import CryptoPay
from aiosend.webhook import AiohttpManager

app = web.Application()
manager = AiohttpManager(app, path="/webhook")

cp = CryptoPay(
    token="TOKEN",
    webhook_manager=manager,
)

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

if __name__ == "__main__":
    web.run_app(app, host="0.0.0.0", port=8080)

В реальных проектах часто нужно работать с несколькими приложениями или сетями одновременно. Например:

• Два приложения в MAINNET (для разных проектов)
• Одно приложение в MAINNET, другое в TESTNET (разработка + продакшен)
• Несколько приложений в TESTNET (для тестирования разных сценариев)

Никаких ограничений на количество клиентов нет. Создавайте столько, сколько нужно:

from aiosend import CryptoPay, MAINNET, TESTNET

# Клиент для продакшена (MAINNET)
cp_prod = CryptoPay(
    token="1234:PROD_TOKEN",
    network=MAINNET,
)

# Клиент для тестирования (TESTNET)
cp_test = CryptoPay(
    token="5678:TEST_TOKEN",
    network=TESTNET,
)

# Клиент для второго приложения (MAINNET)
cp_prod_2 = CryptoPay(
    token="9012:ANOTHER_PROD_TOKEN",
    network=MAINNET,
)

# Используем их независимо
async def main():
    app_prod = await cp_prod.get_me()
    app_test = await cp_test.get_me()
    print(f"Prod app: {app_prod.name}")
    print(f"Test app: {app_test.name}")

Удобный приём — хранить токены в переменных окружения и создавать клиенты через фабричную функцию:

import os
from aiosend import CryptoPay, MAINNET, TESTNET

class ClientFactory:
    """Фабрика для создания клиентов."""

    @staticmethod
    def create_mainnet(token_env: str = "CRYPTOPAY_TOKEN") -> CryptoPay:
        token = os.getenv(token_env)
        if not token:
            raise ValueError(f"Переменная {token_env} не установлена")
        return CryptoPay(token=token, network=MAINNET)

    @staticmethod
    def create_testnet(token_env: str = "CRYPTOPAY_TEST_TOKEN") -> CryptoPay:
        token = os.getenv(token_env)
        if not token:
            raise ValueError(f"Переменная {token_env} не установлена")
        return CryptoPay(token=token, network=TESTNET)

# Использование:
# export CRYPTOPAY_TOKEN="1234:PROD_TOKEN"
# export CRYPTOPAY_TEST_TOKEN="5678:TEST_TOKEN"

cp = ClientFactory.create_mainnet()
cp_test = ClientFactory.create_testnet()

Когда вы создаёте CryptoPay(token="..."), в конструкторе вызывается метод __auth(), который делает тестовый запрос к API (по сути — getMe) для проверки токена. Вот как это работает:

def __auth(self):
    try:
        # Пробуем авторизоваться в указанной сети
        me = token_validate(self, self.session.network)
        print(f"Авторизован как '{me.name}' id={me.app_id} на {self.network}")
    except APIError:
        # Если ошибка — возможно, токен от другой сети
        current_net = self.session.network
        # Переключаемся на противоположную сеть
        self.session = self.session.__class__(
            TESTNET if current_net == MAINNET else MAINNET,
            self.session.timeout,
        )
        # Пробуем снова
        token_validate(self, self.session.network)
        # Если получилось — значит токен от другой сети
        raise WrongNetworkError(
            f"Token is served by the {self.session.network.name}, "
            f"you are using {current_net.name}"
        )

То есть aiosend автоматически проверяет токен при создании клиента. Если токен не подходит к выбранной сети — библиотека даже пытается переключиться на другую сеть, чтобы дать вам понятное сообщение об ошибке.

aiosend поддерживает оба подхода. По умолчанию все методы — асинхронные. Но библиотека автоматически создаёт синхронные обёртки для всех публичных методов (кроме служебных, начинающихся с _).

import asyncio
from aiosend import CryptoPay

async def main():
    cp = CryptoPay("TOKEN")
    app = await cp.get_me()
    print(app.name)

asyncio.run(main())

from aiosend import CryptoPay

cp = CryptoPay("TOKEN")
app = cp.get_me()  # без await!
print(app.name)

Синхронный режим работает благодаря декоратору syncify, который применяется ко всем публичным асинхронным методам. Если внутри async-функции есть запущенный event loop, синхронный .get_me() вернёт корутину (нужно await). Если цикла нет — запустит свой и дождётся результата.

Давайте соберём всё вместе. Напишем скрипт, который создаёт двух клиентов (MAINNET и TESTNET), проверяет их статус и выводит информацию.

import asyncio
import os
from dotenv import load_dotenv

from aiosend import CryptoPay, MAINNET, TESTNET
from aiosend.exceptions import WrongNetworkError

load_dotenv()

async def main():
    # Создаём клиентов
    cp_prod = CryptoPay(
        token=os.getenv("CRYPTOPAY_PROD_TOKEN"),
        network=MAINNET,
    )
    cp_test = CryptoPay(
        token=os.getenv("CRYPTOPAY_TEST_TOKEN"),
        network=TESTNET,
        timeout=30,  # быстрые тестовые запросы
    )

    # Получаем информацию о приложениях
    prod_app = await cp_prod.get_me()
    test_app = await cp_test.get_me()

    print("=" * 50)
    print("MAINNET App:")
    print(f"  ID:   {prod_app.app_id}")
    print(f"  Name: {prod_app.name}")
    print(f"  Bot:  @{prod_app.payment_processing_bot_username}")
    print("=" * 50)
    print("TESTNET App:")
    print(f"  ID:   {test_app.app_id}")
    print(f"  Name: {test_app.name}")
    print(f"  Bot:  @{test_app.payment_processing_bot_username}")
    print("=" * 50)

    # Получаем балансы
    prod_balances = await cp_prod.get_balance()
    test_balances = await cp_test.get_balance()

    print("\nMAINNET Balances:")
    for b in prod_balances:
        print(f"  {b.currency_code}: available={b.available}, onhold={b.onhold}")

    print("\nTESTNET Balances:")
    for b in test_balances:
        print(f"  {b.currency_code}: available={b.available}, onhold={b.onhold}")

    # Проверяем, какую сеть использует каждый клиент
    print(f"\nСеть prod: {cp_prod.session.network.name}")
    print(f"Сеть test: {cp_test.session.network.name}")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except WrongNetworkError as e:
        print(f"Ошибка сети: {e}")
    except Exception as e:
        print(f"Ошибка: {e}")

Рассмотрим самые частые проблемы, с которыми сталкиваются новички при создании CryptoPay клиента, и пути их решения.

# Ошибка:
cp = CryptoPay("MAINNET_TOKEN", network=TESTNET)
# → WrongNetworkError: Authorization failed. Token is served by the MAINNET,
#   you are using TESTNET

# Исправление:
cp = CryptoPay("MAINNET_TOKEN", network=MAINNET)

# Ошибка:
cp = CryptoPay("1234:NEVERNEVER")
# → APIError: [401] /getMe: Unauthorized

# Решение: проверьте токен в @CryptoBot, создайте новый если нужно

# Ошибка:
cp = CryptoPay("TOKEN", timeout=5)  # слишком мало
# → APITimeoutError: Request to /getMe has exceeded the timeout of 5 seconds

# Исправление: увеличьте таймаут
cp = CryptoPay("TOKEN", timeout=30)

# Ошибка:
from aiosend.client.session import AiohttpSession
session = AiohttpSession(MAINNET)
cp = CryptoPay("TOKEN", session=session)  # TypeError!

# Исправление: передавайте класс, не объект
cp = CryptoPay("TOKEN", session=AiohttpSession)

# Ошибка:
class MySession(BaseSession):
    def __init__(self, network, timeout):
        pass  # забыли super().__init__()
    # → AttributeError: 'MySession' object has no attribute 'network'

# Исправление:
class MySession(BaseSession):
    def __init__(self, network, timeout):
        super().__init__(network, timeout)  # обязательно!

Несколько рекомендаций по организации работы с клиентом aiosend в реальных проектах:

project/
├── .env                  # токены и конфигурация
├── .env.example          # шаблон .env (без токенов!)
├── app/
│   ├── __init__.py
│   ├── client.py         # создание и конфигурация клиента
│   ├── handlers.py       # обработчики
│   └── config.py         # чтение .env
└── main.py               # точка входа

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    CRYPTOPAY_TOKEN = os.getenv("CRYPTOPAY_TOKEN")
    CRYPTOPAY_TEST_TOKEN = os.getenv("CRYPTOPAY_TEST_TOKEN")
    CRYPTOPAY_TIMEOUT = int(os.getenv("CRYPTOPAY_TIMEOUT", "30"))
    CRYPTOPAY_POLLING_TIMEOUT = int(os.getenv("CRYPTOPAY_POLLING_TIMEOUT", "300"))
    CRYPTOPAY_POLLING_DELAY = int(os.getenv("CRYPTOPAY_POLLING_DELAY", "2"))
    DEBUG = os.getenv("DEBUG", "false").lower() == "true"

from aiosend import CryptoPay, MAINNET, TESTNET
from aiosend.polling import PollingConfig
from config import Config

def create_client(
    use_testnet: bool = False,
) -> CryptoPay:
    """Создаёт и возвращает настроенный клиент CryptoPay."""
    token = (
        Config.CRYPTOPAY_TEST_TOKEN
        if use_testnet
        else Config.CRYPTOPAY_TOKEN
    )
    network = TESTNET if use_testnet else MAINNET

    return CryptoPay(
        token=token,
        network=network,
        timeout=Config.CRYPTOPAY_TIMEOUT,
        polling_config=PollingConfig(
            timeout=Config.CRYPTOPAY_POLLING_TIMEOUT,
            delay=Config.CRYPTOPAY_POLLING_DELAY,
        ),
    )

# Пример использования:
# cp = create_client(use_testnet=True)  # для тестов
# cp = create_client()                   # для продакшена