skillbase/web3-web3py
Web3.py for EVM blockchain interaction: contract calls, ABI handling, batch/multicall, middleware, typed wrappers, and event processing
SKILL.md
41
You are a senior Python blockchain engineer specializing in Web3.py for EVM chain interactions. You build typed, async-first contract wrappers with robust error handling, batch call optimization, and middleware pipelines.
42
43
This skill covers the full Web3.py workflow: provider setup with middleware, typed contract wrappers over raw ABI calls, Multicall3 for batched reads, transaction building with EIP-1559 fees, event log processing with block-range pagination, and retry middleware. The goal is to produce blockchain integrations that are type-safe, RPC-efficient (minimal round trips), resilient to provider flakiness, and observable via structured logging. Common pitfalls this skill prevents: untyped contract calls scattered across codebase, N+1 RPC calls where Multicall suffices, missing receipt status checks causing silent reverts, and floating-point arithmetic on token amounts.
48
## Provider setup and configuration
49
50
Initialize Web3 with async HTTP provider and middleware:
51
52
```python
53
from web3 import AsyncWeb3
54
from web3.middleware import ExtraDataToPOAMiddleware
55
from web3.providers import AsyncHTTPProvider
56
57
58
def create_web3(
59
rpc_url: str,
60
*,
61
is_poa: bool = False,
62
request_timeout: int = 30,
63
) -> AsyncWeb3:
64
provider = AsyncHTTPProvider(
65
rpc_url,
66
request_kwargs={"timeout": request_timeout},67
)
68
w3 = AsyncWeb3(provider)
69
70
if is_poa:
71
w3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0)
72
73
return w3
74
```
75
76
Configuration with Pydantic Settings:
77
78
```python
79
from pydantic import Field, SecretStr
80
from pydantic_settings import BaseSettings, SettingsConfigDict
81
82
83
class Web3Settings(BaseSettings):
84
model_config = SettingsConfigDict(env_prefix="WEB3_")
85
86
rpc_url: str
87
chain_id: int
88
is_poa: bool = False
89
private_key: SecretStr | None = None
90
max_retries: int = 3
91
request_timeout: int = 30
92
multicall_address: str = "0xcA11bde05977b3631167028862bE2a173976CA11"
93
```
94
95
## ABI management
96
97
Store ABIs as JSON files, load and cache at module level:
98
99
```python
100
import json
101
from functools import lru_cache
102
from pathlib import Path
103
104
ABI_DIR = Path(__file__).parent / "abis"
105
106
107
@lru_cache(maxsize=32)
108
def load_abi(name: str) -> list[dict]:
109
abi_path = ABI_DIR / f"{name}.json"110
return json.loads(abi_path.read_text())
111
```
112
113
Project structure:
114
115
```
116
src/service/
117
└── infrastructure/
118
└── web3/
119
├── abis/
120
│ ├── erc20.json
121
│ ├── vault.json
122
│ └── multicall3.json
123
├── client.py # Web3 provider setup
124
├── contracts.py # Typed contract wrappers
125
├── multicall.py # Batch call utilities
126
└── middleware.py # Custom middleware
127
```
128
129
## Typed contract wrappers
130
131
Wrap raw Web3 contract objects behind typed interfaces:
132
133
```python
134
from dataclasses import dataclass
135
from decimal import Decimal
136
from eth_typing import ChecksumAddress
137
from web3 import AsyncWeb3
138
from web3.contract import AsyncContract
139
140
141
@dataclass(frozen=True, slots=True)
142
class TokenInfo:
143
address: ChecksumAddress
144
name: str
145
symbol: str
146
decimals: int
147
148
149
class ERC20Contract:
150
def __init__(self, w3: AsyncWeb3, address: ChecksumAddress) -> None:
151
self._w3 = w3
152
self._address = address
153
self._contract: AsyncContract = w3.eth.contract(
154
address=address, abi=load_abi("erc20")155
)
156
self._decimals: int | None = None
157
158
@property
159
def address(self) -> ChecksumAddress:
160
return self._address
161
162
async def decimals(self) -> int:
163
if self._decimals is None:
164
self._decimals = await self._contract.functions.decimals().call()
165
return self._decimals
166
167
async def symbol(self) -> str:
168
return await self._contract.functions.symbol().call()
169
170
async def name(self) -> str:
171
return await self._contract.functions.name().call()
172
173
async def balance_of(self, account: ChecksumAddress) -> Decimal:
174
raw = await self._contract.functions.balanceOf(account).call()
175
decimals = await self.decimals()
176
return Decimal(raw) / Decimal(10 ** decimals)
177
178
async def allowance(
179
self, owner: ChecksumAddress, spender: ChecksumAddress
180
) -> Decimal:
181
raw = await self._contract.functions.allowance(owner, spender).call()
182
decimals = await self.decimals()
183
return Decimal(raw) / Decimal(10 ** decimals)
184
185
async def get_info(self) -> TokenInfo:
186
return TokenInfo(
187
address=self._address,
188
name=await self.name(),
189
symbol=await self.symbol(),
190
decimals=await self.decimals(),
191
)
192
193
def to_raw_amount(self, amount: Decimal, decimals: int) -> int:
194
return int(amount * Decimal(10 ** decimals))
195
196
def to_human_amount(self, raw: int, decimals: int) -> Decimal:
197
return Decimal(raw) / Decimal(10 ** decimals)
198
```
199
200
## Building and sending transactions
201
202
Encapsulate transaction building with gas estimation and nonce management:
203
204
```python
205
from eth_account.signers.local import LocalAccount
206
from web3.types import TxParams, TxReceipt
207
import structlog
208
209
logger = structlog.get_logger()
210
211
212
class TransactionBuilder:
213
def __init__(
214
self, w3: AsyncWeb3, account: LocalAccount,
215
*, chain_id: int, gas_multiplier: float = 1.2,
216
) -> None:
217
self._w3 = w3
218
self._account = account
219
self._chain_id = chain_id
220
self._gas_multiplier = gas_multiplier
221
222
async def send(
223
self, contract_fn, *, value: int = 0, gas_limit: int | None = None,
224
) -> TxReceipt:
225
nonce = await self._w3.eth.get_transaction_count(self._account.address)
226
227
tx_params: TxParams = {228
"from": self._account.address,
229
"nonce": nonce,
230
"chainId": self._chain_id,
231
"value": value,
232
}
233
234
if gas_limit is not None:
235
tx_params["gas"] = gas_limit
236
else:
237
estimated = await contract_fn.estimate_gas(tx_params)
238
tx_params["gas"] = int(estimated * self._gas_multiplier)
239
240
# EIP-1559 fee model
241
latest = await self._w3.eth.get_block("latest")242
base_fee = latest.get("baseFeePerGas", 0)243
max_priority = await self._w3.eth.max_priority_fee
244
tx_params["maxFeePerGas"] = int(base_fee * 2 + max_priority)
245
tx_params["maxPriorityFeePerGas"] = max_priority
246
247
tx = await contract_fn.build_transaction(tx_params)
248
signed = self._account.sign_transaction(tx)
249
tx_hash = await self._w3.eth.send_raw_transaction(signed.raw_transaction)
250
251
logger.info("tx_sent", tx_hash=tx_hash.hex(), to=tx.get("to"), nonce=nonce)252
253
receipt = await self._w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)
254
255
if receipt["status"] != 1:
256
logger.error("tx_reverted", tx_hash=tx_hash.hex(), receipt=dict(receipt))257
raise TransactionRevertedError(tx_hash=tx_hash.hex(), receipt=receipt)
258
259
logger.info("tx_confirmed", tx_hash=tx_hash.hex(), block=receipt["blockNumber"], gas_used=receipt["gasUsed"])260
return receipt
261
```
262
263
Usage:
264
265
```python
266
erc20 = w3.eth.contract(address=token_addr, abi=load_abi("erc20"))267
vault = w3.eth.contract(address=vault_addr, abi=load_abi("vault"))268
tx_builder = TransactionBuilder(w3, account, chain_id=1)
269
270
await tx_builder.send(erc20.functions.approve(vault_addr, amount_raw))
271
await tx_builder.send(vault.functions.deposit(amount_raw))
272
```
273
274
## Batch calls with Multicall3
275
276
Batch multiple read calls into a single RPC request:
277
278
```python
279
from dataclasses import dataclass
280
from eth_abi import decode
281
from web3 import AsyncWeb3
282
283
284
@dataclass(frozen=True, slots=True)
285
class Call:
286
target: ChecksumAddress
287
call_data: bytes
288
decode_types: list[str]
289
290
291
@dataclass(frozen=True, slots=True)
292
class CallResult:
293
success: bool
294
data: tuple | None
295
296
297
class Multicall:
298
MULTICALL3_ADDRESS = "0xcA11bde05977b3631167028862bE2a173976CA11"
299
300
def __init__(self, w3: AsyncWeb3, *, address: str | None = None) -> None:
301
self._w3 = w3
302
self._contract = w3.eth.contract(
303
address=AsyncWeb3.to_checksum_address(address or self.MULTICALL3_ADDRESS),
304
abi=load_abi("multicall3"),305
)
306
307
async def call(self, calls: list[Call]) -> list[CallResult]:
308
multicall_input = [(call.target, True, call.call_data) for call in calls]
309
results = await self._contract.functions.aggregate3(multicall_input).call()
310
311
decoded: list[CallResult] = []
312
for (success, return_data), call in zip(results, calls, strict=True):
313
if success and return_data:
314
data = decode(call.decode_types, return_data)
315
decoded.append(CallResult(success=True, data=data))
316
else:
317
decoded.append(CallResult(success=False, data=None))
318
319
return decoded
320
321
def encode_erc20_balance(
322
self, token: ChecksumAddress, account: ChecksumAddress
323
) -> Call:
324
contract = self._w3.eth.contract(address=token, abi=load_abi("erc20"))325
call_data = contract.encode_abi("balanceOf", args=[account])326
return Call(target=token, call_data=call_data, decode_types=["uint256"])
327
```
328
329
Usage — fetch balances across multiple tokens in one RPC call:
330
331
```python
332
async def get_portfolio_balances(
333
multicall: Multicall, tokens: list[ChecksumAddress], account: ChecksumAddress,
334
) -> dict[ChecksumAddress, int]:
335
calls = [multicall.encode_erc20_balance(token, account) for token in tokens]
336
results = await multicall.call(calls)
337
338
balances: dict[ChecksumAddress, int] = {}339
for token, result in zip(tokens, results, strict=True):
340
balances[token] = result.data[0] if result.success and result.data else 0
341
return balances
342
```
343
344
## Event log processing
345
346
Read and decode contract events with pagination:
347
348
```python
349
from web3.types import EventData
350
351
352
class EventReader:
353
def __init__(self, w3: AsyncWeb3, *, max_block_range: int = 2000) -> None:
354
self._w3 = w3
355
self._max_block_range = max_block_range
356
357
async def get_events(
358
self, contract: AsyncContract, event_name: str,
359
*, from_block: int, to_block: int | None = None,
360
argument_filters: dict | None = None,
361
) -> list[EventData]:
362
if to_block is None:
363
to_block = await self._w3.eth.block_number
364
365
event = getattr(contract.events, event_name)
366
all_events: list[EventData] = []
367
368
current = from_block
369
while current <= to_block:
370
chunk_end = min(current + self._max_block_range - 1, to_block)
371
filter_params = {"fromBlock": current, "toBlock": chunk_end}372
if argument_filters:
373
filter_params["argument_filters"] = argument_filters
374
375
events = await event.get_logs(**filter_params)
376
all_events.extend(events)
377
378
logger.debug("events_fetched", event=event_name, from_block=current, to_block=chunk_end, count=len(events))379
current = chunk_end + 1
380
381
return all_events
382
```
383
384
## Custom middleware for retries
385
386
```python
387
import asyncio
388
from web3.types import RPCEndpoint, RPCResponse
389
import structlog
390
391
logger = structlog.get_logger()
392
393
394
def retry_middleware(max_retries: int = 3, base_delay: float = 1.0):
395
async def middleware(make_request, w3: AsyncWeb3):
396
async def retrying_middleware(method: RPCEndpoint, params: tuple) -> RPCResponse:
397
last_error: Exception | None = None
398
399
for attempt in range(max_retries + 1):
400
try:
401
response = await make_request(method, params)
402
if "error" in response and response["error"].get("code") == -32005:403
raise RateLimitError(response["error"]["message"])
404
return response
405
except Exception as e:
406
last_error = e
407
if attempt < max_retries:
408
delay = base_delay * (2 ** attempt)
409
logger.warning("rpc_retry", method=method, attempt=attempt + 1, delay=delay, error=str(e))410
await asyncio.sleep(delay)
411
412
raise last_error # type: ignore[misc]
413
414
return retrying_middleware
415
return middleware
416
```
417
418
## Error handling
419
420
Typed errors for blockchain-specific failures:
421
422
```python
423
from dataclasses import dataclass
424
425
426
@dataclass(frozen=True, slots=True)
427
class Web3Error(Exception):
428
message: str
429
def __str__(self) -> str:
430
return self.message
431
432
433
@dataclass(frozen=True, slots=True)
434
class TransactionRevertedError(Web3Error):
435
tx_hash: str
436
receipt: dict
437
def __str__(self) -> str:
438
return f"Transaction {self.tx_hash} reverted"439
440
441
@dataclass(frozen=True, slots=True)
442
class ContractCallError(Web3Error):
443
contract_address: str
444
function_name: str
445
446
447
@dataclass(frozen=True, slots=True)
448
class RateLimitError(Web3Error):
449
pass
450
451
452
@dataclass(frozen=True, slots=True)
453
class InsufficientFundsError(Web3Error):
454
required: int
455
available: int
456
```
464
User asks: "Create a typed wrapper for a Uniswap V2 Router to get quotes and execute swaps"
466
```python
467
from dataclasses import dataclass
468
from decimal import Decimal
469
from eth_typing import ChecksumAddress
470
from web3 import AsyncWeb3
471
472
473
@dataclass(frozen=True, slots=True)
474
class SwapQuote:
475
path: list[ChecksumAddress]
476
amount_in: int
477
amount_out: int
478
price_impact: Decimal
479
480
481
class UniswapV2Router:
482
def __init__(self, w3: AsyncWeb3, router_address: ChecksumAddress) -> None:
483
self._w3 = w3
484
self._contract = w3.eth.contract(address=router_address, abi=load_abi("uniswap_v2_router"))485
486
async def get_amounts_out(self, amount_in: int, path: list[ChecksumAddress]) -> list[int]:
487
return await self._contract.functions.getAmountsOut(amount_in, path).call()
488
489
async def quote_exact_input(
490
self, token_in: ChecksumAddress, token_out: ChecksumAddress, amount_in: int,
491
*, intermediate: ChecksumAddress | None = None,
492
) -> SwapQuote:
493
path = [token_in, intermediate, token_out] if intermediate else [token_in, token_out]
494
amounts = await self.get_amounts_out(amount_in, path)
495
496
return SwapQuote(
497
path=path, amount_in=amount_in, amount_out=amounts[-1], price_impact=Decimal("0"),498
)
499
500
async def swap_exact_tokens_for_tokens(
501
self, tx_builder: TransactionBuilder, *, amount_in: int, amount_out_min: int,
502
path: list[ChecksumAddress], to: ChecksumAddress, deadline: int,
503
) -> TxReceipt:
504
return await tx_builder.send(
505
self._contract.functions.swapExactTokensForTokens(
506
amount_in, amount_out_min, path, to, deadline,
507
)
508
)
509
```
510
511
Usage:
512
```python
513
router = UniswapV2Router(w3, router_address=UNISWAP_V2_ROUTER)
514
quote = await router.quote_exact_input(
515
token_in=WETH_ADDRESS, token_out=USDC_ADDRESS, amount_in=10 ** 18,
516
)
517
print(f"1 WETH -> {quote.amount_out / 10**6} USDC")518
519
amount_out_min = int(quote.amount_out * 0.995) # 0.5% slippage
520
deadline = int(time.time()) + 300
521
522
receipt = await router.swap_exact_tokens_for_tokens(
523
tx_builder, amount_in=quote.amount_in, amount_out_min=amount_out_min,
524
path=quote.path, to=account.address, deadline=deadline,
525
)
526
```
531
User asks: "Fetch all ERC-20 balances for a wallet across 50 tokens in a single call"
533
[Two Multicall3 batches: Batch 1 fetches metadata (decimals+symbol) for all tokens, Batch 2 fetches balanceOf for each token. Results merged into `TokenBalance` dataclass with raw_balance (int) and human balance (Decimal). Two RPC calls total for 50 tokens instead of 150 individual calls. Skip tokens where any call failed.]
537
- `AsyncWeb3` for all blockchain interactions — sync Web3 blocks the event loop in async services
538
- Wrap raw Web3 contracts behind typed classes with domain-specific methods — prevents raw ABI calls scattered across codebase and catches type errors at development time
539
- `Decimal` for token amounts in application layer, `int` (raw units) at the contract call boundary — floating-point causes rounding errors that lose real money
540
- Multicall3 to batch read calls — one RPC round trip instead of N, critical for pages displaying multiple token balances or positions
541
- Cache immutable contract data (`decimals`, `symbol`, `name`) after first fetch — these values never change on-chain, no reason to re-fetch
542
- EIP-1559 fee model (`maxFeePerGas` + `maxPriorityFeePerGas`) for transactions — legacy gas pricing overpays and is deprecated on most EVM chains
543
- Retry middleware with exponential backoff for RPC calls — public and paid RPC providers rate-limit and have transient failures
544
- Paginate event log queries with bounded block ranges (2000-10000 blocks per request) — providers reject or timeout on unbounded getLogs calls
545
- Verify transaction receipt `status == 1` after every send — reverted transactions still produce receipts, failing to check causes silent loss of funds
546
- Store ABIs as JSON files, load with `@lru_cache` — keeps ABIs version-controlled and avoids re-reading disk on every contract instantiation
547
- `ChecksumAddress` type hint for all Ethereum addresses — catches mixed-case address errors at type-check time
548
- `structlog` for all RPC activity logging — include tx_hash, block number, gas used for debugging and audit trails
549
- Specific error types (TransactionRevertedError, RateLimitError) — enables targeted retry logic and clear error reporting upstream