pythereum package

Submodules

pythereum.builders module

class pythereum.builders.BeaverBuilder(private_key: str | pythereum.common.HexStr | None = None)[source]

Bases: Builder

format_private_transaction(tx: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], max_block_number: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr] | None = None) → list[Any][source]

class pythereum.builders.Builder(url: str, private_transaction_method: str = 'eth_sendPrivateTransaction', bundle_method: str = 'eth_sendBundle', cancel_bundle_method: str = 'eth_cancelBundle', mev_bundle_method: str = 'mev_sendBundle', bundle_params: set = None, private_key: str = None)[source]

Bases: ABC

format_bundle(bundle: dict | pythereum.dclasses.Bundle) → dict[source]

static format_mev_bundle(bundle: MEVBundle) → list[dict][source]

format_private_transaction(tx: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], max_block_number: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr] | None = None) → list[Any][source]

get_flashbots_header(payload: str = '') → dict[source]

get_header(data: Any = None) → Any[source]

class pythereum.builders.Builder0x69(private_key: str | pythereum.common.HexStr | None = None)[source]

Bases: Builder

format_private_transaction(tx: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], max_block_number: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr] | None = None) → list[Any][source]

Bases: object

An RPC class designed for sending raw transactions and bundles to specific block builders

async cancel_bundle(replacement_uuids: str | pythereum.common.HexStr)[source]

async close_session()[source]

async send_bundle(bundle: Bundle) → Any[source]

async send_mev_bundle(bundle: MEVBundle) → Any[source]

async send_private_transaction(tx: str | pythereum.common.HexStr, extra_info: Any = None) → Any[source]

async start_session()[source]

class pythereum.builders.FlashbotsBuilder(private_key: str | pythereum.common.HexStr | None = None)[source]

Bases: Builder

format_private_transaction(tx: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], preferences: dict = None) → list[Any][source]

get_header(payload: str = '') → dict[source]

class pythereum.builders.RsyncBuilder(private_key: str | pythereum.common.HexStr | None = None)[source]

Bases: Builder

format_private_transaction(tx: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], max_block_number: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr] | None = None) → list[Any][source]

class pythereum.builders.TitanBuilder(private_key: str | pythereum.common.HexStr | None = None)[source]

Bases: Builder

format_private_transaction(tx: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], max_block_number: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr] | None = None) → list[dict][source]

pythereum.common module

class pythereum.common.BlockTag(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

Data type encapsulating all possible non-integer values for a DefaultBlock parameter API Documentation at: https://ethereum.org/en/developers/docs/apis/json-rpc/#default-block

earliest = 'earliest'

finalized = 'finalized'

latest = 'latest'

pending = 'pending'

safe = 'safe'

class pythereum.common.EthDenomination(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: float, Enum

An enumeration of all names of eth denominations and their corresponding wei values

babbage = 1000.0

eth = 1e+18

ether = 1e+18

femtoether = 1000.0

finney = 1000000000000000.0

gether = 1e+27

grand = 1e+21

gwei = 1000000000.0

kether = 1e+21

kwei = 1000.0

lovelace = 1000000.0

mether = 1e+24

micro = 1000000000000.0

microether = 1000000000000.0

milli = 1000000000000000.0

milliether = 1000000000000000.0

mwei = 1000000.0

nano = 1000000000.0

nanoether = 1000000000.0

picoether = 1000000.0

shannon = 1000000000.0

szabo = 1000000000000.0

tether = 1e+30

wei = 1.0

class pythereum.common.GasStrategy(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

custom = 7

lower_quartile_price = 6

max_price = 1

mean_price = 3

median_price = 2

min_price = 0

mode_price = 4

upper_quartile_price = 5

class pythereum.common.HexStr(value: str | int | bytes)[source]

Bases: str

Data type representing a base16 hexadecimal number. Extends the functionality of the default string type to support hex operations and functionality.

Attributes: - integer_value: returns integer representing the base10 form of the hexadecimal number. - raw_hex: returns string representing the hexadecimal number without 0x prefix. - hex_bytes: returns bytes object representing the conversion of the value.

HEX_PATTERN = re.compile('^0x[0-9a-fA-F]+$')

property hex_bytes

property integer_value

property raw_hex

class pythereum.common.SubscriptionType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

logs = 'logs'

new_heads = 'newHeads'

new_pending_transactions = 'newPendingTransactions'

syncing = 'syncing'

pythereum.dclasses module

class pythereum.dclasses.Access(address: pythereum.common.HexStr | None, storage_keys: list[pythereum.common.HexStr] | None)[source]

Bases: object

Information on access lists available at https://eips.ethereum.org/EIPS/eip-2930

address: pythereum.common.HexStr | None

dataclass_json_config = {'letter_case': <function camelcase>}

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

storage_keys: list[pythereum.common.HexStr] | None

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

Bases: object

base_fee_per_gas: int | None

dataclass_json_config = {'letter_case': <function camelcase>}

difficulty: int | None

extra_data: pythereum.common.HexStr | None

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

gas_limit: int | None

gas_used: int | None

hash: pythereum.common.HexStr | None

logs_bloom: pythereum.common.HexStr | None

miner: pythereum.common.HexStr | None

mix_hash: pythereum.common.HexStr | None

nonce: int | None

number: int | None

parent_hash: pythereum.common.HexStr | None

receipts_root: pythereum.common.HexStr | None

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

sha3_uncles: pythereum.common.HexStr | None

size: int | None

state_root: pythereum.common.HexStr | None

timestamp: pythereum.common.HexStr | None

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

total_difficulty: int | None

transactions: list[pythereum.dclasses.TransactionFull | pythereum.common.HexStr] | None

transactions_root: pythereum.common.HexStr | None

uncles: list[pythereum.common.HexStr] | None

class pythereum.dclasses.Bundle(txs: list[str] | list[pythereum.common.HexStr], block_number: str | pythereum.common.HexStr | None = None, min_timestamp: int | pythereum.common.HexStr | str | None = None, max_timestamp: int | pythereum.common.HexStr | str | None = None, reverting_tx_hashes: list[str] | list[pythereum.common.HexStr] | None = None, uuid: str | pythereum.common.HexStr | None = None, replacement_uuid: str | pythereum.common.HexStr | None = None, refund_percent: int | pythereum.common.HexStr | str | None = None, refund_index: int | pythereum.common.HexStr | str | None = None, refund_recipient: str | pythereum.common.HexStr | None = None, refund_tx_hashes: list[str] | list[pythereum.common.HexStr] | None = None)[source]: Bases: dict

Bases: object

address: pythereum.common.HexStr | None

block_hash: pythereum.common.HexStr | None

block_number: int | None

data: pythereum.common.HexStr | None

dataclass_json_config = {'letter_case': <function camelcase>}

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

log_index: int | None

removed: bool

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

topics: list[pythereum.common.HexStr] | None

transaction_hash: pythereum.common.HexStr | None

transaction_index: int | None

class pythereum.dclasses.MEVBundle(version: str = 'v0.1', block: pythereum.common.HexStr | int | str = 0, max_block: pythereum.common.HexStr | int | str | None = None, flashbots_hashes: list[pythereum.common.HexStr] | list[str] | None = None, transactions: list[pythereum.common.HexStr] | list[str] | None = None, transactions_can_revert: bool | list[bool] = False, extra_mev_bundles: list[dict] | None = None, refund_addresses: list[str] | None = None, refund_percentages: list[int] | None = None)[source]: Bases: dict

Bases: object

block_hash: pythereum.common.HexStr | None

block_number: int | None

contract_address: pythereum.common.HexStr | None

cumulative_gas_used: int | None

dataclass_json_config = {'letter_case': <function camelcase>}

effective_gas_price: int | None

from_address: pythereum.common.HexStr | None

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

gas_used: int | None

logs: list[pythereum.dclasses.Log] | None

logs_bloom: pythereum.common.HexStr | None

root: pythereum.common.HexStr | None

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

status: int | None

to_address: pythereum.common.HexStr | None

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

transaction_hash: pythereum.common.HexStr | None

transaction_index: int | None

type: int | None

class pythereum.dclasses.Sync(starting_block: pythereum.common.HexStr | None, current_block: pythereum.common.HexStr | None, highest_block: pythereum.common.HexStr | None)[source]

Bases: object

Class representing ethereum sync status

current_block: pythereum.common.HexStr | None

dataclass_json_config = {'letter_case': <function camelcase>}

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

highest_block: pythereum.common.HexStr | None

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

starting_block: pythereum.common.HexStr | None

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

Bases: dict

from_address: Address from which fees should be sent. Transaction must be signed by private key from this account.

to_address: Address to which fees are sent

max_priority_fee_per_gas: The maximum price of the consumed gas to be included as a tip to the validator

max_fee_per_gas: The maximum fee per unit of gas willing to be paid for the transaction (inclusive of baseFeePerGas and maxPriorityFeePerGas)

gas: Maximum gas allocated for execution of the transaction onchain (evm specifies the correct amount to use)

value: Value in wei sent from the from_address to to_address

data: Optional extra data

nonce: A sequentially incrementing counter uniquely identifying each transaction from each account

Bases: object

access_list: list[pythereum.dclasses.Access] | None

block_hash: pythereum.common.HexStr | None

block_number: int | None

chain_id: int | None

dataclass_json_config = {'letter_case': <function camelcase>}

from_address: pythereum.common.HexStr | None

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

gas: int | None

gas_price: int | None

hash: pythereum.common.HexStr | None

input: pythereum.common.HexStr | None

max_fee_per_gas: int | None

max_priority_fee_per_gas: int | None

nonce: int | None

r: pythereum.common.HexStr | None

s: pythereum.common.HexStr | None

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

to_address: pythereum.common.HexStr | None

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

transaction_index: int | None

type: int | None

v: int | None

value: int | None

pythereum.dclasses.access_decoder(access_dict: dict | None) → pythereum.dclasses.Access | None[source]

pythereum.dclasses.access_encoder(access_obj: pythereum.dclasses.Access | None) → dict | None[source]

pythereum.dclasses.access_list_decoder(access_list: list[dict] | None) → list[pythereum.dclasses.Access] | None[source]

pythereum.dclasses.access_list_encoder(access_obj_list: list[pythereum.dclasses.Access] | None) → list[dict] | None[source]

pythereum.dclasses.hex_decoder(hex_string: str | None) → pythereum.common.HexStr | None[source]

pythereum.dclasses.hex_encoder(hex_obj: pythereum.common.HexStr | None) → str | None[source]: Takes in a hex object and returns its hex string representation

pythereum.dclasses.hex_int_decoder(hex_string: str | None) → int | None[source]

pythereum.dclasses.hex_int_encoder(int_val: int | None) → str | None[source]

pythereum.dclasses.hex_list_decoder(hex_string_list: list[str] | None) → list[pythereum.common.HexStr] | None[source]

pythereum.dclasses.hex_list_encoder(hex_obj_list: list[pythereum.common.HexStr]) → list[str] | None[source]

pythereum.dclasses.log_decoder(log_dict: dict | None) → pythereum.dclasses.Log | None[source]

pythereum.dclasses.log_encoder(log_obj: pythereum.dclasses.Log | None) → dict | None[source]

pythereum.dclasses.log_list_decoder(log_list: list[dict] | None) → list[pythereum.dclasses.Log] | None[source]

pythereum.dclasses.log_list_encoder(log_obj_list: list[pythereum.dclasses.Log] | None) → list[dict] | None[source]

pythereum.dclasses.transaction_decoder(transaction_hex: dict | str) → pythereum.dclasses.TransactionFull | pythereum.common.HexStr[source]

pythereum.dclasses.transaction_encoder(transaction_obj: pythereum.common.HexStr | pythereum.dclasses.TransactionFull) → str | dict[source]

pythereum.dclasses.transaction_list_decoder(tr_list: list[dict | str] | None) → list[pythereum.dclasses.TransactionFull | pythereum.common.HexStr] | None[source]

pythereum.dclasses.transaction_list_encoder(tr_list: list[pythereum.dclasses.TransactionFull | pythereum.common.HexStr] | None) → list[dict | str] | None[source]

pythereum.exceptions module

exception pythereum.exceptions.ERPCBaseException(message: str)[source]

Bases: Exception

Base exception class for Ethereum RPC interactions.

exception pythereum.exceptions.ERPCBuilderException(message: str)[source]

Bases: ERPCBaseException

Raised for exceptions related to builders and the BuilderRPC

exception pythereum.exceptions.ERPCDecoderException(message: str)[source]

Bases: ERPCBaseException

Raised when invalid data is input to a decoder and an error is thrown.

exception pythereum.exceptions.ERPCEncoderException(message: str)[source]

Bases: ERPCBaseException

Raised when invalid data is input to an encoder and an error is thrown.

exception pythereum.exceptions.ERPCInvalidReturnException(message: str)[source]

Bases: ERPCBaseException

Raised when the Ethereum RPC returns a value which is incorrectly formatted.

exception pythereum.exceptions.ERPCManagerException(message: str)[source]

Bases: ERPCBaseException

Raised for exceptions related to manager classes such as nonce managers or gas managers

exception pythereum.exceptions.ERPCRequestException(code: int, message: str = 'Generic ERPC Error')[source]

Bases: ERPCBaseException

Raised when an error is returned from the Ethereum RPC.

exception pythereum.exceptions.ERPCSubscriptionException(message: str)[source]

Bases: ERPCBaseException

Raised when a subscription request is rejected by a host or for other generic subscription errors.

pythereum.gas_managers module

Bases: object

Class which allows access to different kinds of gas management strategies and stores their data.

Accepts an EthRPC instance or URL to be used for its gas management strategies It is recommended to start the pool for a given EthRPC instance before using a gas management strategy, otherwise the program will slow down as the pool will be opened and then closed

clear_info()[source]: Clears all stored information in the GasManager object

clear_informed_info()[source]: Clears stored info about informed_manager from the GasManager object

clear_naive_info()[source]: Clears stored info about naive_manager from GasManager object

informed_manager(success_multiplier: float = 0.95, fail_multiplier: float = 1.25, initial_gas_price: int = None, initial_fee_price: int = None, initial_priority_fee_price: int = None) → InformedGasManager[source]

Creates, yields and manages an InformedGasManager object.

This InformedGasManager can fill transactions by calling

im.fill_transaction(tx)

Note that this is not asynchronous like other transaction filling methods, as it relies on no external info

To tell the gas manager the status of a transaction call one of the following functions:

im.gas_fail() # For when a transaction has failed due to gas too low im.execution_fail() # For when a transaction has failed due to an execution reversion im.execution_success() # For when a transaction has succeeded in execution

naive_manager() → NaiveGasManager[source]

Creates, yields and manages a NaiveGasManager object.

This NaiveGasManager can fill transactions by calling

await nm.fill_transaction(tx, strategy=GasStrategy object)

strategy can be replaced with dict of form: {“gas”: GasStrategy object, “maxFeePerGas”: GasStrategy object, “maxPriorityFeePerGas”: GasStrategy object}

class pythereum.gas_managers.InformedGasManager(rpc: EthRPC = None, max_gas_price: float | pythereum.common.EthDenomination | None = None, max_fee_price: float | pythereum.common.EthDenomination | None = None, max_priority_price: float | pythereum.common.EthDenomination | None = None, fail_multiplier: float = 1.25, success_multiplier: float = 0.95)[source]

Bases: object

This InformedGasManager can fill transactions by calling

im.fill_transaction(tx)

Note that this is not asynchronous like other transaction filling methods, as it relies on no external info

To tell the gas manager the status of a transaction call one of the following functions:

im.gas_fail() # For when a transaction has failed due to gas too low im.execution_fail() # For when a transaction has failed due to an execution reversion im.execution_success() # For when a transaction has succeeded in execution

execution_fail()[source]

execution_success()[source]

fill_transaction(tx: dict | pythereum.dclasses.Transaction | list[dict] | list[pythereum.dclasses.Transaction])[source]

gas_fail()[source]

class pythereum.gas_managers.NaiveGasManager(rpc: EthRPC = None, max_gas_price: float | pythereum.common.EthDenomination | None = None, max_fee_price: float | pythereum.common.EthDenomination | None = None, max_priority_price: float | pythereum.common.EthDenomination | None = None)[source]

Bases: object

This NaiveGasManager can fill transactions by calling:

await nm.fill_transaction(tx, strategy=GasStrategy object)

strategy can be replaced with dict of form:

{“gas”: GasStrategy object, “maxFeePerGas”: GasStrategy object, “maxPriorityFeePerGas”: GasStrategy object}

custom_pricing(prices)[source]

async fill_transaction(tx: dict | pythereum.dclasses.Transaction | list[dict] | list[pythereum.dclasses.Transaction], strategy: pythereum.common.GasStrategy | dict[str, pythereum.common.GasStrategy] = GasStrategy.mean_price, use_stored: bool = False) → None[source]

async suggest(strategy: GasStrategy, attribute: str, use_stored_results: bool = False) → float[source]

pythereum.rpc module

class pythereum.rpc.EthRPC(url: str, pool_size: int = 5, use_socket_pool: bool = True)[source]

Bases: object

A class managing communication with an Ethereum node via the Ethereum JSON RPC API.

async call(transaction: dict | pythereum.dclasses.Transaction | list[dict] | list[pythereum.dclasses.Transaction], block_specifier: int | pythereum.common.BlockTag | str | list[int | pythereum.common.BlockTag | str] = BlockTag.latest, websocket: websockets.legacy.client.WebSocketClientProtocol | None = None)[source]: Executes a message call immediately without creating a transaction on the blockchain, useful for tests :param transaction: Transaction call object, represented as a dict or Transaction object :param block_specifier: A specifier, either int or tag, delineating the block number to execute the transaction :param websocket: An optional external websocket for calls to this function :return: Hex value of the executed contract

async close_pool() → None[source]: Closes the socket pool, this is important to not leave hanging open connections

Generates and returns an estimate of how much gas is necessary to allow the transaction to complete. The transaction will not be added to the blockchain. Note that the estimate may be significantly more than the amount of gas actually used by the transaction, for a variety of reasons including EVM mechanics and node performance.

Uses the same parameters as eth_call, see above

async get_accounts(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → list[str | pythereum.common.HexStr][source]: Returns a list of addresses owned by client.

async get_balance(contract_address: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], block_specifier: int | pythereum.common.BlockTag | str | list[int | pythereum.common.BlockTag | str] = BlockTag.latest, websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int | list[int][source]: Gets the balance of the account a given address points to :param contract_address: Contract address, its balance will be gotten at the block specified by quant_or_tag :param block_specifier: A selector for a block, can be a specifier such as ‘latest’ or an integer block number :param websocket: An optional external websocket for calls to this function :return: An integer balance in Wei of a given contract

async get_block_by_hash(data: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], full_object: bool | list[bool] = True, websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → pythereum.dclasses.Block | list[pythereum.dclasses.Block][source]: Returns a Block object which represents a block’s state :param data: Hash of a block :param full_object: Boolean specifying whether the desired return uses full transactions or transaction hashes :param websocket: An optional external websocket for calls to this function :return: A Block object representing blocks by either full transactions or transaction hashes

async get_block_by_number(block_specifier: int | pythereum.common.BlockTag | str | list[int | pythereum.common.BlockTag | str], full_object: bool | list[bool] = True, websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → pythereum.dclasses.Block | list[pythereum.dclasses.Block] | None[source]: Returns a Block object which represents a block’s state :param block_specifier: A specifier, either int or tag, delineating the block number to get :param full_object: Boolean specifying whether the desired return uses full transactions or transaction hashes :param websocket: An optional external websocket for calls to this function :return: A Block object representing blocks by either full transactions or transaction hashes

async get_block_number(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]: cannot be batched :return: Integer number indicating the number of the most recently mined block

async get_chain_id(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]

async get_client_version(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → str[source]: Returns the current client version

async get_code(data: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], block_specifier: int | pythereum.common.BlockTag | str | list[int | pythereum.common.BlockTag | str], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr][source]: Returns code at a given address for a given block number.

async get_coinbase(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → str | pythereum.common.HexStr[source]

Returns a list of all logs matching filter with given id. Used with other filter creation methods taking in their filter numbers as input.

Parameters

filter_id – A filter ID generated by creating a filter previously
websocket – An optional external websocket for calls to this function

Returns

list of Hex strings (or list of lists) indicating changes made since filter was last checked

Returns a list of all logs matching the filter with the provided ID

Parameters

filter_id – A filter ID generated by creating a filter previously
websocket – An optional external websocket for calls to this function

Returns

list or list of lists of Log objects indicating the new logs created since the filter was last checked

async get_gas_price(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]: Returns the current price per gas in Wei Cannot be batched :return: Integer number representing gas price in Wei

async get_hashrate(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]

Parameters

from_block – Block from which the filter is active
to_block – Block to which the filter is active
address – Contract address or list of addresses from which logs should originate
topics – Array of 32 byte data topics, topics are order dependent
websocket – An optional external websocket for calls to this function

Returns

Returns a list of log objects or nothing if no changes have occurred since last poll

async get_net_listening(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → bool[source]: Returns whether a client is actively listening for network connections

async get_net_peer_count(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]: Returns the number of peers connected to the client

async get_net_version(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]: Returns the network version ID

async get_protocol_version(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]

async get_subscription(method: SubscriptionType, websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → str[source]: Supporting function for self.subscribe, opening a subscription for the provided websocket

async get_sync_status(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → bool | pythereum.dclasses.Sync[source]

async get_transaction_by_block_hash_and_index(data: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], index: int | list[int], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → pythereum.dclasses.TransactionFull | list[pythereum.dclasses.TransactionFull][source]

Returns information about a transaction by block hash and transaction index position.

Parameters

data – A hash (or list thereof) delineating the block number(s) to get
index – The index position(s) of the transaction(s) for the given block
websocket – An optional external websocket for calls to this function

Returns

A TransactionFull object (or list thereof) containing information about the selected transaction(s)

Returns information about a transaction by block number and transaction index position.

Parameters

block_specifier – A specifier, either int or tag, (or list thereof) delineating the block number(s) to get
index – The index position(s) of the transaction for the given block(s)
websocket – An optional external websocket for calls to this function

Returns

A TransactionFull object (or list thereof) containing information about the selected transaction(s)

async get_transaction_by_hash(data: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → pythereum.dclasses.TransactionFull | list[pythereum.dclasses.TransactionFull][source]

Returns the information about a transaction requested by transaction hash.

Parameters

data – A hash (or list thereof) delineating the transaction(s) to get
websocket – An optional external websocket for calls to this function

Returns

A TransactionFull object (or list thereof) containing information about the selected transaction(s)

async get_transaction_count(address: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], block_specifier: int | pythereum.common.BlockTag | str | list[int | pythereum.common.BlockTag | str] = BlockTag.latest, websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int | list[int][source]: Gets the number of transactions sent from a given EOA address :param address: The address of an externally owned account :param block_specifier: A selector for a block, can be a specifier such as ‘latest’ or an integer block number :param websocket: An optional external websocket for calls to this function :return: Integer number of transactions

async get_transaction_count_by_hash(data: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int | list[int][source]: Returns the number of transactions in a block from a block matching the given block hash.

async get_transaction_count_by_number(block_specifier: int | pythereum.common.BlockTag | str | list[int | pythereum.common.BlockTag | str], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int | list[int][source]: Returns the number of transactions in a block matching the given block number.

async get_transaction_receipt(tx_hash: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], retries: int = 0, websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → pythereum.dclasses.Receipt | list[pythereum.dclasses.Receipt][source]

Gets the receipt of a transaction given its hash, the definition of a receipt can be seen in dclasses.py

Parameters

tx_hash – Hash of the transaction from which the receipt should be derived
retries – (Optional) The maximum number of retries to attempt until receipt has been generated
websocket – (Optional) external websocket for calls to this function

Returns

Transaction receipt object or list of transaction receipts

Returns information about an uncle of a block by hash and uncle index position.

Parameters

data – A hash delineating the block number to get
index – The index position(s) of the uncle(s) for the given block(s)
websocket – An optional external websocket for calls to this function

Returns

A Block object (or list thereof) containing information about the selected uncle(s)

Returns information about an uncle of a block by number and uncle index position.

Parameters

block_specifier – A specifier, either int or tag, (or list thereof) delineating the block number(s) to get
index – The index position(s) of the uncle(s) for the given block(s)
websocket – An optional external websocket for calls to this function

Returns

A Block object (or list thereof) containing information about the selected uncle(s)

async get_uncle_count_by_hash(data: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int | list[int][source]: Returns the number of uncles in a block from a block matching the given block hash.

async get_uncle_count_by_number(block_specifier: int | pythereum.common.BlockTag | str | list[int | pythereum.common.BlockTag | str], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int | list[int][source]: Returns the number of uncles in a block from a block matching the given block number.

async is_mining(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → bool[source]

async new_block_filter(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]: Creates a filter in the endpoint to notify when a new block arrives. To check if the state has changed use EthRPC.get_filter_changes()

Creates a filter object based on filter parameters to notify when the state changes. To check if the state has changed use EthRPC.get_filter_changes()

Parameters

from_block – Block from which the filter is active
to_block – Block to which the filter is active
address – Contract address or list of addresses from which logs should originate
topics – Array of 32 byte data topics, topics are order dependent
websocket – An optional external websocket for calls to this function

Returns

Returns an integer filter ID

async new_pending_transaction_filter(websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → int[source]: Creates a filter in the endpoint to notify when new pending transactions arrive. To check if the state has changed use EthRPC.get_filter_changes()

pool_connected() → bool[source]

Sends a custom method name method_name to the endpoint’s url with the given parameter list

Parameters

method_name – A string indicating the method name to be called with the given parameters
params – A list of parameters to be sent for the function call
websocket – An optional external websocket for calls to this function

Returns

Returns the result of the given transaction,

if the function does not exist for the given params an error will be raised

async send_raw_transaction(raw_transaction: str | pythereum.common.HexStr | list[str] | list[pythereum.common.HexStr], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None)[source]: Returns the receipt of a transaction by transaction hash :param raw_transaction: The hash of a transaction :param websocket: An optional external websocket for calls to this function :type: 32 Byte Hex :return: Transaction receipt object

async send_transaction(transaction: dict | list[dict], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None)[source]: Creates a new message call transaction or contract creation

Returns Keccak-256 of the given data

Parameters

data – A string of data for keccak conversion
websocket – An optional external websocket for calls to this function

Calculates the ethereum specific signature.

Parameters

data – 20 byte address(es) to sign with
message – N byte message(s) to be signed
websocket – An optional external websocket for calls to this function

Returns

Hex string(s) containing signature(s) for given data

async sign_transaction(tx: pythereum.dclasses.TransactionFull | dict | list[pythereum.dclasses.TransactionFull] | list[dict], websocket: websockets.legacy.client.WebSocketClientProtocol | None = None) → pythereum.common.HexStr | list[pythereum.common.HexStr][source]

Signs a transaction that can be submitted to the network at a later time using with eth_sendRawTransaction.

Parameters

tx – Transaction object(s) or dict(s) defining transaction parameters to be signed
websocket – An optional external websocket for calls to this function

Returns

Hex string(s) containing result(s) of signed transaction(s)

async start_pool() → None[source]: Exposes the ability to start the ERPC’s socket pool before the first method call

subscribe(method: SubscriptionType) → Subscription[source]

Parameters: method – The subscription’s type, determined by a preset enum of possible types

This function is decorated with an async context manager such that it can be used in an async with statement

A subscription object is returned generated with a unique subscription id The subscription has a .recv() method to get the latest data returned by the ethereum endpoint for the given subscription. In the future a custom version of this function may be made to support custom subscription types.

Uninstalls a filter with a given name. Should always be called when a filter is no longer needed.

Parameters

filter_id – A filter ID generated by creating a filter previously
websocket – An optional external websocket for calls to this function

Returns

bool or list of bools indicating the success or failure of each filter uninstallation

async unsubscribe(subscription_id: str | pythereum.common.HexStr, websocket: websockets.legacy.client.WebSocketClientProtocol | None = None)[source]

Parameters

subscription_id – String subscription id returned by eth_subscribe
websocket – An optional external websocket for calls to this function

Returns

The return of this function is not meant to be caught, though it does exist

class pythereum.rpc.NonceManager(rpc: pythereum.rpc.EthRPC | str | None = None)[source]

Bases: object

Manages the nonces of addresses for the purposes of constructing transactions, requires an RPC connection in order to get starting transaction counts for each address.

This currently operates assuming no other sources are creating transactions from a given address.

async fill_transaction(tx: dict | pythereum.dclasses.Transaction | list[dict] | list[pythereum.dclasses.Transaction]) → None[source]: This function mutates input transaction dictionaries such that they are filled with the correct nonce values

async next_nonce(address: str | pythereum.common.HexStr) → int[source]

class pythereum.rpc.Subscription(subscription_id: str, socket: WebSocketClientProtocol, subscription_type: SubscriptionType = SubscriptionType.new_heads)[source]

Bases: object

A representation of a subscription to receive information from an ethereum endpoint Information is automatically decoded into an appropriate format upon being retrieved from the endpoint

static logs_decoder(data: Any) → Log[source]

static new_heads_decoder(data: Any) → Block[source]

static new_pending_transactions_decoder(data: Any) → HexStr[source]

async recv() → pythereum.dclasses.Block | pythereum.dclasses.Log | pythereum.common.HexStr | pythereum.dclasses.Sync[source]: infinite async generator function which will yield new information retrieved from a websocket

static syncing_decoder(data: Any) → Sync[source]

pythereum.rpc.convert_eth(quantity: float | str | pythereum.common.HexStr, convert_from: EthDenomination, covert_to: EthDenomination) → float[source]: Converts eth values from a given denomination to another. Strings passed in are automatically decoded from hexadecimal to integers, as are Hex values

pythereum.rpc.parse_results(res: str | dict, is_subscription: bool = False, builder: str = None) → Any[source]: Validates and parses the results of an RPC

pythereum.socket_pool module

class pythereum.socket_pool.WebsocketPool(url: str, pool_size: int = 6)[source]

Bases: object

Builds a pool of reusable websockets from which to pull Greatly improves speed over having to handshake a new connection for each request

get_socket() → WebSocketClientProtocol[source]

Returns: Returns a list of websockets to use

The websockets will be returned to the main pool upon exiting the with statement in which this should be called

async quit() → None[source]: Pulls all sockets from the object queue, closes them and resets variables

async start() → None[source]: Initialises the correct number of connections Restarts the websocket pool if run while already connected

pythereum package

Submodules

pythereum.builders module

pythereum.common module

pythereum.dclasses module

pythereum.exceptions module

pythereum.gas_managers module

pythereum.rpc module

pythereum.socket_pool module

Module contents