Разработка Python-бэкенда для dApp: web3.py, FastAPI, Celery

Почему Python для бэкенда dApp?

Представьте: DeFi-агрегатор должен каждые 5 минут обновлять цены из 20 пулов, рассчитывать impermanent loss и отправлять транзакции с минимальным slippage. Синхронный Flask тут не справится — блокчейн даёт блок раз в 12–15 секунд, а каждый HTTP-запрос ждёт ответа от RPC. Python с асинхронным стеком решает это: web3.py для блокчейна, FastAPI для API, Celery для фоновых задач. За 7 лет мы реализовали 10+ таких проектов: от NFT-маркетплейсов до DeFi-агрегаторов с APY-калькуляторами.

web3.py provides a smooth interface to interact with Ethereum nodes — документация web3.py

Как web3.py упрощает взаимодействие с блокчейном?

web3.py — зрелая библиотека для работы с EVM-сетями. Основные сложности: проверка checksum-адресов и PoA-мидлвари. Без Web3.to_checksum_address() любое обращение к внешнему источнику вызовет ошибку. Для Polygon и BNB Chain обязательно подключаем geth_poa_middleware. Пример настройки:

from web3 import Web3 from web3.middleware import geth_poa_middleware w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/KEY")) w3.middleware_onion.inject(geth_poa_middleware, layer=0) balance = w3.eth.get_balance("0xChecksumAddress") contract = w3.eth.contract(address=checksum_address, abi=ABI) result = contract.functions.balanceOf(address).call() 

Как асинхронность и Celery решают проблему производительности?

Синхронный вызов блокчейна блокирует event loop. Мы используем FastAPI + async web3, а фоновые задачи выносим в Celery. Пропускная способность растёт с ~50 до 500+ запросов/с. Celery-задачи обрабатывают длительные операции: отправку транзакций, индексацию событий, синхронизацию цен. Пример задачи с retry:

from celery import Celery from celery.schedules import crontab celery_app = Celery("dapp", broker="redis://localhost:6379/0") @celery_app.task(bind=True, max_retries=3) def send_transaction(self, contract_address, function_name, args): try: contract = w3.eth.contract(address=contract_address, abi=ABI) tx_hash = contract.functions[function_name](*args).transact({ "from": hot_wallet.address, "gas": 200000 }) return {"tx_hash": tx_hash.hex(), "status": "pending"} except Exception as exc: raise self.retry(exc=exc, countdown=30) celery_app.conf.beat_schedule = { "sync-prices": { "task": "tasks.sync_token_prices", "schedule": crontab(minute="*/5") } } 

Основные проблемы бэкенда dApp

Nonce management. При параллельной отправке транзакций два воркера могут прочитать одинаковый nonce — одна транзакция застрянет. Решение: Redis-блокировки с TTL или nonce pool. Gas optimization. Каждый лишний вызов eth_call или eth_sendTransaction стоит денег. Кеширование данных на уровне API сокращает расходы на 30–40%. Индексация событий. WebSocket-подписки ненадёжны в production — используем Alchemy Notify + Celery-задачи для пересинхронизации при сбоях.

Наш стек и примеры кода

FastAPI с Pydantic v2 для валидации:

from fastapi import FastAPI, HTTPException from pydantic import BaseModel, validator import re class TransactionRequest(BaseModel): address: str amount: str @validator("address") def validate_eth_address(cls, v): if not re.match(r"^0x[a-fA-F0-9]{40}$", v): raise ValueError("Invalid Ethereum address") return Web3.to_checksum_address(v) app = FastAPI() @app.get("/api/balance/{address}") async def get_balance(address: str): try: checksum = Web3.to_checksum_address(address) except ValueError: raise HTTPException(status_code=400, detail="Invalid address") balance_wei = w3.eth.get_balance(checksum) return { "address": checksum, "balance_eth": Web3.from_wei(balance_wei, "ether"), "balance_wei": str(balance_wei) } 

SQLAlchemy + PostgreSQL и хранение wei как строки:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped from datetime import datetime class Base(DeclarativeBase): pass class Transaction(Base): __tablename__ = "transactions" id: Mapped[int] = mapped_column(primary_key=True) tx_hash: Mapped[str] = mapped_column(unique=True, index=True) from_address: Mapped[str] = mapped_column(index=True) to_address: Mapped[str] = mapped_column(index=True) value_wei: Mapped[str] # строка, не теряет точность block_number: Mapped[int] = mapped_column(index=True) timestamp: Mapped[datetime] status: Mapped[str] 

Процесс работы

1. Анализ. Изучаем смарт-контракты, требования к API, бизнес-логику. Составляем ТЗ.
2. Проектирование. Определяем архитектуру: структура БД, API-методы, Celery-задачи, схему подписания.
3. Реализация. Пишем код, покрываем тестами (pytest), интегрируем с блокчейном.
4. Тестирование. Запускаем на тестовой сети, проверяем сценарии: отправка транзакций, обработка событий, восстановление после сбоев.
5. Деплой. Развёртываем в production с Docker, настраиваем мониторинг (grafana, loki).

Сроки разработки

Полный бэкенд — от 1,5 до 2 недель. Стоимость рассчитывается индивидуально.

Частые ошибки при разработке Python-бэкенда для dApp

• Забыть про nonce management. Без синхронизации при параллельных запросах транзакции застревают.
• Использовать Decimal для wei. Лучше хранить как строку, чтобы избежать потери точности.
• Не добавить middleware для PoA-сетей. Иначе get_block выбросит ошибку.
• Синхронные запросы в API. Блокируют event loop, снижают пропускную способность в 10 раз.

Что вы получаете

• Архитектуру, адаптированную под вашу бизнес-логику.
• REST API с OpenAPI-документацией.
• Индексатор событий с записью в PostgreSQL.
• Фоновые задачи: отправка транзакций, синхронизация цен, health-чеки.
• Безопасное подписание с hot wallet или интеграция с Vault/AWS KMS.
• Docker-контейнеризацию и документацию по развёртыванию.
• Гарантию на код 3 месяца и поддержку после сдачи.

Свяжитесь с нами для консультации — мы поможем выбрать архитектуру и оценим сроки. Закажите разработку бэкенда вашего dApp уже сегодня.