Skip to content

plugins.aea-ledger-ethereum.aea_ledger_ethereum.ethereum

Ethereum module wrapping the public and private key cryptography and ledger api.

SPEED_FAST

safeLow, standard, fast

PRIORITY_FEE_INCREASE_BOUNDARY

percentage

GAS_STATION_FALLBACK_ESTIMATE

gwei

to_eth_unit

def to_eth_unit(number: Union[int, float, str, decimal.Decimal],
                unit_in: str = DEFAULT_CURRENCY_DENOM,
                unit_out: str = GWEI) -> Union[int, Wei, decimal.Decimal]

Covert a number to the given unit.

get_base_fee_multiplier

def get_base_fee_multiplier(
        base_fee_gwei: Union[int, decimal.Decimal]) -> float

Returns multiplier value.

get_default_gas_strategy

def get_default_gas_strategy(chain_id: int) -> Dict[str, Any]

Get default gas strategy for the given chain ID.

estimate_priority_fee

def estimate_priority_fee(web3_object: Web3, block_number: int,
                          default_priority_fee: Optional[int],
                          fee_history_blocks: int, fee_history_percentile: int,
                          min_allowed_tip: int,
                          priority_fee_increase_boundary: int) -> int

Estimate priority fee from base fee.

get_gas_price_strategy_eip1559

def get_gas_price_strategy_eip1559(
    max_gas_fast: int, fee_history_blocks: int, fee_history_percentile: int,
    default_priority_fee: Optional[int], fallback_estimate: Dict[str, Wei],
    min_allowed_tip: int, priority_fee_increase_boundary: int
) -> Callable[[Web3, TxParams], Dict[str, Wei]]

Get the gas price strategy.

get_gas_price_strategy_eip1559_polygon

def get_gas_price_strategy_eip1559_polygon(
        gas_endpoint: str,
        fallback_estimate: Dict[str, Wei],
        speed: Optional[str] = SPEED_FAST
) -> Callable[[Any, Any], Dict[str, Wei]]

Get the gas price strategy.

rpc_gas_price_strategy_wrapper

def rpc_gas_price_strategy_wrapper(
        web3: Web3, transaction_params: TxParams) -> Dict[str, Wei]

RPC gas price strategy wrapper.

get_gas_price_strategy

def get_gas_price_strategy(
    gas_price_strategy: Optional[str] = None,
    gas_price_api_key: Optional[str] = None
) -> Callable[[Web3, TxParams], Dict[str, Wei]]

Get the gas price strategy.

rpc_call_with_timeout

def rpc_call_with_timeout(func: Callable, timeout: int) -> Any

Execute an RPC call with a timeout.

match_items

def match_items(filter_: EventFilterBuilder, event_name: str,
                method_to_matches: Dict[str, Dict[str, Any]])

Build filters for the given match dictionary.

SignedTransactionTranslator Objects

class SignedTransactionTranslator()

Translator for SignedTransaction.

to_dict

@staticmethod
def to_dict(
        signed_transaction: SignedTransaction) -> Dict[str, Union[str, int]]

Write SignedTransaction to dict.

from_dict

@staticmethod
def from_dict(signed_transaction_dict: JSONLike) -> SignedTransaction

Get SignedTransaction from dict.

AttributeDictTranslator Objects

class AttributeDictTranslator()

Translator for AttributeDict.

to_dict

@classmethod
def to_dict(cls, attr_dict: Union[AttributeDict, TxReceipt,
                                  TxData]) -> JSONLike

Simplify to dict.

from_dict

@classmethod
def from_dict(cls, di: JSONLike) -> AttributeDict

Get back attribute dict.

EthereumCrypto Objects

class EthereumCrypto(Crypto[LocalAccount])

Class wrapping the Account Generation from Ethereum ledger.

__init__

def __init__(private_key_path: Optional[str] = None,
             password: Optional[str] = None,
             extra_entropy: Union[str, bytes, int] = "") -> None

Instantiate an ethereum crypto object.

Arguments:

  • private_key_path: the private key path of the agent
  • password: the password to encrypt/decrypt the private key.
  • extra_entropy: add extra randomness to whatever randomness your OS can provide

private_key

@property
def private_key() -> str

Return a private key.

64 random hex characters (i.e. 32 bytes) + "0x" prefix.

Returns:

a private key string in hex format

public_key

@property
def public_key() -> str

Return a public key in hex format.

128 hex characters (i.e. 64 bytes) + "0x" prefix.

Returns:

a public key string in hex format

address

@property
def address() -> str

Return the address for the key pair.

40 hex characters (i.e. 20 bytes) + "0x" prefix.

Returns:

an address string in hex format

load_private_key_from_path

@classmethod
def load_private_key_from_path(cls,
                               file_name: str,
                               password: Optional[str] = None) -> LocalAccount

Load a private key in hex format from a file.

Arguments:

  • file_name: the path to the hex file.
  • password: the password to encrypt/decrypt the private key.

Returns:

the Entity.

sign_message

def sign_message(message: bytes, is_deprecated_mode: bool = False) -> str

Sign a message in bytes string form.

Arguments:

  • message: the message to be signed
  • is_deprecated_mode: if the deprecated signing is used

Returns:

signature of the message in string form

sign_transaction

def sign_transaction(transaction: JSONLike) -> JSONLike

Sign a transaction in bytes string form.

Arguments:

  • transaction: the transaction to be signed

Returns:

signed transaction

generate_private_key

@classmethod
def generate_private_key(cls,
                         extra_entropy: Union[str, bytes,
                                              int] = "") -> LocalAccount

Generate a key pair for ethereum network.

Arguments:

  • extra_entropy: add extra randomness to whatever randomness your OS can provide

Returns:

account object

encrypt

def encrypt(password: str) -> str

Encrypt the private key and return in json.

Arguments:

  • password: the password to decrypt.

Returns:

json string containing encrypted private key.

decrypt

@classmethod
def decrypt(cls, keyfile_json: str, password: str) -> str

Decrypt the private key and return in raw form.

Arguments:

  • keyfile_json: json str containing encrypted private key.
  • password: the password to decrypt.

Returns:

the raw private key (without leading "0x").

EthereumHelper Objects

class EthereumHelper(Helper)

Helper class usable as Mixin for EthereumApi or as standalone class.

is_transaction_settled

@staticmethod
def is_transaction_settled(tx_receipt: JSONLike) -> bool

Check whether a transaction is settled or not.

Arguments:

  • tx_receipt: the receipt associated to the transaction.

Returns:

True if the transaction has been settled, False o/w.

get_contract_address

@staticmethod
def get_contract_address(tx_receipt: JSONLike) -> Optional[str]

Retrieve the contract_address from a transaction receipt.

Arguments:

  • tx_receipt: the receipt of the transaction.

Returns:

the contract address, if present

is_transaction_valid

@staticmethod
def is_transaction_valid(tx: dict, seller: Address, client: Address,
                         tx_nonce: str, amount: int) -> bool

Check whether a transaction is valid or not.

Arguments:

  • tx: the transaction.
  • seller: the address of the seller.
  • client: the address of the client.
  • tx_nonce: the transaction nonce.
  • amount: the amount we expect to get from the transaction.

Returns:

True if the random_message is equals to tx['input']

generate_tx_nonce

@staticmethod
def generate_tx_nonce(seller: Address, client: Address) -> str

Generate a unique hash to distinguish transactions with the same terms.

Arguments:

  • seller: the address of the seller.
  • client: the address of the client.

Returns:

return the hash in hex.

get_address_from_public_key

@classmethod
def get_address_from_public_key(cls, public_key: str) -> str

Get the address from the public key.

Arguments:

  • public_key: the public key

Returns:

str

recover_message

@classmethod
def recover_message(cls,
                    message: bytes,
                    signature: str,
                    is_deprecated_mode: bool = False) -> Tuple[Address, ...]

Recover the addresses from the hash.

Arguments:

  • message: the message we expect
  • signature: the transaction signature
  • is_deprecated_mode: if the deprecated signing was used

Returns:

the recovered addresses

recover_public_keys_from_message

@classmethod
def recover_public_keys_from_message(
        cls,
        message: bytes,
        signature: str,
        is_deprecated_mode: bool = False) -> Tuple[str, ...]

Get the public key used to produce the signature of the message

Arguments:

  • message: raw bytes used to produce signature
  • signature: signature of the message
  • is_deprecated_mode: if the deprecated signing was used

Returns:

the recovered public keys

get_hash

@staticmethod
def get_hash(message: bytes) -> str

Get the hash of a message.

Arguments:

  • message: the message to be hashed.

Returns:

the hash of the message as a hex string.

load_contract_interface

@classmethod
def load_contract_interface(cls, file_path: Path) -> Dict[str, str]

Load contract interface.

Arguments:

  • file_path: the file path to the interface

Returns:

the interface

EthereumApi Objects

class EthereumApi(LedgerApi, EthereumHelper)

Class to interact with the Ethereum Web3 APIs.

__init__

def __init__(**kwargs: Any)

Initialize the Ethereum ledger APIs.

Arguments:

  • kwargs: keyword arguments

api

@property
def api() -> Web3

Get the underlying API object.

get_balance

def get_balance(address: Address, raise_on_try: bool = False) -> Optional[int]

Get the balance of a given account.

get_state

def get_state(callable_name: str,
              *args: Any,
              raise_on_try: bool = False,
              **kwargs: Any) -> Optional[JSONLike]

Call a specified function on the ledger API.

get_transfer_transaction

def get_transfer_transaction(
        sender_address: Address,
        destination_address: Address,
        amount: int,
        tx_fee: int,
        tx_nonce: str,
        chain_id: Optional[int] = None,
        max_fee_per_gas: Optional[int] = None,
        max_priority_fee_per_gas: Optional[str] = None,
        gas_price: Optional[str] = None,
        gas_price_strategy: Optional[str] = None,
        gas_price_strategy_extra_config: Optional[Dict] = None,
        raise_on_try: bool = False,
        **kwargs: Any) -> Optional[JSONLike]

Submit a transfer transaction to the ledger.

Arguments:

  • sender_address: the sender address of the payer.
  • destination_address: the destination address of the payee.
  • amount: the amount of wealth to be transferred (in Wei).
  • tx_fee: the transaction fee (gas) to be used (in Wei).
  • tx_nonce: verifies the authenticity of the tx.
  • chain_id: the Chain ID of the Ethereum transaction.
  • max_fee_per_gas: maximum amount you're willing to pay, inclusive of baseFeePerGas and maxPriorityFeePerGas. The difference between maxFeePerGas and baseFeePerGas + maxPriorityFeePerGas is refunded (in Wei).
  • max_priority_fee_per_gas: the part of the fee that goes to the miner (in Wei).
  • gas_price: the gas price (in Wei)
  • gas_price_strategy: the gas price strategy to be used.
  • gas_price_strategy_extra_config: extra config for gas price strategy.
  • raise_on_try: whether the method will raise or log on error
  • kwargs: keyword arguments

Returns:

the transfer transaction

try_get_gas_pricing

@try_decorator("Unable to retrieve gas price: {}", logger_method="warning")
def try_get_gas_pricing(gas_price_strategy: Optional[str] = None,
                        extra_config: Optional[Dict] = None,
                        old_price: Optional[Dict[str, Wei]] = None,
                        **_kwargs: Any) -> Optional[Dict[str, Wei]]

Try get the gas price based on the provided strategy.

Arguments:

  • gas_price_strategy: the gas price strategy to use, e.g., the EIP-1559 strategy. Can be either eip1559 or gas_station.
  • extra_config: gas price strategy getter parameters.
  • old_price: the old gas price params in case that we are trying to resubmit a transaction.
  • _kwargs: the keyword arguments. Possible kwargs are: raise_on_try: bool flag specifying whether the method will raise or log on error (used by try_decorator)

Returns:

a dictionary with the gas data.

update_with_gas_estimate

def update_with_gas_estimate(transaction: JSONLike,
                             raise_on_try: bool = False) -> JSONLike

Attempts to update the transaction with a gas estimate

Arguments:

  • transaction: the transaction
  • raise_on_try: whether the method will raise or log on error

Returns:

the updated transaction

get_l1_data_fee

def get_l1_data_fee(transaction: JSONLike) -> int

Get the L1 data fee for the transaction on OP stack chains.

Docs: https://docs.optimism.io/builders/app-developers/transactions/estimates#l1-data-fee

Arguments:

  • transaction: the transaction

Returns:

the data fee in wei

send_signed_transaction

def send_signed_transaction(tx_signed: JSONLike,
                            raise_on_try: bool = False) -> Optional[str]

Send a signed transaction and wait for confirmation.

Arguments:

  • tx_signed: the signed transaction
  • raise_on_try: whether the method will raise or log on error

Returns:

tx_digest, if present

get_transaction_receipt

def get_transaction_receipt(tx_digest: str,
                            raise_on_try: bool = False) -> Optional[JSONLike]

Get the transaction receipt for a transaction digest.

Arguments:

  • tx_digest: the digest associated to the transaction.
  • raise_on_try: whether the method will raise or log on error

Returns:

the tx receipt, if present

get_transaction

def get_transaction(tx_digest: str,
                    raise_on_try: bool = False) -> Optional[JSONLike]

Get the transaction for a transaction digest.

Arguments:

  • tx_digest: the digest associated to the transaction.
  • raise_on_try: whether the method will raise or log on error

Returns:

the tx, if present

get_contract_instance

def get_contract_instance(contract_interface: Dict[str, str],
                          contract_address: Optional[str] = None) -> Any

Get the instance of a contract.

Arguments:

  • contract_interface: the contract interface.
  • contract_address: the contract address.

Returns:

the contract instance

get_deploy_transaction

def get_deploy_transaction(contract_interface: Dict[str, str],
                           deployer_address: Address,
                           raise_on_try: bool = False,
                           **kwargs: Any) -> Optional[JSONLike]

Get the transaction to deploy the smart contract.

Arguments:

  • contract_interface: the contract interface.
  • deployer_address: The address that will deploy the contract.
  • raise_on_try: whether the method will raise or log on error
  • kwargs: keyword arguments

Returns:

the transaction dictionary.

is_valid_address

@classmethod
def is_valid_address(cls, address: Address) -> bool

Check if the address is valid.

Arguments:

  • address: the address to validate

Returns:

whether the address is valid

contract_method_call

@classmethod
def contract_method_call(cls, contract_instance: Any, method_name: str,
                         **method_args: Any) -> Optional[JSONLike]

Call a contract's method

Arguments:

  • contract_instance: the contract to use
  • method_name: the contract method to call
  • method_args: the contract call parameters

Returns:

the call result

build_transaction

def build_transaction(contract_instance: Any,
                      method_name: str,
                      method_args: Optional[Dict[Any, Any]],
                      tx_args: Optional[Dict[Any, Any]],
                      raise_on_try: bool = False) -> Optional[JSONLike]

Prepare a transaction

Arguments:

  • contract_instance: the contract to use
  • method_name: the contract method to call
  • method_args: the contract parameters
  • tx_args: the transaction parameters
  • raise_on_try: whether the method will raise or log on error

Returns:

the transaction

get_transaction_transfer_logs

def get_transaction_transfer_logs(
        contract_instance: Any,
        tx_hash: str,
        target_address: Optional[str] = None) -> Optional[JSONLike]

Get all transfer events derived from a transaction.

Arguments:

  • contract_instance: the contract
  • tx_hash: the transaction hash
  • target_address: optional address to filter transfer events to just those that affect it

Returns:

the transfer logs

send_signed_transactions

def send_signed_transactions(signed_transactions: List[JSONLike],
                             raise_on_try: bool = False,
                             **kwargs: Any) -> Optional[List[str]]

Atomically send multiple of transactions.

This operation is not supported for ethereum. Please use the ethereum_flashbots instead.

Arguments:

  • signed_transactions: the signed transactions to bundle together and send.
  • raise_on_try: whether the method will raise or log on error
  • kwargs: the keyword arguments.

batch_filter_wrapper

def batch_filter_wrapper(event: ContractEvent, match_single: Dict[str, Any],
                         match_any: Dict[str, Any], to_block: BlockNumber,
                         from_block: BlockNumber) -> Any

A wrapper for a single batch's event filtering operation.

filter_event

def filter_event(event: ContractEvent,
                 match_single: Dict[str, Any],
                 match_any: Dict[str, Any],
                 to_block: BlockNumber,
                 from_block: BlockNumber = 0,
                 batch_size: int = 5_000,
                 max_retries: int = 5,
                 reduce_factor: float = 0.25,
                 timeout: int = 5 * 60) -> Optional[JSONLike]

Filter an event using batching to avoid RPC timeouts.

Arguments:

  • event: the event to filter for.
  • match_single: the filter parameters with value checking against the event abi. It allows for defining a single match value.
  • match_any: the filter parameters with value checking against the event abi. It allows for defining multiple match values.
  • to_block: the block to which to filter.
  • from_block: the block from which to start filtering.
  • batch_size: the blocks' batch size of the filtering.
  • max_retries: the maximum number of retries.
  • reduce_factor: the percentage by which the batch size is reduced in case of a timeout.
  • timeout: a timeout in seconds to interrupt the operation in case the RPC request hangs.

Returns:

the filtering result.

EthereumFaucetApi Objects

class EthereumFaucetApi(FaucetApi)

Ethereum testnet faucet API.

get_wealth

def get_wealth(address: Address, url: Optional[str] = None) -> None

Get wealth from the faucet for the provided address.

Arguments:

  • address: the address.
  • url: the url

SimpleCacheLockWrapper Objects

class SimpleCacheLockWrapper()

Wrapper for session_cache with threading.Lock.

__init__

def __init__(session_cache: SimpleCache) -> None

Init wrapper.

__contains__

def __contains__(*args: Any, **kwargs: Any) -> Any

Contain item.

__len__

def __len__() -> int

Length of the cache

cache

def cache(key: str, value: Any) -> Dict[str, Any]

session_cache Cache.

get_cache_entry

def get_cache_entry(key: str) -> Any

Get cache entry.

clear

def clear() -> None

Clear cache entries.

items

def items() -> Dict[str, Any]

Return session items.