stage_2_example.py

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)
    res.raise_for_status()
    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 = requests.post(worker_url, json=query)
        res.raise_for_status()
        blocks = res.json()

        last_processed_block = blocks[-1]['header']['number']
        next_block = last_processed_block + 1
        for block in blocks:
            print(json.dumps(block))


if __name__ == '__main__':
    # dump USDC transfers
    dump(
        'https://v2.archive.subsquid.io/network/ethereum-mainnet',
        query={
            "logs": [
                {
                    "address": ["0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"],
                    "topic0": [
                        "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
                    ]
                }
            ],
            "fields": {
                "block": {},
                "log": {
                    "topics": True,
                    "data": True
                }
            }
        },
        first_block=16_000_000,
        last_block=16_010_000
    )