dApp Backend Development with Python

DApp Backend Development with Python

JavaScript stack is not the only reasonable choice for a dApp backend. Python excels where data analysis matters, ML model integration, financial mathematics (DeFi calculations, portfolio valuation, risk metrics). Plus, if your team already writes Python — no need to learn TypeScript just for a backend.

Python DApp Backend Stack

web3.py as Foundation

web3.py — the official Python library for interacting with Ethereum-compatible blockchains. Its API is spiritually close to ethers.js, but with Pythonic syntax:

from web3 import Web3
from web3.middleware import geth_poa_middleware

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

# Polygon and other PoA networks require middleware
w3.middleware_onion.inject(geth_poa_middleware, layer=0)

# Read data
balance = w3.eth.get_balance("0xAddress")
block = w3.eth.get_block("latest")

# Contract
contract = w3.eth.contract(address=checksum_address, abi=ABI)
result = contract.functions.balanceOf(address).call()

Important: web3.py strictly requires checksum addresses. Web3.to_checksum_address("0xaddress") is a mandatory step when working with addresses from external sources.

eth-account — package for working with accounts, signatures, transactions. Often comes bundled with web3.py:

from eth_account import Account
from eth_account.messages import encode_defunct

# Verify SIWE signature
message = encode_defunct(text=raw_message)
recovered_address = Account.recover_message(message, signature=signature)
assert recovered_address.lower() == expected_address.lower()

FastAPI as HTTP Layer

FastAPI + uvicorn — the standard choice for Python web3 backends. Async-first, automatic OpenAPI documentation, Pydantic for data validation:

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

app = FastAPI()

class TransactionRequest(BaseModel):
    address: str
    amount: str  # in 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 (used in FastAPI 0.100+) is significantly faster than v1 thanks to its Rust core. Make sure you're using v2 — the API changed slightly.

Celery for Background Tasks

Typical dApp backend tasks that can't be done in HTTP handlers: sending transactions (can take seconds), indexing events, periodic jobs (price updates, 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)

# Periodic tasks
celery_app.conf.beat_schedule = {
    "sync-prices": {
        "task": "tasks.sync_token_prices",
        "schedule": crontab(minute="*/5")
    }
}

SQLAlchemy + PostgreSQL for Data Storage

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]  # store as string, Decimal loses precision
    block_number: Mapped[int] = mapped_column(index=True)
    timestamp: Mapped[datetime]
    status: Mapped[str]  # "pending" | "confirmed" | "failed"

Decimal from Python's standard library loses precision on very large numbers (uint256). Wei values are better stored as strings in the database and converted via Web3.from_wei() only when displaying.

Signing Transactions on the Backend

For dApps where the server sends transactions (like a gas-less relayer or backend wallet):

from web3 import Web3
from eth_account import Account

PRIVATE_KEY = os.environ["HOT_WALLET_PRIVATE_KEY"]  # never hardcode
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),  # slight buffer
        "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()

Important: nonce management with parallel transactions. If two Celery workers read nonce simultaneously, they'll get the same value — one transaction is lost. Solution: Redis lock or nonce pool.

Monitoring Events: Event Subscription

web3.py supports polling and WebSocket subscriptions for events:

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)

For production — use Alchemy Notify webhooks instead of polling: more reliable and doesn't require a constantly open connection.

Project Structure

dapp-backend/
├── app/
│   ├── api/          # FastAPI routers
│   ├── core/         # web3.py clients, config
│   ├── models/       # SQLAlchemy models
│   ├── services/     # business logic
│   ├── tasks/        # Celery tasks
│   └── schemas/      # Pydantic schemas
├── tests/
├── alembic/          # database migrations
├── docker-compose.yml
└── pyproject.toml    # Poetry / uv

uv instead of pip/poetry — the new standard for managing Python environments, orders of magnitude faster than pip.

Development Timeline

Week 1: Basic architecture, web3.py clients, FastAPI endpoints for reading data, PostgreSQL schema.

Week 2: Celery tasks, event indexer, SIWE authentication, transaction signing.

Full backend with event indexer, API, background tasks, and tests — 1.5-2 weeks depending on business logic complexity.