Урок 2.2 — getMe и getBalance
Научимся получать информацию о своём приложении и проверять балансы через методы get_me(), get_balance() и get_balance_by_asset().
Когда клиент создан, первый логичный шаг — убедиться, что всё работает. Метод get_me() — аналог рукопожатия: он проверяет токен и возвращает базовую информацию о приложении. Метод get_balance() показывает, сколько средств лежит на счету вашего приложения в разных криптовалютах. В этом уроке мы разберём оба метода и их типы возвращаемых значений.
Метод get_me() — информация о приложении
Метод get_me() — самый простой метод Crypto Pay API. Он не требует параметров и возвращает объект App с базовой информацией о вашем приложении. Это аналог /getMe в Telegram Bot API.
Сигнатура метода
async def get_me(self) -> App:
Не требует параметров. Возвращает App.
import asyncio
from aiosend import CryptoPay
async def main():
cp = CryptoPay(token="1234:TOKEN")
app = await cp.get_me()
print(app)
asyncio.run(main())
# Пример вывода:
# app_id=12345 name='My Crypto App' payment_processing_bot_username='CryptoBot'
Объект App и его поля
Класс App — это Pydantic-модель с тремя полями. Он наследуется от базового класса CryptoPayObject.
| Поле | Тип | Описание |
|---|---|---|
app_id |
int |
Уникальный идентификатор вашего приложения в Crypto Pay. Выдаётся при создании токена. |
name |
str |
Название вашего приложения, которое вы указали при создании токена. |
payment_processing_bot_username |
str |
Telegram username бота, обрабатывающего платежи: "CryptoBot" (MAINNET) или "CryptoTestnetBot" (TESTNET). |
import asyncio
from aiosend import CryptoPay
async def main():
cp = CryptoPay(token="1234:TOKEN")
app = await cp.get_me()
# app_id — числовой идентификатор
print(f"ID приложения: {app.app_id}") # 12345
# name — название приложения
print(f"Название: {app.name}") # My App
# payment_processing_bot_username — бот для платежей
bot = app.payment_processing_bot_username
print(f"Платёжный бот: @{bot}") # @CryptoBot
# Если сеть TESTNET — бот будет @CryptoTestnetBot
# Это удобно для отладки: можно проверить, в какой сети работает клиент
asyncio.run(main())
⚠️ get_me() вызывается автоматически при создании клиента
Конструктор CryptoPay уже вызывает get_me() внутри для проверки токена (через __auth). Если токен неверный, вы получите ошибку сразу при создании клиента, а не при первом вызове метода.
Тем не менее, явный вызов get_me() полезен:
- Для проверки, что клиент всё ещё авторизован (токен не отозван)
- Для получения имени приложения в логах
- Для определения, с какой сетью работает клиент (через
payment_processing_bot_username)
import asyncio
from aiosend import CryptoPay
async def main():
cp = CryptoPay(token="1234:TOKEN")
app = await cp.get_me()
if "testnet" in app.payment_processing_bot_username.lower():
print("⚠️ Работаем в TESTNET (тестовой сети)!")
else:
print("✅ Работаем в MAINNET (основной сети)")
print(f"Приложение: {app.name} (ID: {app.app_id})")
asyncio.run(main())
🔍 Исходный код App
class App(CryptoPayObject):
app_id: int # ID приложения
name: str # Название приложения
payment_processing_bot_username: str # Username платёжного бота
Метод get_balance() — балансы приложения
Метод get_balance() возвращает список всех балансов вашего приложения. Каждый элемент списка — объект Balance, содержащий информацию о доступных и замороженных средствах в конкретной криптовалюте.
Сигнатура метода
async def get_balance(self) -> list[Balance]:
Не требует параметров. Возвращает список Balance.
Объект Balance и его поля
| Поле | Тип | Описание |
|---|---|---|
currency_code |
Asset | str |
Код криптовалюты: "USDT", "TON", "BTC", "ETH", "LTC", "BNB", "TRX", "USDC". В TESTNET также "JET". |
available |
float |
Сумма средств, доступная для использования (переводов, создания чеков и т.д.). |
onhold |
float |
Сумма средств, временно замороженная (например, на未оплаченных инвойсах). Недоступна для использования. |
import asyncio
from aiosend import CryptoPay
async def main():
cp = CryptoPay(token="1234:TOKEN")
balances = await cp.get_balance()
print("=" * 60)
print(f"{'Валюта':<10} {'Доступно':<20} {'Заморожено':<20} {'Всего':<10}")
print("-" * 60)
for b in balances:
total = b.available + b.onhold
print(f"{b.currency_code:<10} {b.available:<20.2f} {b.onhold:<20.2f} {total:<10.2f}")
print("=" * 60)
asyncio.run(main())
# Пример вывода:
# ============================================================
# Валюта Доступно Заморожено Всего
# ------------------------------------------------------------
# USDT 150.50 10.00 160.50
# TON 500.00 0.00 500.00
# BTC 0.50 0.00 0.50
# ============================================================
⚠️ onhold — что это?
onhold — средства, которые временно заморожены. Они "зарезервированы" под активные (неоплаченные) инвойсы. Как только инвойс оплачивается или истекает, средства переходят в available или возвращаются. Если вы создали инвойс на 100 USDT и он ещё не оплачен — эти 100 USDT могут быть в onhold.
from aiosend import CryptoPay
cp = CryptoPay("1234:TOKEN")
balances = cp.get_balance() # синхронный вызов, без await!
for b in balances:
print(f"{b.currency_code}: {b.available} (доступно), {b.onhold} (заморожено)")
Обратите внимание: Balance также имеет метод update(), который обновляет данные объекта из API:
import asyncio
from aiosend import CryptoPay
async def main():
cp = CryptoPay(token="1234:TOKEN")
balances = await cp.get_balance()
# Берём баланс USDT
usdt_balance = None
for b in balances:
if b.currency_code == "USDT":
usdt_balance = b
break
if usdt_balance:
print(f"До обновления: {usdt_balance.available} USDT")
await usdt_balance.update() # обновляем из API
print(f"После обновления: {usdt_balance.available} USDT")
asyncio.run(main())
🔍 Исходный код Balance
class Balance(CryptoPayObject):
currency_code: Asset | str # код валюты
available: float # доступно
onhold: float # заморожено
async def update(self) -> None:
"""Обновить данные баланса из API"""
balance = await self._client.get_balance_by_asset(self.currency_code)
self.__dict__ = balance.__dict__
Метод get_balance_by_asset() — баланс конкретного актива
Метод get_balance_by_asset() — это удобная обёртка над get_balance(). Вместо того чтобы получать все балансы и фильтровать их вручную, вы можете сразу запросить баланс конкретного актива.
Сигнатура метода
async def get_balance_by_asset(self, asset: Asset | str) -> Balance:
Принимает asset (строка или Enum Asset). Возвращает Balance. Выбрасывает CryptoPayError, если актив не найден.
| Параметр | Тип | Описание |
|---|---|---|
asset |
Asset | str |
Код криптовалюты. Можно передать строку "USDT" или элемент Enum Asset.USDT. |
import asyncio
from aiosend import CryptoPay
from aiosend.enums import Asset
async def main():
cp = CryptoPay(token="1234:TOKEN")
# Вариант 1: строка
usdt = await cp.get_balance_by_asset("USDT")
print(f"USDT: {usdt.available} (доступно), {usdt.onhold} (заморожено)")
# Вариант 2: Enum Asset
ton = await cp.get_balance_by_asset(Asset.TON)
print(f"TON: {ton.available} (доступно), {ton.onhold} (заморожено)")
# Вариант 3: через метод update() на объекте
ethereum = await cp.get_balance_by_asset("ETH")
print(f"ETH до: {ethereum.available}")
await ethereum.update()
print(f"ETH после: {ethereum.available}")
asyncio.run(main())
import asyncio
from aiosend import CryptoPay
from aiosend.exceptions import CryptoPayError
async def main():
cp = CryptoPay(token="1234:TOKEN")
try:
balance = await cp.get_balance_by_asset("SOL")
print(f"SOL: {balance.available}")
except CryptoPayError as e:
print(f"Баланс для SOL не найден: {e}")
# Например: "Balance for SOL not found"
asyncio.run(main())
⚠️ Когда get_balance_by_asset может не найти актив?
Если на вашем балансе никогда не было средств в этом активе — его не будет в списке. На пустом кошельке актив отсутствует, а не присутствует с нулевым балансом. В этом случае метод выбрасывает CryptoPayError.
Как работает get_balance_by_asset внутри? Это простой перебор:
async def get_balance_by_asset(self, asset):
# Получаем ВСЕ балансы
balances = await self.get_balance()
# Ищем нужный актив
for balance in balances:
if balance.currency_code == asset:
return balance
# Если не нашли — ошибка
msg = f"Balance for {asset} not found"
raise CryptoPayError(msg)
Практический пример: мониторинг баланса
Давайте напишем простой, но полезный скрипт — мониторинг баланса. Он будет периодически проверять баланс USDT и уведомлять об изменениях.
import asyncio
import time
from datetime import datetime
from aiosend import CryptoPay
from aiosend.exceptions import CryptoPayError
class BalanceMonitor:
"""Мониторинг баланса USDT."""
def __init__(self, token: str, interval: int = 60):
self.cp = CryptoPay(token=token)
self.interval = interval
self._last_available = 0.0
async def check(self) -> None:
"""Проверяет текущий баланс и логирует изменения."""
try:
balance = await self.cp.get_balance_by_asset("USDT")
now = datetime.now().strftime("%H:%M:%S")
if balance.available != self._last_available:
diff = balance.available - self._last_available
sign = "+" if diff > 0 else ""
print(
f"[{now}] USDT: {balance.available:.2f} "
f"({sign}{diff:.2f}) | "
f"onhold: {balance.onhold:.2f}"
)
self._last_available = balance.available
except CryptoPayError:
print(f"[{datetime.now().strftime('%H:%M:%S')}] USDT не найден на балансе")
async def run(self) -> None:
"""Запускает циклическую проверку."""
print(f"Мониторинг баланса USDT (интервал: {self.interval}с)")
print("Нажмите Ctrl+C для остановки\n")
try:
while True:
await self.check()
await asyncio.sleep(self.interval)
except KeyboardInterrupt:
print("\nМониторинг остановлен.")
async def main():
monitor = BalanceMonitor(
token="1234:TOKEN",
interval=30, # проверять каждые 30 секунд
)
await monitor.run()
if __name__ == "__main__":
asyncio.run(main())
Пример вывода:
Мониторинг баланса USDT (интервал: 30с)
Нажмите Ctrl+C для остановки
[10:00:00] USDT: 150.50 (+0.00) | onhold: 0.00
[10:00:30] USDT: 150.50 (+0.00) | onhold: 10.00 ← создан инвойс
[10:01:00] USDT: 140.50 (-10.00) | onhold: 0.00 ← инвойс оплачен
[10:01:30] USDT: 190.50 (+50.00) | onhold: 0.00 ← пополнение
💡 Идеи для расширения мониторинга
- Отправлять уведомления в Telegram при изменении баланса
- Следить за несколькими активами одновременно
- Записывать историю изменений в CSV или БД
- Устанавливать пороговые значения и оповещать при падении ниже лимита
import asyncio
from datetime import datetime
from aiosend import CryptoPay
class AllBalanceMonitor:
"""Мониторинг всех балансов."""
def __init__(self, token: str, interval: int = 60):
self.cp = CryptoPay(token=token)
self.interval = interval
async def check_all(self) -> None:
"""Выводит все балансы."""
balances = await self.cp.get_balance()
now = datetime.now().strftime("%H:%M:%S")
print(f"\n[{now}] Балансы:")
print("-" * 50)
for b in balances:
total = b.available + b.onhold
print(f" {b.currency_code:<6} | {b.available:>10.2f} дост. | "
f"{b.onhold:>10.2f} замор. | {total:>10.2f} всего")
print("-" * 50)
async def run(self) -> None:
print(f"Мониторинг всех балансов (интервал: {self.interval}с)")
try:
while True:
await self.check_all()
await asyncio.sleep(self.interval)
except KeyboardInterrupt:
print("\nМониторинг остановлен.")
async def main():
monitor = AllBalanceMonitor("TOKEN", interval=30)
await monitor.run()
asyncio.run(main())
Сравнение методов get_me и get_balance
| Характеристика | get_me() | get_balance() | get_balance_by_asset() |
|---|---|---|---|
| Параметры | Нет | Нет | asset |
| Возвращает | App |
list[Balance] |
Balance |
| Исключения | APIError (неверный токен) |
APIError |
APIError, CryptoPayError (актив не найден) |
| Частота вызова | Один раз при старте | Периодически | Когда нужен конкретный актив |
| Назначение | Проверка токена, информация о приложении | Полная картина балансов | Быстрый доступ к одному балансу |
Какой метод выбрать?
- Нужно проверить токен? →
get_me() - Нужно показать все балансы пользователю? →
get_balance() - Нужно проверить, хватает ли USDT для перевода? →
get_balance_by_asset("USDT") - Нужно обновить конкретный Balance объект? →
balance.update()
Интеграция get_me и get_balance — информационная панель
Объединим оба метода для создания информационной панели (dashboard) вашего Crypto Pay приложения.
import asyncio
from datetime import datetime
from aiosend import CryptoPay
async def show_dashboard(token: str) -> None:
"""Показывает панель управления приложением."""
cp = CryptoPay(token=token)
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# Получаем информацию о приложении
app = await cp.get_me()
# Получаем балансы
balances = await cp.get_balance()
# Выводим дашборд
print("=" * 60)
print(f" 📊 CRYPTO PAY DASHBOARD")
print(f" {now}")
print("=" * 60)
print(f" Приложение: {app.name}")
print(f" ID: {app.app_id}")
print(f" Платёжный бот: @{app.payment_processing_bot_username}")
print(f" Сеть: {cp.session.network.name}")
print("-" * 60)
print(f" {'Актив':<8} {'Доступно':<15} {'Заморожено':<15} {'Всего':<10}")
print("-" * 60)
total_usd_value = 0.0
for b in balances:
total = b.available + b.onhold
print(f" {b.currency_code:<8} {b.available:<15.2f} {b.onhold:<15.2f} {total:<10.2f}")
print("-" * 60)
print(f" Всего активов: {len(balances)}")
print("=" * 60)
async def main():
await show_dashboard("1234:TOKEN")
asyncio.run(main())
# Пример вывода:
# ============================================================
# 📊 CRYPTO PAY DASHBOARD
# 2026-06-23 10:30:00
# ============================================================
# Приложение: My Shop Bot
# ID: 12345
# Платёжный бот: @CryptoBot
# Сеть: MAINNET
# -----------------------------------------------------------
# Актив Доступно Заморожено Всего
# -----------------------------------------------------------
# USDT 150.50 10.00 160.50
# TON 500.00 0.00 500.00
# BTC 0.50 0.00 0.50
# -----------------------------------------------------------
# Всего активов: 3
# ============================================================
Типичные ошибки и их обработка
Ошибка 1: APIError при get_me()
Токен отозван, неверный формат или превышены лимиты запросов.
try:
app = await cp.get_me()
except APIError as e:
print(f"Ошибка API [{e.error.code}]: {e.error.name}")
# [401]: Unauthorized — токен недействителен
# [429]: Too Many Requests — превышен лимит
Ошибка 2: CryptoPayError при get_balance_by_asset()
Актив не найден в списке балансов (никогда не было средств).
try:
balance = await cp.get_balance_by_asset("SOL")
except CryptoPayError:
print("SOL отсутствует на балансе. Вероятно, на нём никогда не было средств.")
# Можно установить баланс в 0 вручную
balance = 0.0
Ошибка 3: APITimeoutError
Сервер не отвечает. Возможно, проблемы с сетью или блокировки.
from aiosend.exceptions import APITimeoutError
try:
balances = await cp.get_balance()
except APITimeoutError:
print("Таймаут при получении баланса. Попробуйте позже.")
except APIError as e:
print(f"Ошибка API: {e}")
Что важно запомнить
Практическая задача
Задача: Проверка баланса перед переводом
Напишите асинхронную функцию check_balance_before_transfer, которая:
- Принимает
cp(клиент CryptoPay),asset(строка) иrequired_amount(float). - Получает баланс указанного актива через
get_balance_by_asset(). - Сравнивает
availableсrequired_amount. - Если средств достаточно — возвращает
Trueи печатает "Достаточно средств". - Если не хватает — возвращает
Falseи печатает "Недостаточно средств. Нужно X, доступно Y". - Если актив не найден — печатает "Актив X отсутствует на балансе" и возвращает
False.
Пример использования:
cp = CryptoPay("TOKEN")
result = await check_balance_before_transfer(cp, "USDT", 100.0)
# Вывод: Достаточно средств (если available >= 100)
# или: Недостаточно средств. Нужно 100.0 USDT, доступно 50.0 USDT
Урок 2.2: getMe и getBalance
15 вопросов