Урок 1.3 — Первое приложение на aiosend

Библиотека aiosend полностью асинхронна. Это значит, что все методы, работающие с API Crypto Pay, являются корутинами (асинхронными функциями) и требуют использования синтаксиса async / await. Давай разберёмся, как это работает.

Что такое async/await?

async def — это ключевое слово для объявления асинхронной функции (корутины). Внутри такой функции можно использовать await для ожидания результата другой асинхронной операции. Пока одна корутина ожидает ответа от сети, event loop может выполнять другие задачи.

import asyncio

async def say_hello():
    print("Привет!")
    await asyncio.sleep(1)  # имитация ожидания
    print("Прошла 1 секунда")

asyncio.run(say_hello())

Точка входа: asyncio.run()

Каждая асинхронная программа нуждается в точке входа — месте, где запускается event loop. В Python для этого используется asyncio.run(main()). Это стандартный паттерн для всех скриптов aiosend:

import asyncio
from aiosend import CryptoPay

async def main():
    cp = CryptoPay(token="ВАШ_ТОКЕН")
    app = await cp.get_me()
    print(f"Приложение: {app.name}")

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

Асинхронный контекстный менеджер

Хотя это и не обязательно (сессия создаётся автоматически), ты можешь использовать CryptoPay как асинхронный контекстный менеджер:

async def main():
    async with CryptoPay(token="ТОКЕН") as cp:
        app = await cp.get_me()
        print(app.name)
    # Здесь сессия автоматически закрыта

Однако в большинстве примеров мы будем использовать обычное создание клиента — это проще и не требует дополнительной вложенности.

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)

Класс CryptoPay — центральный класс библиотеки. Через него проходят все запросы к API. Мы уже подробно разбирали его в уроке 2.1, здесь же приведём краткую справку:

class CryptoPay:
    def __init__(
        self,
        token: str,                    # API-токен (обязательно)
        network: Network = MAINNET,    # MAINNET или TESTNET
        session: type[BaseSession] = AiohttpSession,
        timeout: int | float = 300,    # таймаут HTTP-запроса
        polling_config: PollingConfig = PollingConfig(),
        webhook_manager: WebhookManager | None = None,
    )

Для нашего первого приложения нам понадобится только token. Всё остальное останется по умолчанию.

Метод get_me() — самый простой метод API Crypto Pay. Он возвращает информацию о твоём приложении: его ID, название, обработчик платежей и другие данные. Это идеальный первый запрос для проверки, что клиент работает.

from aiosend import CryptoPay

cp = CryptoPay(token="ВАШ_ТОКЕН")

# Асинхронно:
app = await cp.get_me()

# Синхронно:
app = cp.get_me()

Метод возвращает объект App — Pydantic-модель с информацией о приложении. Вот как выглядят её поля:

Таблица полей объекта App

Как видишь, App содержит всего 4 поля, но они дают полную информацию о приложении. Поле supported_assets особенно полезно — оно показывает, какие криптовалюты доступны для создания инвойсов.

from aiosend import CryptoPay

cp = CryptoPay(token="ВАШ_ТОКЕН")
app = await cp.get_me()

print("Информация о приложении:")
print(f"  ID:         {app.app_id}")
print(f"  Название:   {app.name}")
print(f"  Бот:        @{app.payment_processing_bot_username}")
print(f"  Валюты:     {[a.asset_code for a in app.supported_assets]}")

# Пример вывода:
# Информация о приложении:
#   ID:         12345
#   Название:   My Crypto App
#   Бот:        @CryptoTestnetBot
#   Валюты:     ['USDT', 'TON', 'BTC', 'ETH', 'LTC', 'BNB', 'TRX', 'USDC', 'JET', 'SEND', 'SOL', 'XAUT']

В aiosend есть специальный объект .sync, который предоставляет доступ ко всем методам в синхронном виде. Это удобно, когда ты хочешь явно указать, что используешь синхронный режим:

from aiosend import CryptoPay

cp = CryptoPay(token="ВАШ_ТОКЕН")

# Явный синхронный вызов через .sync
app = cp.sync.get_me()
print(f"Приложение: {app.name} (ID: {app.app_id})")

# Такой же синхронный вызов, но без .sync (автоматическая обёртка)
app2 = cp.get_me()  # тоже работает синхронно

Разница между cp.get_me() и cp.sync.get_me():

• cp.get_me() — метод автоматически определяет контекст: если есть event loop — возвращает корутину (нужен await), иначе — выполняет синхронно.
• cp.sync.get_me() — всегда синхронный, всегда возвращает результат, никогда не требует await.

Соберём всё вместе и напишем полноценное асинхронное приложение. Этот скрипт создаёт клиента, получает информацию о приложении и выводит балансы.

import asyncio
import os
from dotenv import load_dotenv

from aiosend import CryptoPay
from aiosend.exceptions import CryptoPayError

load_dotenv()

TOKEN = os.getenv("CRYPTOPAY_TOKEN")


async def main():
    """Главная асинхронная функция приложения."""

    print("Запуск первого aiosend приложения...")
    print(f"Используется токен: {TOKEN[:8]}...{TOKEN[-4:] if TOKEN else 'None'}")

    # Создаём клиента
    cp = CryptoPay(token=TOKEN)

    try:
        # Получаем информацию о приложении
        print("\nПолучаем информацию о приложении...")
        app = await cp.get_me()

        print(f"\n{'='*50}")
        print(f"  Приложение успешно подключено!")
        print(f"  ID:          {app.app_id}")
        print(f"  Название:    {app.name}")
        print(f"  Платёжный бот: @{app.payment_processing_bot_username}")
        print(f"{'='*50}")

        # Получаем балансы
        print("\nПолучаем балансы...")
        balances = await cp.get_balance()

        if balances:
            print(f"\n{'='*50}")
            print(f"  Балансы приложения:")
            for balance in balances:
                print(f"    {balance.currency_code}:")
                print(f"      Доступно:  {balance.available}")
                print(f"      Заморожено: {balance.onhold}")
            print(f"{'='*50}")
        else:
            print("\nБалансы пусты.")

        # Показываем поддерживаемые активы
        print(f"\nПоддерживаемые криптовалюты ({len(app.supported_assets)}):")
        assets_list = [a.asset_code for a in app.supported_assets]
        print(f"  {', '.join(assets_list)}")

    except CryptoPayError as e:
        print(f"\n❌ Ошибка Crypto Pay: {e}")
    except Exception as e:
        print(f"\n❌ Неожиданная ошибка: {e}")
    finally:
        print("\nПриложение завершило работу.")


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

А вот так выглядит тот же скрипт в синхронном варианте. Обрати внимание: никакого asyncio, никаких await — просто последовательные вызовы.

import os
from dotenv import load_dotenv

from aiosend import CryptoPay
from aiosend.exceptions import CryptoPayError

load_dotenv()

TOKEN = os.getenv("CRYPTOPAY_TOKEN")


def main():
    """Главная синхронная функция приложения."""

    print("Запуск первого синхронного aiosend приложения...")

    # Создаём клиента
    cp = CryptoPay(token=TOKEN)

    try:
        # Получаем информацию о приложении (без await!)
        print("\nПолучаем информацию о приложении...")
        app = cp.get_me()

        print(f"\n{'='*50}")
        print(f"  Приложение успешно подключено!")
        print(f"  ID:          {app.app_id}")
        print(f"  Название:    {app.name}")
        print(f"  Платёжный бот: @{app.payment_processing_bot_username}")
        print(f"{'='*50}")

        # Получаем балансы (без await!)
        print("\nПолучаем балансы...")
        balances = cp.get_balance()

        if balances:
            print(f"\n{'='*50}")
            print(f"  Балансы приложения:")
            for b in balances:
                print(f"    {b.currency_code}:")
                print(f"      Доступно:  {b.available}")
                print(f"      Заморожено: {b.onhold}")
            print(f"{'='*50}")
        else:
            print("\nБалансы пусты.")

        # Поддерживаемые активы
        print(f"\nПоддерживаемые криптовалюты ({len(app.supported_assets)}):")
        print(f"  {', '.join(a.asset_code for a in app.supported_assets)}")

    except CryptoPayError as e:
        print(f"\n❌ Ошибка Crypto Pay: {e}")
    except Exception as e:
        print(f"\n❌ Неожиданная ошибка: {e}")
    finally:
        print("\nПриложение завершило работу.")


if __name__ == "__main__":
    main()

В реальных проектах токены и конфиденциальные данные не хранят в коде. Вместо этого используют переменные окружения и файл .env. Давай разберём правильную структуру проекта.

Рекомендуемая структура файлов

my_crypto_bot/
├── .env                    # Токены и конфигурация (НЕ в git!)
├── .env.example            # Шаблон .env (в git, без токенов)
├── .gitignore              # .env добавлен в игнор
├── main.py                 # Точка входа
├── config.py               # Чтение переменных окружения
├── requirements.txt        # Зависимости (aiosend, python-dotenv)
└── README.md               # Документация

Файл .env

В файле .env хранятся все настройки, специфичные для окружения:

# Токены Crypto Pay
CRYPTOPAY_TOKEN=1234:ABCdefGHIjklmNOPqrstUVwxyz
CRYPTOPAY_TEST_TOKEN=5678:XYZabcDEFghijklMNOpqrstu

# Настройки таймаутов
CRYPTOPAY_TIMEOUT=30
CRYPTOPAY_POLLING_TIMEOUT=300
CRYPTOPAY_POLLING_DELAY=2

# Режим отладки
DEBUG=true

python-dotenv — загрузка переменных

Библиотека python-dotenv загружает переменные из .env в os.environ. Установка:

pip install python-dotenv
# или
poetry add python-dotenv
# или
echo "python-dotenv>=1.0" >> requirements.txt

import os
from dotenv import load_dotenv

load_dotenv()


class Config:
    """Конфигурация приложения из переменных окружения."""

    # Токены
    CRYPTOPAY_TOKEN: str | None = os.getenv("CRYPTOPAY_TOKEN")
    CRYPTOPAY_TEST_TOKEN: str | None = os.getenv("CRYPTOPAY_TEST_TOKEN")

    # Таймауты
    TIMEOUT: int = int(os.getenv("CRYPTOPAY_TIMEOUT", "30"))
    POLLING_TIMEOUT: int = int(os.getenv("CRYPTOPAY_POLLING_TIMEOUT", "300"))
    POLLING_DELAY: int = int(os.getenv("CRYPTOPAY_POLLING_DELAY", "2"))

    # Режим
    DEBUG: bool = os.getenv("DEBUG", "false").lower() == "true"

    @classmethod
    def validate(cls):
        """Проверяет, что обязательные переменные установлены."""
        if not cls.CRYPTOPAY_TOKEN:
            raise ValueError(
                "CRYPTOPAY_TOKEN не установлен. "
                "Создайте .env файл на основе .env.example"
            )


# Проверяем конфигурацию при импорте
Config.validate()

.gitignore — защита секретов

Обязательно добавь .env в .gitignore, чтобы случайно не закоммитить токены в репозиторий:

# Файлы с секретами
.env
.env.local

# Виртуальное окружение
venv/
.venv/
__pycache__/
*.pyc

# IDE
.vscode/
.idea/

При запуске первого приложения ты можешь столкнуться с несколькими типичными ошибками. Рассмотрим каждую и способ её решения.

# Ошибка:
cp = CryptoPay(token="1234:INVALID_TOKEN")
app = await cp.get_me()
# → aiosend.exceptions.APIError: [401] Unauthorized

# Решение:
# 1. Проверьте токен в @CryptoBot или @CryptoTestnetBot
# 2. Создайте новый токен, если старый отозван
# 3. Убедитесь, что используете правильную сеть

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

# Решение: укажите правильную сеть
cp = CryptoPay(token="MAINNET_TOKEN", network=MAINNET)

# Ошибка:
cp = CryptoPay(token="TOKEN", timeout=5)  # слишком мало
app = await cp.get_me()
# → APITimeoutError: Request exceeded timeout of 5 seconds

# Решение: увеличьте таймаут
cp = CryptoPay(token="TOKEN", timeout=30)

# Ошибка:
TOKEN = os.getenv("CRYPTOPAY_TOKEN")
# TOKEN = None, потому что .env не загружен
cp = CryptoPay(token=TOKEN)
# → ValueError: token must be a string

# Решение: вызовите load_dotenv() перед чтением переменных
load_dotenv()
TOKEN = os.getenv("CRYPTOPAY_TOKEN")

Всегда оборачивай запуск приложения в try/except для обработки этих ошибок. Вот универсальный шаблон:

import asyncio
from aiosend import CryptoPay
from aiosend.exceptions import (
    APIError,
    APITimeoutError,
    WrongNetworkError,
)

async def main():
    try:
        cp = CryptoPay(token="ТОКЕН")
        app = await cp.get_me()
        print(f"Успех: {app.name}")
    except WrongNetworkError as e:
        print(f"Неверная сеть: {e}")
    except APIError as e:
        print(f"Ошибка API [{e.code}]: {e.name}")
    except APITimeoutError as e:
        print(f"Таймаут: {e}")
    except Exception as e:
        print(f"Неизвестная ошибка: {e}")

asyncio.run(main())

Финальный пример — правильно организованное приложение с разделением на модули. Этот шаблон можно использовать как основу для любого проекта на aiosend.

import asyncio
import logging

from aiosend import CryptoPay
from aiosend.exceptions import CryptoPayError

from config import Config

# Настройка логирования
logging.basicConfig(
    level=logging.DEBUG if Config.DEBUG else logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)


def create_client() -> CryptoPay:
    """Создаёт и возвращает настроенный клиент CryptoPay."""
    token = (
        Config.CRYPTOPAY_TEST_TOKEN
        if Config.DEBUG
        else Config.CRYPTOPAY_TOKEN
    )
    logger.info(
        "Создание клиента для %s",
        "TESTNET" if Config.DEBUG else "MAINNET",
    )
    return CryptoPay(
        token=token,
        timeout=Config.TIMEOUT,
    )


async def main():
    """Главная функция приложения."""
    logger.info("Запуск приложения aiosend")

    cp = create_client()

    try:
        # Проверяем подключение
        app = await cp.get_me()
        logger.info(
            "Подключено: %s (ID: %d, бот: @%s)",
            app.name,
            app.app_id,
            app.payment_processing_bot_username,
        )

        # Получаем баланс
        balances = await cp.get_balance()
        for b in balances:
            logger.info(
                "Баланс %s: доступно %s, заморожено %s",
                b.currency_code,
                b.available,
                b.onhold,
            )

    except CryptoPayError as e:
        logger.error("Ошибка Crypto Pay: %s", e)
        return 1
    except Exception as e:
        logger.exception("Неожиданная ошибка: %s", e)
        return 1

    logger.info("Приложение завершило работу успешно")
    return 0


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

После написания первого приложения полезно знать, как проверить его работу и диагностировать возможные проблемы. Вот чек-лист для первого запуска: