Skip to content

Module: rpc

RPC

Wrapper around web3py to interact with Ethereum-compatible JSON-RPC endpoints.

This module provides a simple interface to interact with Ethereum-compatible JSON-RPC endpoints. It is designed to be used with the Infernet client library, but can be used independently.

Example

from infernet_client.chain.rpc import RPC


async def main():
    rpc = RPC("https://mainnet.infura.io/v3/your_project_id")
    rpc = await rpc.initialize_with_private_key("yourkey")

    chain_id = await rpc.get_chain_id()

Public Methods

  • initialize_with_private_key(private_key: str) -> RPC: Initializes RPC client with private key
  • get_checksum_address(address: str) -> ChecksumAddress: Returns a checksummed Ethereum address
  • get_contract(address: ChecksumAddress, abi: ABI) -> AsyncContract: Returns a web3py async contract instance
  • get_balance(address: ChecksumAddress) -> int: Collects balance for an address
  • get_nonce(address: ChecksumAddress) -> Nonce: Collects nonce for an address
  • get_chain_id() -> int: Collects connected RPC's chain ID
  • get_tx_receipt(tx_hash: HexBytes): Returns transaction receipt
  • send_transaction(tx: TxParams) -> HexBytes: Sends a transaction

RPC

Source code in src/infernet_client/chain/rpc.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class RPC:
    def __init__(self, rpc_url: str) -> None:
        """Initializes new Ethereum-compatible JSON-RPC client

        Args:
            rpc_url (str): HTTP(s) RPC url

        Raises:
            ValueError: RPC URL is incorrectly formatted
        """

        # Setup new Web3 HTTP provider w/ 10 minute timeout
        # Long timeout is useful for event polling, subscriptions
        provider = AsyncHTTPProvider(
            endpoint_uri=rpc_url, request_kwargs={"timeout": 60 * 10}
        )

        self._web3: AsyncWeb3 = AsyncWeb3(provider)
        self._account: Optional[Account] = None

    @property
    def account(self: RPC) -> Account:
        """Returns account instance, if it exists

        Returns:
            Account: Account instance
        """

        if self._account is None:
            raise ValueError("Account not initialized")
        return self._account

    async def initialize_with_private_key(self: RPC, private_key: str) -> RPC:
        """Initializes RPC client with private key

        Args:
            private_key (str): Private key
        """
        account = self._web3.eth.account.from_key(private_key)
        self._account = account
        self._web3.middleware_onion.add(
            await async_construct_sign_and_send_raw_middleware(account)
        )
        self._web3.eth.default_account = account.address
        return self

    def get_checksum_address(self: RPC, address: str) -> ChecksumAddress:
        """Returns a checksummed Ethereum address

        Args:
            address (str): Stringified address

        Returns:
            ChecksumAddress: Checksum-validated Ethereum address
        """
        return self._web3.to_checksum_address(address)

    def get_contract(
        self,
        address: Optional[ChecksumAddress] = None,
        abi: Optional[ABI] = None,
        bytecode: Optional[bytes] = None,
    ) -> AsyncContract | Type[AsyncContract]:
        """Returns a web3py async contract instance

        Args:
            address (Optional[ChecksumAddress]): Contract address. Defaults to None.
            abi (Optional[ABI]): Contract ABI. Defaults to None.
            bytecode (Optional[bytes]): Contract bytecode. Defaults to
                None.

        Returns:
            AsyncContract | Type[AsyncContract]: Web3py async contract instance
        """
        return self._web3.eth.contract(address=address, abi=abi, bytecode=bytecode)

    async def get_balance(self, address: ChecksumAddress) -> int:
        """Collects balance for an address

        Args:
            address (ChecksumAddress): Address to collect balance

        Returns:
            int: Balance in wei
        """
        return await self._web3.eth.get_balance(address)

    async def get_nonce(self, address: ChecksumAddress) -> Nonce:
        """Collects nonce for an address

        Args:
            address (ChecksumAddress): Address to collect tx count

        Returns:
            Nonce: Transaction count (nonce)
        """
        return await self._web3.eth.get_transaction_count(address)

    async def get_chain_id(self) -> int:
        """Collects connected RPC's chain ID

        Returns:
            int: Chain ID
        """
        return await self._web3.eth.chain_id

    async def get_tx_receipt(self, tx_hash: HexBytes) -> TxReceipt:
        """Returns transaction receipt
        Args:
            tx_hash (HexBytes): Transaction hash
        """
        return await self._web3.eth.wait_for_transaction_receipt(tx_hash)

    async def send_transaction(self, tx: TxParams) -> HexBytes:
        """Sends a transaction

        Args:
            tx (dict): Transaction dictionary

        Returns:
            HexBytes: Transaction hash
        """
        return await self._web3.eth.send_transaction(tx)

account: Account property

Returns account instance, if it exists

Returns:

Name Type Description
Account Account

Account instance

__init__(rpc_url)

Initializes new Ethereum-compatible JSON-RPC client

Parameters:

Name Type Description Default
rpc_url str

HTTP(s) RPC url

 required

Raises:

Type Description
ValueError

RPC URL is incorrectly formatted
Source code in src/infernet_client/chain/rpc.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(self, rpc_url: str) -> None:
    """Initializes new Ethereum-compatible JSON-RPC client

    Args:
        rpc_url (str): HTTP(s) RPC url

    Raises:
        ValueError: RPC URL is incorrectly formatted
    """

    # Setup new Web3 HTTP provider w/ 10 minute timeout
    # Long timeout is useful for event polling, subscriptions
    provider = AsyncHTTPProvider(
        endpoint_uri=rpc_url, request_kwargs={"timeout": 60 * 10}
    )

    self._web3: AsyncWeb3 = AsyncWeb3(provider)
    self._account: Optional[Account] = None

get_balance(address) async

Collects balance for an address

Parameters:

Name Type Description Default
address ChecksumAddress

Address to collect balance

 required

Returns:

Name Type Description
int int

Balance in wei
Source code in src/infernet_client/chain/rpc.py 
126
127
128
129
130
131
132
133
134
135
async def get_balance(self, address: ChecksumAddress) -> int:
    """Collects balance for an address

    Args:
        address (ChecksumAddress): Address to collect balance

    Returns:
        int: Balance in wei
    """
    return await self._web3.eth.get_balance(address)

get_chain_id() async

Collects connected RPC's chain ID

Returns:

Name Type Description
int int

Chain ID
Source code in src/infernet_client/chain/rpc.py 
148
149
150
151
152
153
154
async def get_chain_id(self) -> int:
    """Collects connected RPC's chain ID

    Returns:
        int: Chain ID
    """
    return await self._web3.eth.chain_id

get_checksum_address(address)

Returns a checksummed Ethereum address

Parameters:

Name Type Description Default
address str

Stringified address

 required

Returns:

Name Type Description
ChecksumAddress ChecksumAddress

Checksum-validated Ethereum address
Source code in src/infernet_client/chain/rpc.py 
 96
 97
 98
 99
100
101
102
103
104
105
def get_checksum_address(self: RPC, address: str) -> ChecksumAddress:
    """Returns a checksummed Ethereum address

    Args:
        address (str): Stringified address

    Returns:
        ChecksumAddress: Checksum-validated Ethereum address
    """
    return self._web3.to_checksum_address(address)

get_contract(address=None, abi=None, bytecode=None)

Returns a web3py async contract instance

Parameters:

Name Type Description Default
address Optional[ChecksumAddress]

Contract address. Defaults to None.

 None
abi Optional[ABI]

Contract ABI. Defaults to None.

 None
bytecode Optional[bytes]

Contract bytecode. Defaults to None.

 None

Returns:

Type Description
AsyncContract | Type[AsyncContract]

AsyncContract | Type[AsyncContract]: Web3py async contract instance
Source code in src/infernet_client/chain/rpc.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def get_contract(
    self,
    address: Optional[ChecksumAddress] = None,
    abi: Optional[ABI] = None,
    bytecode: Optional[bytes] = None,
) -> AsyncContract | Type[AsyncContract]:
    """Returns a web3py async contract instance

    Args:
        address (Optional[ChecksumAddress]): Contract address. Defaults to None.
        abi (Optional[ABI]): Contract ABI. Defaults to None.
        bytecode (Optional[bytes]): Contract bytecode. Defaults to
            None.

    Returns:
        AsyncContract | Type[AsyncContract]: Web3py async contract instance
    """
    return self._web3.eth.contract(address=address, abi=abi, bytecode=bytecode)

get_nonce(address) async

Collects nonce for an address

Parameters:

Name Type Description Default
address ChecksumAddress

Address to collect tx count

 required

Returns:

Name Type Description
Nonce Nonce

Transaction count (nonce)
Source code in src/infernet_client/chain/rpc.py 
137
138
139
140
141
142
143
144
145
146
async def get_nonce(self, address: ChecksumAddress) -> Nonce:
    """Collects nonce for an address

    Args:
        address (ChecksumAddress): Address to collect tx count

    Returns:
        Nonce: Transaction count (nonce)
    """
    return await self._web3.eth.get_transaction_count(address)

get_tx_receipt(tx_hash) async

Returns transaction receipt Args: tx_hash (HexBytes): Transaction hash

Source code in src/infernet_client/chain/rpc.py 
156
157
158
159
160
161
async def get_tx_receipt(self, tx_hash: HexBytes) -> TxReceipt:
    """Returns transaction receipt
    Args:
        tx_hash (HexBytes): Transaction hash
    """
    return await self._web3.eth.wait_for_transaction_receipt(tx_hash)

initialize_with_private_key(private_key) async

Initializes RPC client with private key

Parameters:

Name Type Description Default
private_key str

Private key

 required
Source code in src/infernet_client/chain/rpc.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
async def initialize_with_private_key(self: RPC, private_key: str) -> RPC:
    """Initializes RPC client with private key

    Args:
        private_key (str): Private key
    """
    account = self._web3.eth.account.from_key(private_key)
    self._account = account
    self._web3.middleware_onion.add(
        await async_construct_sign_and_send_raw_middleware(account)
    )
    self._web3.eth.default_account = account.address
    return self

send_transaction(tx) async

Sends a transaction

Parameters:

Name Type Description Default
tx dict

Transaction dictionary

 required

Returns:

Name Type Description
HexBytes HexBytes

Transaction hash
Source code in src/infernet_client/chain/rpc.py 
163
164
165
166
167
168
169
170
171
172
async def send_transaction(self, tx: TxParams) -> HexBytes:
    """Sends a transaction

    Args:
        tx (dict): Transaction dictionary

    Returns:
        HexBytes: Transaction hash
    """
    return await self._web3.eth.send_transaction(tx)