transaction.py

from typing import List, Optional

from eth_utils import to_checksum_address
from hexbytes import HexBytes
from msgspec import Struct


class address(str):
    pass


class uint(int):
    pass


class hash32(HexBytes):
    pass


transaction_rename = {
    "sender": "from",  # the culprit
    "chain_id": "chainId",
    "max_fee_per_gas": "maxFeePerGas",
    "max_priority_fee_per_gas": "maxPriorityFeePerGas",
    "access_list": "accessList",
    "gas_price": "gasPrice",
    "block_hash": "blockHash",
    "block_number": "blockNumber",
    "transaction_index": "transactionIndex",
}
rename_fn = lambda item: transaction_rename.get(item, item)


class AccessListEntry(Struct, rename="camel"):
    address: address
    storage_keys: List[hash32]


AccessList = List[AccessListEntry]


class TransactionBase(Struct, rename=rename_fn):
    # `type` field is omitted since it's used in the tagged union

    nonce: uint
    gas: uint
    value: uint
    input: HexBytes
    chain_id: Optional[uint]
    to: Optional[address] = None

    # details
    block_hash: hash32
    block_number: uint
    sender: address
    hash: hash32
    transaction_index: uint

    # signature
    v: uint
    r: uint
    s: uint


class Transaction1559(TransactionBase, tag="0x2"):
    max_fee_per_gas: uint
    max_priority_fee_per_gas: uint
    access_list: Optional[AccessList] = None


class Transaction2930(TransactionBase, tag="0x1"):
    gas_price: uint
    access_list: Optional[AccessList] = None


class TransactionLegacy(TransactionBase, tag="0x0"):
    gas_price: uint


Transaction = Transaction1559 | Transaction2930 | TransactionLegacy


class Block(Struct, rename="camel"):
    parent_hash: hash32
    sha3_uncles: hash32
    miner: address
    state_root: hash32
    transactions_root: hash32
    receipts_root: hash32
    logs_bloom: HexBytes
    number: uint
    gas_limit: uint
    gas_used: uint
    timestamp: uint
    extra_data: HexBytes
    mix_hash: hash32
    nonce: HexBytes
    size: uint
    uncles: List[hash32]
    transactions: List[Transaction]
    
    
class Response(Struct):
    result: Raw = None  # type: ignore
    error: Optional[Error] = None
    id: Optional[str | int] = None
    jsonrpc: Literal["2.0"] = "2.0"
    

class Error(Struct):
    code: int
    message: str
    data: Optional[Any] = None


def dec_hook(type: Type, obj: Any) -> Any:
    if type is uint:
        return uint(obj, 16)
    if type in (HexBytes, hash32):
        return type(obj)
    if type is address:
        return type(to_checksum_address(obj))


response_decoder = Decoder(Response).decode(resp.content)
block_decoder = Decoder(Block, dec_hook=dec_hook)

def decode_block(buffer: bytes) -> Block:
    response = response_decoder.decode(buffer)
    if response.error:
        raise ValueError(response.error.message)

    return block_decoder.decode(response.result)