Урок 2.5 — Обработка ошибок и исключения

Все исключения в aiosend наследуются от базового класса CryptoPayError. Это позволяет легко перехватывать любую ошибку библиотеки одним except-блоком или обрабатывать конкретные типы отдельно.

Exception
  └── CryptoPayError                    # базовое исключение aiosend
        ├── APIError                    # ошибка от Crypto Pay API (код + название)
        │     ├── 401 Unauthorized      # неверный или отозванный токен
        │     ├── 403 Not Allowed       # метод недоступен для приложения
        │     ├── 404 Not Found         # ресурс не найден
        │     └── 429 Too Many Requests # превышение лимита запросов
        ├── APITimeoutError             # превышен таймаут HTTP-запроса
        ├── WrongNetworkError           # токен не соответствует выбранной сети
        ├── MethodValuesError           # неверные значения параметров метода
        └── DeserializationError        # ошибка разбора ответа API

APIError — самое частое исключение. Оно выбрасывается, когда Crypto Pay API возвращает HTTP-ошибку. Содержит два важных атрибута: code (HTTP-статус) и name (текст ошибки).

from aiosend import CryptoPay
from aiosend.exceptions import APIError

cp = CryptoPay(token="TOKEN")

try:
    invoice = await cp.create_invoice(amount=100, asset="USDT")
except APIError as e:
    print(f"Ошибка API [{e.code}]: {e.name}")
    # e.code   → 401, 403, 404, 422, 429
    # e.name   → "Unauthorized", "Not Found" и т.д.
    # str(e)   → полное сообщение

    if e.code == 401:
        print("Требуется новый токен!")
    elif e.code == 403:
        print("Метод не разрешён для этого приложения")
    elif e.code == 429:
        print("Слишком много запросов, нужна пауза")
    elif e.code == 404:
        print("Ресурс не найден")
    else:
        print(f"Неизвестная ошибка API: {e}")

APITimeoutError выбрасывается, когда HTTP-запрос к Crypto Pay API превышает установленный таймаут. Это может произойти из-за проблем с сетью, медленного ответа сервера или слишком маленького значения таймаута.

from aiosend import CryptoPay
from aiosend.exceptions import APITimeoutError

# Слишком маленький таймаут
cp = CryptoPay(token="TOKEN", timeout=2)

try:
    app = await cp.get_me()
except APITimeoutError as e:
    print(f"Запрос превысил таймаут: {e}")
    # → "Request to getMe has exceeded the timeout of 2 seconds"
    # Стратегия: увеличить таймаут и повторить
    cp2 = CryptoPay(token="TOKEN", timeout=30)
    app = await cp2.get_me()

Настройка таймаута для конкретного запроса

Таймаут задаётся при создании клиента и применяется ко всем его запросам. Если нужно изменить таймаут для одного конкретного вызова — создай временного клиента:

from aiosend import CryptoPay

# Основной клиент — быстрые операции
cp_fast = CryptoPay(token="TOKEN", timeout=10)

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

# Используем
invoice = await cp_slow.create_invoice(
    amount=1000, asset="TON", expires_in=86400
)
balances = await cp_fast.get_balance()

WrongNetworkError — неверная сеть

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

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

try:
    # MAINNET-токен с TESTNET-сетью
    cp = CryptoPay(token="1234:MAINNET_TOKEN", network=TESTNET)
except WrongNetworkError as e:
    print(f"Ошибка сети: {e}")
    # → "Authorization failed. Token is served by the MAINNET,
    #    you are using TESTNET"
    # Исправляем сеть
    cp = CryptoPay(token="1234:MAINNET_TOKEN", network=MAINNET)

MethodValuesError — неверные параметры

Выбрасывается при попытке передать недопустимые значения в методы API. Например, неверный тип валюты, отрицательную сумму или неподдерживаемый active:

from aiosend import CryptoPay
from aiosend.exceptions import MethodValuesError

cp = CryptoPay(token="TOKEN")

try:
    # Неверный актив (FAKE не существует)
    invoice = await cp.create_invoice(amount=100, asset="FAKE")
except MethodValuesError as e:
    print(f"Неверное значение параметра: {e}")
    # → "Invalid value for 'asset': 'FAKE' is not a valid Asset"

try:
    # Неверный paid_btn_name
    invoice = await cp.create_invoice(
        amount=100,
        asset="USDT",
        paid_btn_name="invalidButton",
    )
except MethodValuesError as e:
    print(f"Неверное значение: {e}")

DeserializationError — ошибка разбора ответа

Редкое исключение, которое выбрасывается, когда aiosend не может разобрать ответ от API. Обычно это свидетельствует о проблемах на стороне сервера или несовместимости версий:

from aiosend import CryptoPay
from aiosend.exceptions import DeserializationError

cp = CryptoPay(token="TOKEN")

try:
    app = await cp.get_me()
except DeserializationError as e:
    print(f"Ошибка разбора ответа: {e}")
    # Обычно означает проблему с API или версией библиотеки
    # Рекомендуется обновить aiosend:
    # pip install --upgrade aiosend

Пример 1: Базовая обработка с finally

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

async def main():
    cp = CryptoPay(token="TOKEN", timeout=15)

    try:
        print("Получаем информацию о приложении...")
        app = await cp.get_me()
        print(f"✓ Успех: {app.name}")

        print("Создаём инвойс...")
        invoice = await cp.create_invoice(
            amount=100, asset="USDT",
        )
        print(f"✓ Инвойс #{invoice.invoice_id} создан")

    except APITimeoutError:
        print("✗ Таймаут запроса. Возможно проблемы с сетью.")
    except APIError as e:
        print(f"✗ Ошибка API [{e.code}]: {e.name}")
    except CryptoPayError as e:
        print(f"✗ Ошибка aiosend: {e}")
    except Exception as e:
        print(f"✗ Неожиданная ошибка: {e}")
    finally:
        print("✓ Завершение работы")

asyncio.run(main())

Пример 2: Обработка ошибок создания инвойса

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

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


async def create_invoice_safe(
    cp: CryptoPay,
    amount: float,
    asset: str,
) -> Invoice | None:
    """Безопасное создание инвойса с обработкой всех ошибок."""
    try:
        invoice = await cp.create_invoice(
            amount=amount,
            asset=asset,
            description="Тестовый платёж",
        )
        logger.info(
            "Создан инвойс #%d на %s %s",
            invoice.invoice_id, invoice.amount, invoice.asset,
        )
        return invoice

    except MethodValuesError as e:
        logger.error("Неверные параметры: %s", e)
    except APIError as e:
        logger.error("Ошибка API [%d]: %s", e.code, e.name)
        if e.code == 429:
            logger.warning("Rate limit exceeded, retry later")
    except APITimeoutError as e:
        logger.error("Таймаут при создании инвойса: %s", e)
    except Exception as e:
        logger.exception("Неожиданная ошибка: %s", e)

    return None


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

    invoice = await create_invoice_safe(cp, 100, "USDT")
    if invoice:
        print(f"Ссылка на оплату: {invoice.bot_invoice_url}")
    else:
        print("Не удалось создать инвойс")

asyncio.run(main())

Пример 3: Retry-стратегия с exponential backoff

import asyncio
import logging
from aiosend import CryptoPay
from aiosend.exceptions import APIError, APITimeoutError

logger = logging.getLogger(__name__)


async def with_retry(
    cp: CryptoPay,
    method,
    max_retries: int = 3,
    base_delay: float = 1.0,
    **kwargs,
):
    """Универсальная функция с повторными попытками."""
    last_exception = None

    for attempt in range(1, max_retries + 1):
        try:
            return await method(**kwargs)
        except APITimeoutError as e:
            logger.warning(
                "Таймаут (попытка %d/%d): %s",
                attempt, max_retries, e,
            )
            last_exception = e
        except APIError as e:
            if e.code == 429:
                delay = base_delay * (2 ** (attempt - 1))
                logger.warning(
                    "Rate limit (попытка %d/%d), "
                    "повтор через %.1fс...",
                    attempt, max_retries, delay,
                )
                await asyncio.sleep(delay)
                last_exception = e
                continue
            else:
                raise

        if attempt < max_retries:
            delay = base_delay * (2 ** (attempt - 1))
            await asyncio.sleep(delay)

    raise last_exception


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

    try:
        app = await with_retry(
            cp,
            cp.get_me,
            max_retries=3,
            base_delay=1.0,
        )
        print(f"Приложение: {app.name}")
    except Exception as e:
        print(f"Не удалось получить данные после 3 попыток: {e}")

asyncio.run(main())

Пример 4: Логирование ошибок с контекстом

import asyncio
import logging
import json
from datetime import datetime
from aiosend import CryptoPay
from aiosend.exceptions import CryptoPayError

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("payment")


class PaymentError(Exception):
    """Кастомное исключение для бизнес-логики."""
    pass


async def process_payment(
    cp: CryptoPay,
    amount: float,
    asset: str,
    user_id: int,
):
    """Обработка платежа с расширенным логированием."""
    context = {
        "user_id": user_id,
        "amount": amount,
        "asset": asset,
        "timestamp": datetime.utcnow().isoformat(),
    }

    logger.info("Начало обработки платежа: %s", json.dumps(context))

    try:
        invoice = await cp.create_invoice(
            amount=amount,
            asset=asset,
            payload=f"user_{user_id}",
            description=f"Платёж пользователя #{user_id}",
        )

        context["invoice_id"] = invoice.invoice_id
        logger.info(
            "Инвойс создан: %s",
            json.dumps(context, indent=2),
        )
        return invoice

    except CryptoPayError as e:
        context["error"] = str(e)
        context["error_type"] = type(e).__name__
        logger.error(
            "Ошибка платежа: %s",
            json.dumps(context, indent=2),
        )

        if hasattr(e, "code"):
            context["api_code"] = e.code

        raise PaymentError(
            f"Платёж не удался: {e}"
        ) from e


async def main():
    cp = CryptoPay(token="TOKEN")
    try:
        invoice = await process_payment(
            cp, 50, "USDT", user_id=12345,
        )
        print(f"Счёт: {invoice.bot_invoice_url}")
    except PaymentError as e:
        print(f"Ошибка: {e}")

asyncio.run(main())

Правильное логирование — ключ к отладке и мониторингу приложения. aiosend использует стандартный модуль logging. Рекомендуется настроить логирование в каждом проекте:

import logging

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging.StreamHandler(),                          # вывод в консоль
        logging.FileHandler("app_errors.log"),            # запись в файл
    ],
)

logger = logging.getLogger("aiosend_app")

# Пример использования
try:
    app = await cp.get_me()
except Exception as e:
    logger.exception("Ошибка при получении информации о приложении")
    # logger.exception() автоматически добавляет traceback

Логирование с дополнительным контекстом

Для более информативных логов добавляй контекст: ID пользователя, тип операции, параметры запроса:

import logging

logger = logging.getLogger(__name__)

async def create_invoice_with_logging(
    cp, amount, asset, user_id
):
    logger.info(
        "Создание инвойса: user=%d, amount=%s, asset=%s",
        user_id, amount, asset,
    )
    try:
        invoice = await cp.create_invoice(
            amount=amount, asset=asset,
            payload=str(user_id),
        )
        logger.info(
            "Инвойс #%d создан для user=%d",
            invoice.invoice_id, user_id,
        )
        return invoice
    except Exception as e:
        logger.error(
            "Ошибка создания инвойса: user=%d, error=%s",
            user_id, e,
        )
        raise

Не все ошибки фатальны. Некоторые (особенно 429 Too Many Requests и APITimeoutError) могут быть временными. Для таких случаев используй retry-стратегии.

Простая retry-функция с exponential backoff

import asyncio
import random
from functools import wraps
from aiosend.exceptions import APIError, APITimeoutError


def retry(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    retryable_errors: tuple = (APITimeoutError,),
):
    """Декоратор для повторных попыток с exponential backoff."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(1, max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except retryable_errors as e:
                    last_exception = e
                    delay = min(
                        base_delay * (2 ** (attempt - 1)),
                        max_delay,
                    )
                    jitter = random.uniform(0, delay * 0.1)
                    total_delay = delay + jitter
                    print(
                        f"Попытка {attempt}/{max_retries} "
                        f"через {total_delay:.1f}с: {e}"
                    )
                    await asyncio.sleep(total_delay)
                except APIError as e:
                    if e.code == 429:
                        delay = min(
                            base_delay * (2 ** (attempt - 1)),
                            max_delay,
                        )
                        print(
                            f"Rate limit, повтор через {delay:.1f}с..."
                        )
                        await asyncio.sleep(delay)
                        last_exception = e
                        continue
                    raise
            raise last_exception
        return wrapper
    return decorator


# Использование:
@retry(max_retries=3, base_delay=1.0)
async def get_app_safe(cp):
    return await cp.get_me()

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

import asyncio
import logging
from typing import TypeVar, Callable, Awaitable
from aiosend import CryptoPay
from aiosend.exceptions import (
    CryptoPayError,
    APIError,
    APITimeoutError,
    WrongNetworkError,
    MethodValuesError,
)

T = TypeVar("T")

logger = logging.getLogger("aiosend_app")


class CryptoPayClient:
    """Обёртка над CryptoPay с обработкой ошибок."""

    def __init__(self, token: str, timeout: int = 30):
        try:
            self._client = CryptoPay(
                token=token,
                timeout=timeout,
            )
            logger.info("Клиент CryptoPay создан")
        except WrongNetworkError as e:
            logger.critical("Неверная сеть: %s", e)
            raise
        except Exception as e:
            logger.critical(
                "Не удалось создать клиент: %s", e,
            )
            raise

    async def safe_call(
        self,
        action: Callable[..., Awaitable[T]],
        *args,
        max_retries: int = 2,
        **kwargs,
    ) -> T | None:
        """Безопасный вызов API с retry."""
        for attempt in range(1, max_retries + 1):
            try:
                return await action(*args, **kwargs)
            except APITimeoutError as e:
                logger.warning(
                    "Таймаут (попытка %d/%d): %s",
                    attempt, max_retries, e,
                )
                if attempt < max_retries:
                    await asyncio.sleep(2 ** attempt)
            except APIError as e:
                logger.error(
                    "Ошибка API [%d]: %s", e.code, e.name,
                )
                if e.code == 429 and attempt < max_retries:
                    await asyncio.sleep(5 * attempt)
                    continue
                raise
            except MethodValuesError as e:
                logger.error("Неверные параметры: %s", e)
                raise
            except CryptoPayError as e:
                logger.error("Ошибка aiosend: %s", e)
                raise
        return None

    async def get_app_info(self):
        return await self.safe_call(self._client.get_me)

    async def create_invoice(self, amount, asset):
        return await self.safe_call(
            self._client.create_invoice,
            amount=amount,
            asset=asset,
        )


async def main():
    logging.basicConfig(level=logging.INFO)
    client = CryptoPayClient(token="TOKEN")

    app = await client.get_app_info()
    if app:
        logger.info("Приложение: %s (ID: %d)", app.name, app.app_id)

    invoice = await client.create_invoice(100, "USDT")
    if invoice:
        logger.info(
            "Инвойс #%d: %s", invoice.invoice_id,
            invoice.bot_invoice_url,
        )


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

Помимо исключений aiosend, при работе с моделями данных ты можешь столкнуться с ошибками валидации Pydantic. Они возникают при попытке создать модель с неверными типами данных:

from pydantic import ValidationError
from aiosend import Invoice

try:
    # Неверный тип данных для amount (ожидается str)
    invoice = Invoice(
        invoice_id="not_an_int",  # ошибка!
        amount=100,               # должно быть str
        asset="USDT",
        status="active",
        currency_type="crypto",
        created_at="2024-01-01",
        bot_invoice_url="https://...",
        mini_app_invoice_url="https://...",
        web_app_invoice_url="https://...",
    )
except ValidationError as e:
    print(f"Ошибка валидации: {e}")
    # Pydantic покажет, какие поля не прошли валидацию
    # и какие типы ожидались

    # Можно получить структурированную информацию
    for error in e.errors():
        print(f"  Поле: {error['loc']}")
        print(f"  Ошибка: {error['msg']}")
        print(f"  Тип: {error['type']}")

Обработка ошибок при десериализации

При загрузке данных из внешних источников (файлы, база данных) всегда оборачивай создание моделей в try/except:

import json
from pydantic import ValidationError
from aiosend import App, Balance

def load_data_from_file(filepath: str) -> dict:
    """Загрузка данных с обработкой ошибок."""
    try:
        with open(filepath, "r") as f:
            data = json.load(f)

        # Восстанавливаем модели
        app = App.model_validate(data["app"])
        balances = [
            Balance.model_validate(b)
            for b in data.get("balances", [])
        ]
        return {"app": app, "balances": balances}

    except FileNotFoundError:
        print(f"Файл {filepath} не найден")
        return {}
    except json.JSONDecodeError as e:
        print(f"Ошибка парсинга JSON: {e}")
        return {}
    except ValidationError as e:
        print(f"Ошибка валидации данных: {e}")
        return {}

# Использование
result = load_data_from_file("app_export.json")
if result:
    app = result["app"]
    print(f"Загружено приложение: {app.name}")