Web3

web3 wrapper 
x = 'a'
isinstance(x, (list, dict))
False

source

MultiRPCWeb3

 MultiRPCWeb3 (*rpcs, providers=None, poa=False)

Web3 object that tries to execute a method in multiple RPCs until one succeeds.

source

MultiRPCWeb3.sort_providers

 MultiRPCWeb3.sort_providers (tolerance:int=0)

Sort provider by block number, so that the one with the highest block number is first.

Type Default Details
tolerance int 0 number of blocks to consider the same

source

MultiRPCWeb3.__getitem__

 MultiRPCWeb3.__getitem__ (block_number:int)

The user can get the timestamp of a block by using the syntax web3[block_number].

Type Details
block_number int block number to get the timestamp of

source

MultiRPCWeb3.find_block_at_timestamp

 MultiRPCWeb3.find_block_at_timestamp
                                       (timestamp:Union[datetime.datetime,
                                       int], low:Optional[int]=None,
                                       high:Optional[int]=None, how:Litera
                                       l['after','before']='after')

Finds a block at a specific timestamp.

Type Default Details
timestamp Union timestamp to search for
low Optional None block number to start the search from
high Optional None block number to end the search at
how Literal after whether to search for the block after or before the timestamp

source

MultiRPCWeb3.abatch_method

 MultiRPCWeb3.abatch_method (method, input_list, method_args=(),
                             method_kwargs={}, batch_size=72)

Execute a method asynchronously in batches.

Type Default Details
method method to execute
input_list list of arguments to pass to the method
method_args tuple () extra arguments to pass to the method
method_kwargs dict {} extra keyword arguments to pass to the method
batch_size int 72 batch size to use

Example

from datetime import datetime, timezone, timedelta
w3 = MultiRPCWeb3.from_rpcs('http://bad_provider:1234', 'http://localhost:8545')
test_eq(['http://bad_provider:1234', 'http://localhost:8545'], [p.endpoint_uri for p in w3.providers])
w3.sort_providers(tolerance=0)
test_eq(['http://localhost:8545', 'http://bad_provider:1234'], [p.endpoint_uri for p in w3.providers])
RPC connection http://bad_provider:1234 failed with: HTTPConnectionPool(host='bad_provider', port=1234): Max retries exceeded with url: / (Caused by NameResolutionError("<urllib3.connection.HTTPConnection object>: Failed to resolve 'bad_provider' ([Errno 8] nodename nor servname provided, or not known)"))
w3 = MultiRPCWeb3.from_rpcs('http://localhost:8545')
test_eq(len(w3), w3.eth.get_block_number())
test_eq(w3.find_block_at_timestamp(datetime(2022, 10, 1, tzinfo=timezone.utc).timestamp(), low=FIRST_POS_BLOCK, how='after'), 15649595)
test_fail(w3.find_block_at_timestamp, args=(datetime.today() + timedelta(days=1),), kwargs=dict(low=FIRST_POS_BLOCK, how='after'))

Async Interface

aw3 = MultiRPCWeb3.async_from_rpcs('http://bad_provider:1234', 'http://localhost:8545')
test_eq(['http://bad_provider:1234', 'http://localhost:8545'], [p.endpoint_uri for p in aw3.providers])
await aw3.asort_providers(tolerance=0)
test_eq(['http://localhost:8545', 'http://bad_provider:1234'], [p.endpoint_uri for p in aw3.providers])
RPC connection http://bad_provider:1234 failed with: Cannot connect to host bad_provider:1234 ssl:default [nodename nor servname provided, or not known]
RPC connection http://bad_provider:1234 failed with: Cannot connect to host bad_provider:1234 ssl:default [nodename nor servname provided, or not known]
aw3 = MultiRPCWeb3.async_from_rpcs('http://localhost:8545')
test_eq(await aw3.alen(), await aw3.eth.get_block_number())
test_eq(await aw3.agetitem(19989282), 1717152047)
test_eq(await aw3.afind_block_at_timestamp(datetime(2022, 10, 1, tzinfo=timezone.utc).timestamp(), low=FIRST_POS_BLOCK, how='after'), 15649595)

Get Many Blocks Fast with abatch_method

import web3

aw3 = MultiRPCWeb3.async_from_rpcs('http://localhost:8545')
block_numbers = range(FIRST_POS_BLOCK, FIRST_POS_BLOCK + 72)
blocks = await aw3.abatch_method('get_block', block_numbers, method_args=(True,), method_kwargs={}, batch_size=72)
test_eq(len(blocks), 72)
test_eq(all([isinstance(b, web3.datastructures.AttributeDict) for b in blocks]), True)