pythereum 1.0.4

A lightweight Ethereum JSON RPC library written in python

pythereum

(formerly eth_rpc)

A lightweight Ethereum RPC library for Python

Features include:

• Ability to initiate remote procedure calls on a wide variety of ethereum functions
  • More functions are added whenever possible to this list
• Typed function outputs for easy manipulation of results
• "eth_subscribe" functionality
• Websocket pooling for high performance calls
• Support for RPC batching, allowing multiple calls to the same function at once
• Currency conversion for wei, so you don't have to rely on external libs like Web3.py

Implemented RPC methods

• eth_blockNumber
• eth_getTransactionCount
• eth_getBalance
• eth_gasPrice
• eth_getBlockByNumber
• eth_getblockByHash
• eth_call
• eth_getTransactionReceipt
• eth_sendRawTransaction
• eth_sendTransaction
• eth_syncing
• eth_coinbase
• eth_chainId
• eth_mining
• eth_hashrate
• eth_accounts
• eth_subscribe
• eth_getBlockTransactionCountByHash
• eth_getBlocktransactionCountbyNumber
• eth_getUncleCountByBlockHash
• eth_getUncleCountByBlockNumber
• eth_getCode
• eth_estimateGas
• eth_sign
• eth_getTransactionByHash
• eth_getTransactionByBlockHashAndIndex
• eth_getTransactionByBlockNumberAndIndex
• eth_getUncleByBlockHashAndIndex
• eth_getUncleByBlockNumberAndIndex
• eth_newFilter
  • Additional testing needed for batching this function
• eth_newBlockFilter
• eth_newPendingTransactionFilter
• eth_uninstallFilter
• eth_getFilterLogs
• eth_getFilterChanges

RPC methods to implement

• Aiming to complete all methods listed here.

Example usage

Basic single function call

# Example simple function
import asyncio
from pythereum import EthRPC

TEST_URL = "ws://127.0.0.1:8545"
erpc = EthRPC(TEST_URL, pool_size=2)


async def test_transaction_count():
  # Optional step to start your thread pool before your RPC call
  await erpc.start_pool()
  # Gets the number of transactions sent from a given EOA address
  r = await erpc.get_transaction_count("0xabcdefghijklmnopqrstuvwxyz1234567890")
  print(r)
  # Ensures no hanging connections are left
  await erpc.close_pool()


if __name__ == "__main__":
  asyncio.run(test_transaction_count())

Example subscription

# Example subscription
import asyncio
from pythereum import EthRPC, SubscriptionType

TEST_URL = "ws://127.0.0.1:8545"
erpc = EthRPC(TEST_URL, pool_size=2)


async def test_subscription(subscription_type: SubscriptionType):
  """
  Creates a subscription to receive data about all new heads
  Prints each new subscription result as it is received
  """
  async with erpc.subscribe(subscription_type) as sc:
    # The following will iterate as each item is gotten by sc.recv()
    async for item in sc.recv():
      # 'item' is formatted into the appropriate form for its subscription type
      # this is done by the sc.recv() automatically
      print(item)


if __name__ == "__main__":
  asyncio.run(test_subscription(SubscriptionType.new_heads))

Example batch call

# Example batch call
import asyncio
from pythereum import EthRPC, BlockTag

TEST_URL = "ws://127.0.0.1:8545"
erpc = EthRPC(TEST_URL, pool_size=2)


async def test_batching():
  await erpc.start_pool()
  # Batch calls can be applied to any parameterised method
  # Each parameter must be passed in as a list 
  # With list length k where k is the batch size
  r = await erpc.get_block_by_number(
    block_specifier=[
      i for i in range(40000, 40010)
    ],
    full_object=[
      True for i in range(10)
    ]
  )
  print(r)
  await erpc.close_pool()


if __name__ == "__main__":
  asyncio.run(test_batching())

Example currency conversion

>>> from pythereum import EthDenomination, convert_eth
>>> convert_eth(1_000_000, convert_from=EthDenomination.wei, covert_to=EthDenomination.ether)
1e-12
>>> convert_eth(1_000, convert_from=EthDenomination.babbage, covert_to=EthDenomination.finney)
1e-09

More examples available in the demo folder.

Getting started

Poetry

This project and its dependencies are managed by python poetry, which will automatically manage the versions of each library / python version upon which this project depends.

Install poetry with the instructions here.

Installation

Poetry Installation

The library currently requires python versions >=3.11,<3.13.

If you want to include this library for use in another project via Poetry you must simply add the following to your pyproject.toml file under [tool.poetry.dependencies]

pythereum = {git = "https://github.com/gabedonnan/pythereum.git"}

or

pythereum = "^1.0.4"

If you would like to install the library via pypi instead of via this git repository. This git repository will always be the most up to date but the releases on pypi should be more thoroughly tested.

Pip / PyPi installation

The library is now available via pip!! (I had to change the whole project name to get it there)

It can be installed with the following command, or downloaded here:

python3 -m pip install pythereum

Testing your programs

Testing a program built with this library can be done with actual ethereum nodes, though they may rate limit you or cost eth to run. As such using testing programs such as Anvil from the Foundry suite of products allows for faster and more productive testing.

Install foundry

Instructions available at this link.

Run anvil

Anvil is a blockchain testing application included with foundry.

The following command will run an instance of anvil representing the blockchain's status at block number EXAMPLE_BLOCK_NUM via url EXAMPLE_RPC_URL.

This is helpful for ensuring consistency in tests.

anvil rpc-url EXAMPLE_RPC_URL@EXAMPLE_BLOCK_NUM

Acknowledgements

Special thanks to @totlsota as a more experienced blockchain developer than I, for giving me pointers when I needed them and generally assisting in the development of this project.