Class JSONRPCClient

raiden/network/rpc/client.py:1018–1646

Ethereum JSON RPC client.

class JSONRPCClient:
    """Ethereum JSON RPC client."""

    def __init__(
        self,
        web3: Web3,
        privkey: PrivateKey,
        gas_price_strategy: Callable = rpc_gas_price_strategy,
        block_num_confirmations: int = 0,
    ) -> None:
        if len(privkey) != 32:
            raise ValueError("Invalid private key")

        if block_num_confirmations < 0:
            raise ValueError("Number of confirmations has to be positive")

        monkey_patch_web3(web3, gas_price_strategy)

        version = web3.clientVersion
        supported, eth_node, _ = is_supported_client(version)

        if eth_node is None or supported is VersionSupport.UNSUPPORTED:
            raise EthNodeInterfaceError(f'Unsupported Ethereum client "{version}"')
        if supported is VersionSupport.WARN:
            log.warning(f'Unsupported Ethereum client version "{version}"')

        address = privatekey_to_address(privkey)
        available_nonce = discover_next_available_nonce(web3, eth_node, address)

        self.eth_node = eth_node
        self.privkey = privkey
        self.address = address
        self.web3 = web3
        self.default_block_num_confirmations = block_num_confirmations

        # Ask for the chain id only once and store it here
        self.chain_id = ChainID(self.web3.eth.chain_id)

        self._available_nonce = available_nonce
        self._nonce_lock = Semaphore()

        log.debug(
            "JSONRPCClient created",
            node=to_checksum_address(self.address),
            available_nonce=available_nonce,
            client=version,
        )

    def __repr__(self) -> str:
        return (
            f"<JSONRPCClient "
            f"node:{to_checksum_address(self.address)} nonce:{self._available_nonce}"
            f">"
        )

    def block_number(self) -> BlockNumber:
        """Return the most recent block."""
        return self.web3.eth.block_number