ETHHelper HTTP Connectors

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).

HTTP Low-level API

GethHttpAbstract

class ethhelper.connectors.http.GethHttpAbstract(url: str, logger: Logger)

A basic abstraction over Geth’s HTTP interface wrapper.

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.

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")

abstract async is_connected() → bool

Check connectivity.

Subclasses must implement methods to detect connectivity.

Returns:

A bool value for connectivity is good or not.

GethHttpCustomized

class ethhelper.connectors.http.GethHttpCustomized(url: str, logger: Logger)

An interface wrapper for direct HTTP access to Geth.

The GethHttpCustomized class provides an interface wrapper that allows for direct HTTP access to a Geth node.

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.

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.

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.

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.

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 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 is_connected() → bool

Checks the connectivity of the Geth node.

This function checks the connectivity of the Geth node by sending a net_version request to the node. If the request is successful, 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.

Returns:

A bool value indicating whether the Geth node is connected and available or not.

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")

GethHttpWeb3

class ethhelper.connectors.http.GethHttpWeb3(url: str, logger: Logger)

A wrapper based on Web3.py to access Geth HTTP interface.

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.

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")

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.

GethEthHttp

class ethhelper.connectors.http.GethEthHttp(url: str, logger: Logger)

Class for interacting with an Ethereum node using HTTP with the Geth JSON-RPC API.

This class extends the GethHttpWeb3 class and provides additional methods specifically for interacting with the Ethereum network using the Geth JSON-RPC API.

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.

eth: AsyncEth

Asynchronous Eth interface for Web3. Used to simplify the access path.

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_hashrate() → int

Get the hash rate of the current Ethereum node.

Returns:

The hash rate of the Ethereum 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_chain_id() → int

Get the chain ID of the current Ethereum node.

Returns:

The chain ID of the current Ethereum node.

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_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_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_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_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_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_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_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_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_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_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_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_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_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_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_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_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_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 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_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_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_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 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.

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")

GethNetHttp

class ethhelper.connectors.http.GethNetHttp(url: str, logger: Logger)

Class for interacting with a Geth node using HTTP with the Geth JSON-RPC API.

This class extends the GethHttpWeb3 class and provides additional methods specifically for interacting with the network using the Geth JSON-RPC API.

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.

net: AsyncNet

Asynchronous Net interface for Web3. Used to simplify the access path.

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 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_version() → str

Get the version number of the Geth node.

Returns:

The version as a string.

Return type:

str

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")

GethTxpoolHttp

class ethhelper.connectors.http.GethTxpoolHttp(url: str, logger: Logger)

A class that provides an interface to interact with the Geth txpool using HTTP.

The GethTxpoolHttp class inherits from the GethHttpCustomized class and provides methods to interact with the Geth txpool using HTTP.

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.

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.

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_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 is_connected() → bool

Checks the connectivity of the Geth node.

This function checks the connectivity of the Geth node by sending a net_version request to the node. If the request is successful, 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.

Returns:

A bool value indicating whether the Geth node is connected and available or not.

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_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.

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.

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")

GethCustomHttp

class ethhelper.connectors.http.GethCustomHttp(url: str, logger: Logger)

A customized HTTP interface for Geth nodes that inherits from GethEthHttp, GethNetHttp, and GethTxpoolHttp. Provides additional functionalities to access Geth nodes via HTTP.

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.

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 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 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_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 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 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_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 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.

eth: AsyncEth

Asynchronous Eth interface for Web3. Used to simplify the access path.

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")

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.

GethGraphQL

class ethhelper.connectors.http.GethGraphQL(url: str, logger: Logger, graphql_url: str | None = None)

A GraphQL interface for Geth nodes that inherits from GethHttpAbstract. Provides additional functionalities to access Geth nodes via HTTP and GraphQL.

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.

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 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 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_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.

abstract async is_connected() → bool

Check connectivity.

Subclasses must implement methods to detect connectivity.

Returns:

A bool value for connectivity is good or not.

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")