Розробка DApp бекенду на Python
JavaScript-стек — не єдиний розумний вибір для dApp бекенду. Python виграє там, де важлива аналітична обробка даних, інтеграція з ML-моделями, робота з фінансовою математикою (DeFi розрахунки, оцінка портфелів, risk-метрики). Крім того, якщо команда вже пише на 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 адреси. 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 для фонових завдань
Типові завдання бекенду 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] # зберігаємо як строку, 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 при паралельних транзакціях. Якщо два 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 webhook'и замість polling: надійніше та не вимагає постійно відкритого з'єднання.
Структура проекту
dapp-backend/
├── app/
│ ├── api/ # FastAPI router'и
│ ├── core/ # web3.py клієнти, конфіг
│ ├── models/ # SQLAlchemy моделі
│ ├── services/ # бізнес-логіка
│ ├── tasks/ # Celery завдання
│ └── schemas/ # Pydantic схеми
├── tests/
├── alembic/ # міграції БД
├── docker-compose.yml
└── pyproject.toml # Poetry / uv
uv замість pip/poetry — новий стандарт управління Python окруженнями, порядку швидше pip.
Графік розробки
Тиждень 1: Базова архітектура, web3.py клієнти, FastAPI endpoint'и для читання даних, PostgreSQL схема.
Тиждень 2: Celery завдання, event indexer, SIWE аутентифікація, підписання транзакцій.
Повний бекенд з індексером подій, API, фоновими завданнями та тестами — 1,5-2 тижні залежно від складності бізнес-логіки.