Разработка бэкенда dApp на Python

Разработка бэкенда dApp на Python

JavaScript-стек — не единственный разумный выбор для dApp бэкенда. Python выигрывает там, где важна аналитическая обработка данных, интеграция с ML-моделями, работа с финансовой математикой (DeFi расчёты, оценка портфелей, риск-метрики). Кроме того, если команда уже пишет на Python — не нужно учить TypeScript ради бэкенда.

Стек для Python dApp бэкенда

web3.py как основа

web3.py — официальная Python библиотека для взаимодействия с Ethereum-совместимыми блокчейнами. API по духу близок к ethers.js, но с Pythonic синтаксом:

from web3 import Web3
from web3.middleware import geth_poa_middleware

w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/KEY"))

# Polygon и другие PoA сети требуют middleware
w3.middleware_onion.inject(geth_poa_middleware, layer=0)

# Чтение данных
balance = w3.eth.get_balance("0xAddress")
block = w3.eth.get_block("latest")

# Контракт
contract = w3.eth.contract(address=checksum_address, abi=ABI)
result = contract.functions.balanceOf(address).call()

Важный момент: web3.py строго требует checksum addresses. Web3.to_checksum_address("0xaddress") — обязательный шаг при работе с адресами из внешних источников.

eth-account — пакет для работы с аккаунтами, подписями, транзакциями. Часто идёт в комплекте с web3.py:

from eth_account import Account
from eth_account.messages import encode_defunct

# Верификация SIWE подписи
message = encode_defunct(text=raw_message)
recovered_address = Account.recover_message(message, signature=signature)
assert recovered_address.lower() == expected_address.lower()

FastAPI как HTTP слой

FastAPI + uvicorn — стандартный выбор для Python web3 бэкендов. Async-first, автоматическая OpenAPI документация, Pydantic для валидации данных:

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, validator
import re

app = FastAPI()

class TransactionRequest(BaseModel):
    address: str
    amount: str  # в ETH

    @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.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)
    }

Pydantic v2 (используется в FastAPI 0.100+) значительно быстрее v1 за счёт Rust-ядра. Стоит убедиться что используется именно v2 — API немного изменился.

Celery для фоновых задач

Типичные задачи bачкенда dApp, которые нельзя делать в HTTP handler: отправка транзакций (может занимать секунды), индексирование событий, периодические джобы (обновление курсов, health checks).

from celery import Celery
from celery.schedules import crontab

celery_app = Celery(
    "dapp",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1"
)

@celery_app.task(bind=True, max_retries=3)
def send_transaction(self, contract_address: str, function_name: str, args: list):
    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")
    }
}

SQLAlchemy + PostgreSQL для хранения данных

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
from datetime import datetime
from decimal import Decimal

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]  # храним как string, Decimal теряет точность
    block_number: Mapped[int] = mapped_column(index=True)
    timestamp: Mapped[datetime]
    status: Mapped[str]  # "pending" | "confirmed" | "failed"

Decimal из Python стандартной библиотеки теряет точность на очень больших числах (uint256). Значения Wei лучше хранить как строки в БД и конвертировать через Web3.from_wei() только при отображении.

Подписание транзакций в бэкенде

Для dApp с серверной стороной, которая отправляет транзакции (например, gas-less relayer, backend wallet):

from web3 import Web3
from eth_account import Account

PRIVATE_KEY = os.environ["HOT_WALLET_PRIVATE_KEY"]  # никогда не в коде
account = Account.from_key(PRIVATE_KEY)

def send_signed_transaction(to: str, value_eth: float, data: bytes = b"") -> str:
    nonce = w3.eth.get_transaction_count(account.address)
    gas_price = w3.eth.gas_price

    tx = {
        "nonce": nonce,
        "to": Web3.to_checksum_address(to),
        "value": Web3.to_wei(value_eth, "ether"),
        "gas": 21000,
        "gasPrice": int(gas_price * 1.1),  # небольшой буфер
        "chainId": w3.eth.chain_id,
        "data": data
    }

    signed = account.sign_transaction(tx)
    tx_hash = w3.eth.send_raw_transaction(signed.rawTransaction)
    return tx_hash.hex()

Важно: nonce management при параллельных транзакциях. Если два Celery воркера одновременно читают nonce, оба получат одинаковое значение — одна транзакция потеряется. Решение: Redis lock или nonce pool.

Мониторинг событий: event subscription

web3.py поддерживает polling и WebSocket подписки на события:

import asyncio
from web3 import AsyncWeb3

async def watch_events(contract_address: str):
    w3 = AsyncWeb3(AsyncWeb3.AsyncWebsocketProvider("wss://eth-mainnet.g.alchemy.com/v2/KEY"))
    contract = w3.eth.contract(address=contract_address, abi=ABI)

    event_filter = await contract.events.Transfer.create_filter(fromBlock="latest")

    while True:
        events = await event_filter.get_new_entries()
        for event in events:
            await process_transfer_event(event)
        await asyncio.sleep(2)

Для production — использовать Alchemy Notify webhooks вместо polling: надёжнее и не требует постоянного открытого соединения.

Структура проекта

dapp-backend/
├── app/
│   ├── api/          # FastAPI routers
│   ├── core/         # web3.py clients, config
│   ├── models/       # SQLAlchemy models
│   ├── services/     # бизнес-логика
│   ├── tasks/        # Celery tasks
│   └── schemas/      # Pydantic schemas
├── tests/
├── alembic/          # миграции БД
├── docker-compose.yml
└── pyproject.toml    # Poetry / uv

uv вместо pip/poetry — новый стандарт управления Python окружениями, порядка быстрее pip.

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

Неделя 1: Базовая архитектура, web3.py клиенты, FastAPI endpoints для чтения данных, PostgreSQL схема.

Неделя 2: Celery задачи, event indexer, SIWE аутентификация, подписание транзакций.

Полный бэкенд с индексером событий, API, фоновыми задачами и тестами — 1.5-2 недели в зависимости от сложности бизнес-логики.