Skip to content

Instantly share code, notes, and snippets.

Last active May 10, 2023 10:44
Show Gist options
  • Save eldargab/2e007a293ac9f82031d023f1af581a7d to your computer and use it in GitHub Desktop.
Save eldargab/2e007a293ac9f82031d023f1af581a7d to your computer and use it in GitHub Desktop.
import json
from typing import TypedDict, NotRequired
import requests
class BlockFieldSelection(TypedDict, total=False):
number: bool
hash: bool
parentHash: bool
timestamp: bool
transactionsRoot: bool
receiptsRoot: bool
stateRoot: bool
logsBloom: bool
sha3Uncles: bool
extraData: bool
miner: bool
nonce: bool
mixHash: bool
size: bool
gasLimit: bool
gasUsed: bool
difficulty: bool
totalDifficulty: bool
baseFeePerGas: bool
TxFieldSelection = TypedDict('TxFieldSelection', {
'transactionIndex': bool,
'hash': bool,
'nonce': bool,
'from': bool,
'to': bool,
'input': bool,
'value': bool,
'gas': bool,
'gasPrice': bool,
'maxFeePerGas': bool,
'maxPriorityFeePerGas': bool,
'v': bool,
'r': bool,
's': bool,
'yParity': bool,
'chainId': bool,
'sighash': bool,
'gasUsed': bool,
'cumulativeGasUsed': bool,
'effectiveGasUsed': bool,
'type': bool,
'status': bool
}, total=False)
class LogFieldSelection(TypedDict, total=False):
logIndex: bool
transactionIndex: bool
transactionHash: bool
address: bool
data: bool
topics: bool
class TraceFieldSelection(TypedDict, total=False):
traceAddress: bool
subtraces: bool
transactionIndex: bool
type: bool
error: bool
createFrom: bool
createValue: bool
createGas: bool
createInit: bool
createResultGasUsed: bool
createResultCode: bool
createResultAddress: bool
callFrom: bool
callTo: bool
callValue: bool
callGas: bool
callInput: bool
callType: bool
callResultGasUsed: bool
callResultOutput: bool
suicideAddress: bool
suicideRefundAddress: bool
suicideBalance: bool
rewardAuthor: bool
rewardValue: bool
rewardType: bool
class StateDiffFieldSelection(TypedDict, total=False):
transactionIndex: bool
address: bool
key: bool
kind: bool
prev: bool
next: bool
class FieldSelection(TypedDict, total=False):
block: BlockFieldSelection
transaction: TxFieldSelection
log: LogFieldSelection
trace: TraceFieldSelection
stateDiff: StateDiffFieldSelection
class LogRequest(TypedDict, total=False):
address: list[str]
topic0: list[str]
transaction: bool
TxRequest = TypedDict('TxRequest', {
'from': list[str],
'to': list[str],
'sighash': list[str],
'logs': bool,
'traces': bool,
'stateDiffs': bool
}, total=False)
class TraceRequest(TypedDict, total=False):
type: list[str]
createFrom: list[str]
callFrom: list[str]
callTo: list[str]
callSighash: list[str]
suicideRefundAddress: list[str]
rewardAuthor: list[str]
transaction: bool
subtraces: bool
class StateDiffRequest(TypedDict, total=False):
address: list[str]
key: list[str]
kind: list[str]
transaction: bool
class Query(TypedDict):
# fromBlock: int
# toBlock: NotRequired[int]
includeAllBlocks: NotRequired[bool]
fields: NotRequired[FieldSelection]
logs: NotRequired[list[LogRequest]]
transactions: NotRequired[list[TxRequest]]
traces: NotRequired[list[TraceRequest]]
stateDiffs: NotRequired[list[StateDiffRequest]]
def get_text(url: str) -> str:
res = requests.get(url)
return res.text
def dump(
archive_url: str,
query: Query,
first_block: int,
last_block: int
) -> None:
assert 0 <= first_block <= last_block
query = dict(query) # copy query to mess with it later
archived_height = int(get_text(f'{archive_url}/height'))
next_block = first_block
last_block = min(last_block, archived_height)
while next_block <= last_block:
# FIXME: retries for 503, 504, 502 and network failures
# are required for a sequence of 2 queries below
worker_url = get_text(f'{archive_url}/{next_block}/worker')
query['fromBlock'] = next_block
query['toBlock'] = last_block
res =, json=query)
blocks = res.json()
last_processed_block = blocks[-1]['header']['number']
next_block = last_processed_block + 1
for block in blocks:
if __name__ == '__main__':
# dump USDC transfers
"logs": [
"address": ["0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"],
"topic0": [
"fields": {
"block": {},
"log": {
"topics": True,
"data": True
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment