ETHHelper API

Note

All interfaces in this repo rely on the asynchronous environment, so the asyncio environment is required when using the REPL for interactive coding. This can be done by running python -m asyncio directly on the command line or shell (instead of python).

High-level API 

GethHttpConnector 

class ethhelper.GethHttpConnector(url: str, logger: Logger | None = None, graphql_url: str | None = None)

GethHttpConnector is an asynchronous wrapper for all HTTP interfaces supported by ETHHelper.

Due to the flat design style, all interfaces use a simple naming structure as the member methods of this class.

The url is used to indicate the path of the HTTP service of the Geth node, usually in the form of http://host:port/. The use of third-party nodes may be out of the ordinary.

The logger is used to assign a logger of the Python logging module to this class. Explicitly assigning a logger can be used to control the output location of the logger, which is convenient for debugging. If this parameter is not provided or None is giving, this class will automatically call logging.getLogger("GethHttpConnector") to generate a default logger.

The graphql_url is used to specify the URL for the GraphQL interface of the Geth node, usually in the form of http://host:port/graphql. If not provided, it will generate from url by appending /graphql.

async eth_accounts() → list[ethhelper.datatypes.base.Address]

Get a list of all accounts in the currently connected Ethereum node.

Returns:: A list containing the addresses of all accounts in the current node.

async eth_block_number() → BlockNumber

Get the latest block number on the current Ethereum node.

Returns:: The latest block number on the current Ethereum node.

async eth_call(transaction: TxParams, block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 = 'latest', state_override: dict[eth_typing.evm.ChecksumAddress, ethhelper.datatypes.geth.CallOverrideParams] = {}, ccip_read_enabled: bool = False) → HexBytes

Execute a read-only transaction (which won’t change the blockchain state) and return the result.

Parameters:

transaction – The transaction parameters.
block_identifier – The block identifier. Defaults to “latest”.
state_override – The state override object. Defaults to {}.
ccip_read_enabled – Whether to enable CCIP read. Defaults to False.

Returns:

The result of the executed transaction as a HexBytes object.

async eth_chain_id() → int

Get the chain ID of the current Ethereum node.

Returns:: The chain ID of the current Ethereum node.

async eth_estimate_gas(transaction: TxParams, block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 = 'latest') → Gas

Estimate the gas cost of a transaction.

Parameters:

transaction – The transaction parameters.
block_identifier – The block identifier. Defaults to “latest”.

Returns:

The estimated gas cost of the transaction as a Gas object.

async eth_fee_history(block_count: int, newest_block: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber, reward_percentiles: list[float] | None = None) → FeeHistory

Retrieve historical fee data for a specified number of the latest blocks.

Parameters:

block_count – The number of blocks to retrieve.
newest_block – The parameter or block number of the latest block.
reward_percentiles – The list of fee percentile values to retrieve. Defaults to None.

Returns:

A FeeHistory object containing the historical fee data.

async eth_filter(filter: Literal['latest', 'pending'] | FilterParams) → AsyncFilter

Create a filter to watch for specific events on the Ethereum network.

Parameters:: filter – The filter specification.
Returns:: The created filter as an AsyncFilter object.

async eth_gas_price() → Wei

Get the latest average gas price on the current Ethereum node.

Returns:: The latest average gas price on the current Ethereum node.

async eth_get_account_nonce(account: Address | ENS, block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 = 'latest') → Nonce

Get the nonce of an Ethereum account.

Parameters:

account – The address or ENS name of the account.
block_identifier – The block identifier. Defaults to “latest”.

Returns:

The nonce of the account as a Nonce object.

async eth_get_balance(account: Address | ENS, block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 = 'latest') → Wei

Get the balance of an Ethereum account.

Parameters:

account – The address or ENS name of the account.
block_identifier – The block identifier. Defaults to “latest”.

Returns:

The balance of the account as a Wei object.

async eth_get_block(block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32, full_transactions: bool = False) → Block

Get information about a specific block.

Parameters:

block_identifier – The block identifier.
full_transactions – Whether to include full transaction details. Defaults to False.

Returns:

The block information as a Block object.

Return type:

Block

async eth_get_code(account: Address | ENS, block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 = 'latest') → HexBytes

Get the bytecode of a contract deployed on the Ethereum network.

Parameters:

account – The address or ENS name of the contract.
block_identifier – The block identifier. Defaults to “latest”.

Returns:

The bytecode of the contract as a HexBytes object.

async eth_get_raw_transaction(transaction_hash: Hash32) → HexBytes

Get the raw bytes of a specific transaction by its hash.

Parameters:: transaction_hash – The hash of the transaction to retrieve.
Returns:: The raw bytes of the transaction as a HexBytes object.

async eth_get_raw_transaction_by_block(block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32, index: int) → HexBytes

Get the raw bytes of a specific transaction by its block and index.

Parameters:

block_identifier – The block identifier containing the transaction.
index – The index of the transaction within the block.

Returns:

The raw bytes of the transaction as a HexBytes object.

async eth_get_storage_at(account: Address | ENS, position: int, block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 = 'latest') → HexBytes

Get the value stored at a specific position in an Ethereum account’s storage.

Parameters:

account – The address or ENS name of the account.
position – The position in storage to retrieve.
block_identifier – The block identifier. Defaults to “latest”.

Returns:

The value stored at the specified position as a HexBytes object.

async eth_get_transaction(transaction_hash: Hash32) → Transaction

Get information about a specific transaction by its hash.

Parameters:: transaction_hash – The hash of the transaction to retrieve.
Returns:: The transaction information as a Transaction object.

async eth_get_transaction_by_block(block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32, index: int) → Transaction

Get information about a specific transaction by its block and index.

Parameters:

block_identifier – The block identifier containing the transaction.
index – The index of the transaction within the block.

Returns:

The transaction information as a Transaction object.

async eth_get_transaction_cnt_by_block(block_identifier: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32) → int

Get the number of transactions in a specific block.

Parameters:: block_identifier – The block identifier.
Returns:: The number of transactions in the block.

async eth_get_transaction_receipt(transaction_hash: Hash32) → Receipt

Get the receipt of a specific transaction.

Parameters:

transaction_hash – The hash of the transaction to retrieve the
for. (receipt) –

Returns:

The transaction receipt as a Receipt object.

async eth_hashrate() → int

Get the hash rate of the current Ethereum node.

Returns:: The hash rate of the Ethereum node.

async eth_max_priority_fee_per_gas() → Wei

Get the latest maximum priority fee per gas unit on the current Ethereum node.

Returns:: The latest maximum priority fee per gas unit on the current Ethereum node.

async eth_mining() → bool

Check whether the current Ethereum node is mining.

Returns:: True if the current Ethereum node is mining, False otherwise.

async eth_send_raw_transaction(transaction: HexBytes) → Hash32

Send a raw transaction to the Ethereum network.

Parameters:: transaction – The raw transaction as a HexBytes object.
Returns:: The hash of the sent transaction as a Hash32 object.

async eth_sign(account: Address | ENS, data: HexBytes) → HexBytes

Sign data with an Ethereum account.

Parameters:

account – The address or ENS name of the account to sign with.
data – The data to sign as a HexBytes object.

Returns:

The signature as a HexBytes object.

Return type:

HexBytes

async eth_syncing() → SyncStatus | bool

Check whether the current Ethereum node is syncing the blockchain.

Returns:: If the current Ethereum node is syncing the blockchain, returns a SyncStatus object; otherwise, returns False if it has completed syncing.

async eth_uninstall_filter(filter: AsyncFilter) → bool

Uninstall a previously installed filter.

Parameters:: filter – The filter to uninstall.
Returns:: True if the filter was successfully uninstalled, False otherwise.

async eth_wait_for_transaction_receipt(transaction_hash: Hash32, timeout: float = 120, poll_latency: float = 0.1) → Receipt

Wait for a transaction to be included in a block and return the receipt.

Parameters:

transaction_hash – The hash of the transaction to wait for.
timeout – The maximum time to wait in seconds. Defaults to 120.
poll_latency – The time between polling attempts in seconds. Defaults to 0.1.

Returns:

The transaction receipt as a Receipt object.

async get_block_ts_by_number(height: BlockNumber) → int

Retrieves the timestamp of a block by its block number.

Parameters:: height – The block number.
Returns:: The timestamp of the block as an integer.
Raises:: GethGraphQLError – If the Geth node returns an error.

async get_blocks_by_numbers(numbers: list[eth_typing.evm.BlockNumber], step: int = 200) → list[ethhelper.datatypes.eth.Block]

Get the blocks with the given block numbers.

Parameters:

numbers – The block numbers.
step – The maximum number of blocks per request, preventing a single request from requiring too much memory and taking too long.

Returns:

A list of Block instances that represent the blocks with the given block numbers.

async get_blocks_by_numbers_range(start: BlockNumber, end: BlockNumber, step: int = 200) → list[ethhelper.datatypes.eth.Block]

Get the blocks with the given block numbers range.

Parameters:

start – The start block number.
end – The end block number.
step – The maximum number of blocks per request, preventing a single request from requiring too much memory and taking too long.

Returns:

A list of Block instances that represent the blocks with the given block numbers range.

async get_blocks_ts_by_numbers_range(from_height: BlockNumber, to_height: BlockNumber, step: int = 5000) → dict[eth_typing.evm.BlockNumber, int]

Retrieves the timestamps of blocks within a range of block numbers.

Parameters:

from_height – The starting block number.
to_height – The ending block number.
step – The maximum number of blocks per request.

Returns:

A dictionary mapping block numbers to their corresponding timestamps.

Raises:

GethGraphQLError – If the Geth node returns an error.

async get_height_after_ts(timestamp: int) → BlockNumber

Get the block height closest to the target timestamp.

Parameters:: timestamp – The target timestamp in seconds since the epoch.
Returns:: A BlockNumber instance that represents the block height closest to the target timestamp.

async get_logs(filter: FilterParams) → list[ethhelper.datatypes.eth.Log]

Retrieve a list of logs from the Geth node using the given FilterParams.

Parameters:: filter – A FilterParams object used to specify filter parameters for the logs.
Returns:: A list of Log objects parsed from the logs returned by the Geth node.
Raises:: ethhelper.types.GethError – Raised when the Geth node returns an error.

async get_logs_by_blocks(start_height: BlockNumber, end_height: BlockNumber, address: Address | list[ethhelper.datatypes.base.Address] | None = None, topics: Sequence[Hash32 | Sequence[Hash32]] | None = None, step: int = 200) → list[ethhelper.datatypes.eth.Log]

Retrieve a list of logs within a range of blocks specified by block heights.

Parameters:

start_height – The block height to start retrieving logs from.
end_height – The block height to stop retrieving logs from.
address – An address or list of addresses to filter the logs by.
topics – A list of topics or nested lists of topics to filter the logs by.
step – The maximum number of blocks per request, preventing a single request from requiring too much memory and taking too long.

Returns:

A list of Log objects parsed from the logs returned by the Geth node.

Raises:

ethhelper.types.GethError – Raised when the Geth node returns an error.

async is_connected() → bool

Checks the connectivity of the Geth node.

This function checks the connectivity of the Geth node by calling the is_connected() method of the Web3.py library. If the method returns a boolean value, it indicates that the Geth node is available and returns True. If not, it returns False. If an exception occurs during the request, the function logs a warning and returns False.

Raises:: NotImplementedError – If the is_connected() method of the Web3.py library returns a boolean value, it indicates that Web3.py is not running in asyncio.
Returns:: A bool value indicating whether the Geth node is connected and available or not.

async net_listening() → bool

Check if the Geth node is currently listening for connections.

Returns:: True if the node is listening, False otherwise.

async net_peer_count() → int

Get the number of peers currently connected to the Geth node.

Returns:: The number of peers.

async net_version() → str

Get the version number of the Geth node.

Returns:: The version as a string.
Return type:: str

parse_multiple_responses(raw_res: str) → tuple[list[ethhelper.datatypes.geth.GethSuccessResponse], list[ethhelper.datatypes.geth.GethErrorResponse]]

Parse the content of the Geth node responses.

Geth provides an interface that allows multiple Geth requests to be served in a single HTTP request. Likewise, the response will be multiple. This function is used to parse multiple Geth responses from a single HTTP response.

This function will attempt to parse the input into a list. Afterwards, this function assumes that the elements of this list are in the form of GethSuccessResponse` or ``GethErrorResponse, and use pydantic to parse these elements.

Parameters:

raw_res – the content of the HTTP response of the Geth node in string form.

Returns:

A tuple with two lists. One is a list of GethSuccessResponse and the other one is a list of GethErrorResponse. They respectively indicate whether these are normalized success responses or error responses.

Raises:

orjson.JSONDecodeError – Raised when the raw content cannot be json decoded.
pydantic.ValidationError – Raised when any element in json decoded list cannot be parsed into either GethSuccessResponse or GethErrorResponse.

parse_response(raw_res: str) → GethSuccessResponse | GethErrorResponse

Parse the content of the Geth node response.

This function will attempt to parse the input into a GethSuccessResponse, and failing that, into a GethErrorResponse.

Parameters:: raw_res – the content of the HTTP response of the Geth node in string form.
Returns:: An object of either GethSuccessResponse or GethErrorResponse, respectively indicates whether this is a normalized success response or an error response.
Raises:: pydantic.ValidationError – Raised when input data cannot be parsed into either GethSuccessResponse or GethErrorResponse.

async send(method: str, params: list[Any] | None = None) → Any

Send a Geth request to Geth node and return the data of Geth response.

HTTP communication with Geth follows the JSON-RPC specification. It stipulates that each request needs to provide a method and a series of parameters. This function is the encapsulation of this behavior. An id is automatically generated inside the function. The function will form a GethRequest object with the id and the input method and params, encode it into json text by pydantic and send it to the Geth node.

Parameters:

method – The method name of the Geth HTTP interface to call.
params – A series of parameters used by Geth to make the request.

Returns:

A basic type of object that represents the result returned by Geth after executing the request.

Raises:

httpx.RequestError – Raised when an HTTP request fails.
ethhelper.types.GethError – Raised when response is a Geth error.
ethhelper.types.IdNotMatch – Raised when received response id not match request id.

async send_multiple(raw_requests: list[tuple[str, list[Any] | None]]) → tuple[list[ethhelper.datatypes.geth.GethSuccessResponse], list[ethhelper.datatypes.geth.GethErrorResponse]]

Send multiple Geth requests to the Geth node and return the data of the Geth responses.

HTTP communication with Geth follows the JSON-RPC specification. It stipulates that each request needs to provide a method and a series of parameters. This function is the encapsulation of this behavior. Multiple requests can be sent in one HTTP request.

Parameters:

raw_requests – A list of tuples, where each tuple contains a string method and a list of parameters params. The parameters can be None if there are no parameters for the method.

Returns:

A tuple with two lists. One is a list of GethSuccessResponse and the other one is a list of GethErrorResponse. They respectively indicate whether these are normalized success responses or error responses.

Raises:

httpx.RequestError – Raised when an HTTP request fails.
orjson.JSONDecodeError – Raised when the raw content cannot be json decoded.
pydantic.ValidationError – Raised when any element in the json decoded list cannot be parsed into either GethSuccessResponse or GethErrorResponse.

async send_query(query: str) → dict[str, Any]

Sends a GraphQL query to the Geth node.

Parameters:: query – The GraphQL query string.
Returns:: A dictionary containing the result of the GraphQL query.
Raises:: GethGraphQLError – If the Geth node returns an error.

async send_raw(raw: str) → str

Send json text to Geth node and return text of the response.

Parameters:: raw – The json text will be sent.
Returns:: A string of the json content of the response in text.
Raises:: httpx.RequestError – Raised when an HTTP request fails.

async test_connection() → bool

Test connectivity to the Geth node via both Web3.py and the customized HTTP interface.

Returns:: A bool indicating whether the connection to the Geth node is successful or not.

async txpool_content() → TxpoolContent

Returns an object representing the contents of the Geth txpool.

This function sends a txpool_content request to the Geth node and returns the response as a TxpoolContent object.

Returns:

A TxpoolContent object representing the contents of the Geth txpool.

Raises:

ethhelper.types.GethError – Raised when response is a Geth error.
ethhelper.types.IdNotMatch – Raised when received response id not match request id.

async txpool_content_from(address: HexAddress) → TxpoolContentFrom

Returns an object representing the transactions from a specific address in the Geth txpool.

This function sends a txpool_contentFrom request to the Geth node with the specified address and returns the response as a TxpoolContentFrom object.

Parameters:

address – The address to get the transactions from.

Returns:

A TxpoolContentFrom object representing the transactions from the specified address in the Geth txpool.

Raises:

ethhelper.types.GethError – Raised when response is a Geth error.
ethhelper.types.IdNotMatch – Raised when received response id not match request id.

async txpool_inspect() → TxpoolInspect

Returns an object representing the contents of the Geth txpool.

This function sends a txpool_inspect request to the Geth node and returns the response as a TxpoolInspect object.

Returns:

A TxpoolInspect object representing the contents of the Geth txpool.

Raises:

ethhelper.types.GethError – Raised when response is a Geth error.
ethhelper.types.IdNotMatch – Raised when received response id not match request id.

async txpool_status() → TxpoolStatus

Returns an object representing the status of the Geth txpool.

This function sends a txpool_status request to the Geth node and returns the response as a TxpoolStatus object.

Returns:

A TxpoolStatus object representing the status of the Geth txpool.

Raises:

ethhelper.types.GethError – Raised when response is a Geth error.
ethhelper.types.IdNotMatch – Raised when received response id not match request id.

url: str

The url giving from the constructor of this class.

>>> from ethhelper import GethHttpConnector
>>> connector = GethHttpConnector("http://localhost:8545/")
>>> connector.url
'http://localhost:8545/'

logger: Logger: The logger giving from the constructor of this class or default by logging.getLogger("GethHttpConnector")

eth: AsyncEth: Asynchronous Eth interface for Web3. Used to simplify the access path.

net: AsyncNet: Asynchronous Net interface for Web3. Used to simplify the access path.

id: int: The id used when sending requests to Geth, an integer starting from 1 and incrementing for each request. Reset when it exceeds 100000000.

GethNewBlockSubscriber 

class ethhelper.GethNewBlockSubscriber(url: str, logger: Logger | None = None)

async subscribe_new_block() → None: Subscribe to new block notifications on the Geth node.

async after_connection() → None

A method that is called after the connection to the Geth node has been established.

This method is overridden to wait for the Geth node to finish syncing and then subscribe to new block notifications.

async handle(data: GethWSResponse | GethSuccessResponse) → None

A method that is called when a message is received from the Geth node.

This method is overridden to handle new block notifications from the Geth node.

Parameters:: data – The message received from the Geth node.

abstract async on_block(block: Block) → None

A method that is called when a new block notification is received from the Geth node.

This method must be overridden to handle new block notifications from the Geth node.

Parameters:: block – The new block that was received from the Geth node.

async on_other(data: GethSuccessResponse) → None

A method that is called when a non-new-block message is received from the Geth node.

This method can be overridden to handle non-new-block messages from the Geth node.

Parameters:: data – The message received from the Geth node.

async bind() → Task[None]

Bind the subscriber to the Geth node and start listening for messages.

Returns:: A task that will run the subscriber until it is closed.

async close() → None: Close the connection to the Geth node.

async run() → None: The main loop that listens for messages from the Geth node.

async send(method: str, params: list[Any]) → int

Send a request to the Geth node over the websocket connection.

Parameters:

method – The JSON-RPC method to call.
params – The parameters to send with the request.

Returns:

The ID of the request, which can be used to match responses with requests.

async subscribe(param: str) → int

Subscribe to a specific event or method on the Geth node.

Parameters:: param – The name of the event or method to subscribe to.
Returns:: The ID of the subscription, which can be used to unsubscribe later.

Low-level API

ETHHelper API

High-level API

GethHttpConnector

GethNewBlockSubscriber

High-level API 

GethHttpConnector 

GethNewBlockSubscriber 