Skip to content

Instantly share code, notes, and snippets.

@jakublipinski
Last active February 7, 2023 07:40
Show Gist options
  • Save jakublipinski/4f1dd698e2bdefdba53d8c69eba3da80 to your computer and use it in GitHub Desktop.
Save jakublipinski/4f1dd698e2bdefdba53d8c69eba3da80 to your computer and use it in GitHub Desktop.
Web3 JSON RPC
import timeit
import asyncio
import contextlib
import requests
from aiohttp import ClientSession
from web3.providers.base import JSONBaseProvider
from web3.providers import HTTPProvider
from web3 import Web3
@contextlib.contextmanager
def profile(name):
start_time = timeit.default_timer()
yield
print('{}: {:.3f}s'.format(name, timeit.default_timer() - start_time))
# synchronously request receipts for given transactions
def sync_receipts(web3, transactions):
for tran in transactions:
web3.eth.getTransactionReceipt(tran)
# asynchronous JSON RPC API request
async def async_make_request(session, url, method, params):
base_provider = JSONBaseProvider()
request_data = base_provider.encode_rpc_request(method, params)
async with session.post(url, data=request_data,
headers={'Content-Type': 'application/json'}) as response:
content = await response.read()
response = base_provider.decode_rpc_response(content)
return response
async def run(node_address, transactions):
tasks = []
# Fetch all responses within one Client session,
# keep connection alive for all requests.
async with ClientSession() as session:
for tran in transactions:
task = asyncio.ensure_future(async_make_request(session, node_address,
'eth_getTransactionReceipt',[tran.hex()]))
tasks.append(task)
responses = await asyncio.gather(*tasks)
# batch request
def batch(node_address, transactions):
base_provider = JSONBaseProvider()
request_data = b'[' + b','.join(
[base_provider.encode_rpc_request('eth_getTransactionReceipt', [tran.hex()]) for tran in transactions]
) + b']'
r = requests.post(eth_node_address, data=request_data, headers={'Content-Type': 'application/json'})
responses = base_provider.decode_rpc_response(r.content)
if __name__ == "__main__":
eth_node_address = "http://localhost:8545"
web3 = Web3(HTTPProvider(eth_node_address))
transactions = []
for i in range(0,20):
block = web3.eth.getBlock(web3.eth.blockNumber-i)
transactions.extend(block['transactions'])
with profile('sync'):
sync_receipts(web3, transactions)
with profile('async'):
loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run(eth_node_address, transactions))
loop.run_until_complete(future)
with profile('batch'):
batch(eth_node_address, transactions)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment