API¶
Clients¶
- class pons.Client(provider: Provider)[source]¶
An Ethereum RPC client.
- async with session() AsyncIterator[ClientSession][source]¶
Opens a session to the client allowing the backend to optimize sequential requests.
- class pons.ClientSession[source]¶
An open session to the provider.
The methods of this class may raise the following exceptions:
ProviderError,ContractLegacyError,ContractError,ContractPanic,TransactionFailed,BadResponseFormat,ABIDecodingError,- await broadcast_transact(signer: Signer, call: BaseBoundMethodCall, amount: None | Amount = None, gas: int | None = None) TxHash[source]¶
Broadcasts the transaction without waiting for it to be finalized. See
transact()for the information on the parameters.
- await broadcast_transfer(signer: Signer, destination_address: Address, amount: Amount, gas: int | None = None) TxHash[source]¶
Broadcasts the fund transfer transaction, but does not wait for it to be processed. If
gasisNone, the required amount of gas is estimated first, otherwise the provided value is used.
- await call(call: BaseBoundMethodCall, block: int | BlockLabel = BlockLabel.LATEST, sender_address: None | Address = None) Any[source]¶
Sends a prepared contact method call to the provided address. Returns the decoded output.
If
sender_addressis provided, it will be included in the call and affect the return value if the method usesmsg.senderinternally.
- await deploy(signer: Signer, call: BoundConstructorCall, amount: None | Amount = None, gas: int | None = None) DeployedContract[source]¶
Deploys the contract passing
argsto the constructor. IfgasisNone, the required amount of gas is estimated first, otherwise the provided value is used. Waits for the transaction to be confirmed.amountdenotes the amount of currency to send to the constructor.Raises
TransactionFailedif the transaction was submitted successfully, but could not be processed. If gas estimation is run, see the additional errors that may be raised in the docs forestimate_deploy().
- await estimate_deploy(sender_address: Address, call: BoundConstructorCall, amount: None | Amount = None, block: int | BlockLabel = BlockLabel.LATEST) int[source]¶
Estimates the amount of gas required to deploy the contract with the given args. Use with the same
amountargument you would use to deploy the contract in production, and thesender_addressequal to the address of the transaction signer.Raises
ContractPanic,ContractLegacyError, orContractErrorif a known error was caught during the dry run. If the error was unknown, falls back toethereum_rpc.RPCError.
- await estimate_transact(sender_address: Address, call: BaseBoundMethodCall, amount: None | Amount = None, block: int | BlockLabel = BlockLabel.LATEST) int[source]¶
Estimates the amount of gas required to transact with a contract. Use with the same
amountargument you would use to transact with the contract in production, and thesender_addressequal to the address of the transaction signer.Raises
ContractPanic,ContractLegacyError, orContractErrorif a known error was caught during the dry run. If the error was unknown, falls back toethereum_rpc.RPCError.
- await estimate_transfer(source_address: Address, destination_address: Address, amount: Amount, block: int | BlockLabel = BlockLabel.LATEST) int[source]¶
Estimates the amount of gas required to transfer
amount. Raises aethereum_rpc.RPCErrorif there is not enough funds insource_address.
- await get_balance(address: Address, block: int | BlockLabel = BlockLabel.LATEST) Amount[source]¶
Query the balance of
addressatblock.
- await get_block(block_id: BlockHash | int | BlockLabel, *, with_transactions: bool = False) None | BlockInfo[source]¶
- async for ... in iter_blocks(poll_interval: int = 1) AsyncIterator[BlockHash][source]¶
Yields hashes of new blocks being mined.
- async for ... in iter_events(event_filter: BoundEventFilter, poll_interval: int = 1, from_block: int | BlockLabel = BlockLabel.LATEST, to_block: int | BlockLabel = BlockLabel.LATEST) AsyncIterator[dict[str, Any]][source]¶
Yields decoded log entries produced by the filter. The fields that were hashed when converted to topics (that is, fields of reference types) are set to
None.
- async for ... in iter_pending_transactions(poll_interval: int = 1) AsyncIterator[TxHash][source]¶
Yields hashes of new transactions being submitted.
- await transact(signer: Signer, call: BaseBoundMethodCall, amount: None | Amount = None, gas: int | None = None, return_events: None | Sequence[BoundEvent] = None) dict[BoundEvent, list[dict[str, Any]]][source]¶
Transacts with the contract using a prepared method call. If
gasisNone, the required amount of gas is estimated first, otherwise the provided value is used. Waits for the transaction to be confirmed.amountdenotes the amount of currency to send with the transaction.If any bound events are given in return_events, the provider will be queried for any firing of these events originating from the hash of the completed transaction (from the contract addresses the events are bound to), and the results will be returned as a dictionary keyed by the corresponding event object.
Raises
TransactionFailedif the transaction was submitted successfully, but could not be processed. If gas estimation is run, see the additional errors that may be raised in the docs forestimate_transact().
- await transfer(signer: Signer, destination_address: Address, amount: Amount, gas: int | None = None) None[source]¶
Transfers funds from the address of the attached signer to the destination address. If
gasisNone, the required amount of gas is estimated first, otherwise the provided value is used. Waits for the transaction to be confirmed.Raises
TransactionFailedif the transaction was submitted successfully, but could not be processed.
- await wait_for_transaction_receipt(tx_hash: TxHash, poll_latency: float = 1.0) TxReceipt[source]¶
Queries the transaction receipt waiting for
poll_latencybetween each attempt.
- property rpc: ClientSessionRPC¶
- class pons.ClientSessionRPC[source]¶
The hub for methods which directly correspond to Ethereum RPC calls.
The methods of this class may raise
ProviderError(coming from the lower level) orBadResponseFormat(failed to deserialize the response into the expected type).- await eth_call(call: BaseBoundMethodCall, block: int | BlockLabel = BlockLabel.LATEST, sender_address: None | Address = None) Any[source]¶
Sends a prepared contact method call to the provided address. Returns the decoded output.
If
sender_addressis provided, it will be included in the call and affect the return value if the method usesmsg.senderinternally.
- await eth_estimate_gas(params: EstimateGasParams, block: int | BlockLabel) int[source]¶
Calls the
eth_estimateGasRPC method.
- await eth_get_balance(address: Address, block: int | BlockLabel = BlockLabel.LATEST) Amount[source]¶
Calls the
eth_getBalanceRPC method.
- await eth_get_block_by_hash(block_hash: BlockHash, *, with_transactions: bool = False) None | BlockInfo[source]¶
Calls the
eth_getBlockByHashRPC method.
- await eth_get_block_by_number(block: int | BlockLabel = BlockLabel.LATEST, *, with_transactions: bool = False) None | BlockInfo[source]¶
Calls the
eth_getBlockByNumberRPC method.
- await eth_get_code(address: Address, block: int | BlockLabel = BlockLabel.LATEST) bytes[source]¶
Calls the
eth_getCodeRPC method.
- await eth_get_filter_changes(filter_: BlockFilter | PendingTransactionFilter | LogFilter) tuple[BlockHash, ...] | tuple[TxHash, ...] | tuple[LogEntry, ...][source]¶
Calls the
eth_getFilterChangesRPC method. Depending on whatfilter_was, returns a tuple of corresponding results.
- await eth_get_filter_logs(filter_: BlockFilter | PendingTransactionFilter | LogFilter) tuple[BlockHash, ...] | tuple[TxHash, ...] | tuple[LogEntry, ...][source]¶
Calls the
eth_getFilterLogsRPC method.
- await eth_get_logs(source: None | Address | Iterable[Address] = None, event_filter: None | EventFilter = None, from_block: int | BlockLabel = BlockLabel.LATEST, to_block: int | BlockLabel = BlockLabel.LATEST) tuple[LogEntry, ...][source]¶
Calls the
eth_getLogsRPC method.
- await eth_get_storage_at(address: Address, position: int, block: int | BlockLabel = BlockLabel.LATEST) bytes[source]¶
Calls the
eth_getCodeRPC method.
- await eth_get_transaction_by_hash(tx_hash: TxHash) None | TxInfo[source]¶
Calls the
eth_getTransactionByHashRPC method.
- await eth_get_transaction_count(address: Address, block: int | BlockLabel = BlockLabel.LATEST) int[source]¶
Calls the
eth_getTransactionCountRPC method.
- await eth_get_transaction_receipt(tx_hash: TxHash) None | TxReceipt[source]¶
Calls the
eth_getTransactionReceiptRPC method.
- await eth_new_block_filter() BlockFilter[source]¶
Calls the
eth_newBlockFilterRPC method.
- await eth_new_filter(source: None | Address | Iterable[Address] = None, event_filter: None | EventFilter = None, from_block: int | BlockLabel = BlockLabel.LATEST, to_block: int | BlockLabel = BlockLabel.LATEST) LogFilter[source]¶
Calls the
eth_newFilterRPC method.
- await eth_new_pending_transaction_filter() PendingTransactionFilter[source]¶
Calls the
eth_newPendingTransactionFilterRPC method.
Providers¶
Fallback providers¶
- class pons.FallbackProvider(providers: Mapping[str, Provider], strategy: FallbackStrategyFactory | None = None, *, same_provider: bool = False)[source]¶
Bases:
ProviderA provider that encompasses several providers and for every request tries every one of them until the request is successful. The order is chosen according to the given strategy.
If
same_providerisTrue, the given providers are treated as endpoints pointing to the same physical provider, for the purpose of stateful requests (e.g. filter creation).If
strategyisNone, an instance ofPriorityFallbackis used.If a request attempt results in an error for which
use_fallbackreturnsTrue, the next provider based on the chosen strategy will be selected. Otherwise (or if it is the last provider), the error is raised normally.- errors() list[tuple[ProviderPath, ProviderError]][source]¶
Returns the list of recorded errors for sub-providers.
Only the most recent error for every sub-provider is recorded. Querying this method clears the recorded errors.
- class pons.CycleFallback(weights: None | Iterable[int] = None)[source]¶
Bases:
FallbackStrategyFactoryCreates a strategy where the providers are cycled such that the number of times a given provider is first in the priority list is equal to the corresponding entry in
weights(the length of which should match the number of providers). Ifweightsis not given, a list of1will be used.
- class pons.PriorityFallback[source]¶
Bases:
FallbackStrategyFactory
- class pons.FallbackStrategyFactory[source]¶
Bases:
ABCAn abstract class defining a fallback strategy factory for multiple providers. This will be called in
FallbackProviderto create an actual strategy object (which may be mutated).- abstractmethod make_strategy(provider_ids: Set[str]) FallbackStrategy[source]¶
Returns a strategy object.
Errors¶
Provider level¶
- class pons.ProviderError(error: RPCError | Unreachable | InvalidResponse | ProtocolError)[source]¶
Bases:
ExceptionDescribes an error on the provider’s side.
- error: RPCError | Unreachable | InvalidResponse | ProtocolError¶
The specific error.
- class pons.InvalidResponse[source]¶
Bases:
ExceptionRaised when the remote server’s response is not of an expected format.
- class pons.Unreachable[source]¶
Bases:
ExceptionRaised when there is a problem connecting to the provider.
- class pons.ProtocolError[source]¶
-
A protocol-specific error, indicating that the provider returned an error status with no additional information allowing to categorize the error further.
See the provider-specifc derived class for this exception for more details.
- class pons.HTTPError(status_code: int, message: str)[source]¶
Bases:
ProtocolErrorRaised when the provider returns a response with a status code other than 200, and no
"error"field in the associated JSON data.
Client level¶
- class pons.BadResponseFormat[source]¶
Bases:
ExceptionRaised if the RPC provider returned an unexpectedly formatted response.
- class pons.ABIDecodingError[source]¶
Bases:
ExceptionRaised on an error when decoding a value in an Eth ABI encoded bytestring.
- class pons.TransactionFailed[source]¶
Bases:
ExceptionRaised for invalid transactions that are not contract executions (e.g. transfers or contract deployments).
- class pons._client.ContractPanicReason(value)[source]¶
Reasons leading to a contract call panicking.
- classmethod from_int(val: int) ContractPanicReason[source]¶
- ASSERTION = 1¶
If you call assert with an argument that evaluates to
false.
- COMPILER = 0¶
Used for generic compiler inserted panics.
- DIVISION_BY_ZERO = 18¶
If you divide or modulo by zero (e.g.
5 / 0or23 % 0).
- EMPTY_ARRAY = 49¶
If you call
.pop()on an empty array.
- INVALID_ENCODING = 34¶
If you access a storage byte array that is incorrectly encoded.
- INVALID_ENUM_VALUE = 33¶
If you convert a value that is too big or negative into an
enumtype.
- OUT_OF_BOUNDS = 50¶
If you access an array,
bytesNor an array slice at an out-of-bounds or negative index (i.e.x[i]wherei >= x.lengthori < 0).
- OUT_OF_MEMORY = 65¶
If you allocate too much memory or create an array that is too large.
- OVERFLOW = 17¶
If an arithmetic operation results in underflow or overflow outside of an
unchecked { ... }block.
- UNKNOWN = -1¶
Unknown panic code.
- ZERO_DEREFERENCE = 81¶
If you call a zero-initialized variable of internal function type.
- class pons.ContractPanic[source]¶
Bases:
ExceptionA panic raised in a contract call.
- Reason¶
alias of
ContractPanicReason
- classmethod from_code(code: int) ContractPanic[source]¶
- reason: ContractPanicReason¶
Parsed panic reason.
Signers¶
- class pons.Signer[source]¶
Bases:
ABCThe base class for transaction signers.
- class pons.AccountSigner(account: LocalAccount)[source]¶
Bases:
SignerA signer wrapper for
LocalAccountfrometh-accountpackage.- staticmethod create() AccountSigner[source]¶
Creates an account with a random private key.
- sign_transaction(tx_dict: Dict[str, Sequence[Dict[str, HexStr | Sequence[HexStr]]] | bytes | HexStr | int]) bytes[source]¶
Signs the given transaction and returns the RLP-packed transaction along with the signature.
- property account: LocalAccount¶
Returns the account object used to create this signer.
Contract ABI¶
- class pons.ABI_JSON¶
A JSON-ifiable object (
bool,int,float,str,None, iterable ofJSON, or mapping ofstrtoJSON).
- class pons.ContractABI(constructor: None | Constructor = None, fallback: None | Fallback = None, receive: None | Receive = None, methods: None | Iterable[Method | MultiMethod] = None, events: None | Iterable[Event] = None, errors: None | Iterable[Error] = None)[source]¶
A wrapper for contract ABI.
Contract methods are grouped by type and are accessible via the attributes below.
- classmethod from_json(json_abi: None | bool | int | float | str | Sequence[ABI_JSON] | Mapping[str, ABI_JSON]) ContractABI[source]¶
Creates this object from a JSON ABI (e.g. generated by a Solidity compiler).
- resolve_error(error_data: bytes) tuple[Error, dict[str, Any]][source]¶
Given the packed error data, attempts to find the error in the ABI and decode the data into its fields.
- constructor: Constructor¶
Contract’s constructor.
- method: Methods[Method | MultiMethod]¶
Contract’s regular methods.
- class pons.Mutability(value)[source]¶
Possible states of a contract’s method mutability.
- classmethod from_json(entry: None | bool | int | float | str | Sequence[ABI_JSON] | Mapping[str, ABI_JSON]) Mutability[source]¶
- NONPAYABLE = 'nonpayable'¶
Solidity’s
nonpayable(may write the contract state).
- PAYABLE = 'payable'¶
Solidity’s
payable(may write the contract state and accept associated funds with transactions).
- PURE = 'pure'¶
Solidity’s
pure(does not read or write the contract state).
- VIEW = 'view'¶
Solidity’s
view(may read the contract state).
- class pons.Constructor(inputs: Mapping[str, Type] | Sequence[Type], *, payable: bool = False)[source]¶
Contract constructor.
Note
If the name of a parameter given to the constructor matches a Python keyword,
_will be appended to it.- __call__(*args: Any, **kwargs: Any) ConstructorCall[source]¶
Returns an encoded call with given arguments.
- class pons.Method(name: str, mutability: Mutability, inputs: Mapping[str, Type] | Sequence[Type], outputs: None | Mapping[str, Type] | Sequence[Type] | Type = None)[source]¶
A contract method.
Note
If the name of a parameter (input or output) given to the constructor matches a Python keyword,
_will be appended to it.- __call__(*args: Any, **kwargs: Any) MethodCall[source]¶
Returns an encoded call with given arguments.
- bind(*args: Any, **kwargs: Any) BoundArguments[source]¶
- call_bound(bound_args: BoundArguments) MethodCall[source]¶
- classmethod from_json(method_entry: None | bool | int | float | str | Sequence[ABI_JSON] | Mapping[str, ABI_JSON]) Method[source]¶
Creates this object from a JSON ABI method entry.
- with_method(method: Method) MultiMethod[source]¶
- class pons.MultiMethod(*methods: Method)[source]¶
An overloaded contract method, containing several
Methodobjects with the same name but different input signatures.- __call__(*args: Any, **kwds: Any) MethodCall[source]¶
Returns an encoded call with given arguments.
- with_method(method: Method) MultiMethod[source]¶
Returns a new
MultiMethodwith the given method included.
- class pons.Event(name: str, fields: Mapping[str, Type], indexed: Set[str], *, anonymous: bool = False)[source]¶
A contract event.
Note
If the name of a field given to the constructor matches a Python keyword,
_will be appended to it.
- class pons.Error(name: str, fields: Mapping[str, Type])[source]¶
A custom contract error.
- decode_fields(data_bytes: bytes) dict[str, Any][source]¶
Decodes the error fields from the given packed data.
- class pons.Fallback(*, payable: bool = False)[source]¶
A fallback method.
Testing utilities¶
pons exposes several types useful for testing applications that connect to Ethereum RPC servers. Not intended for the production environment.
Install the feature local-provider for the local_provider module to be available.
Install the feature http-provider-server for the http_provider_server module to be available.
Compiler¶
Install with the feature compiler for it to be available.
Multicall contract¶
The library includes a helper for interacting with the Multicall contract (https://github.com/mds1/multicall3).
- class pons.Multicall(multicall3_address: Address)[source]¶
A helper for interacting with the Multicall contract (v3).
- aggregate(calls: Iterable[BoundMethodCall], *, allow_failure: bool = False) BoundMultiMethodCall[source]¶
Creates an aggregated call out of provided
calls. The return value is a list of tuples(success: bool, result: Any)in the same order ascalls.If at least one of the calls is mutating, the resulting call will be mutating as well.
If
allow_failureisFalse, a contract error in one of the calls will result in a contract error being raised. If it isTrue, any contract errors will be recorded in the corresponding return values, but no exception will be raised.Note
If
allow_failureisTrue, and one of the calls reverts, the successful calls will not be reverted as well.
- aggregate_value(calls: Iterable[tuple[BoundMethodCall, Amount]], *, allow_failure: bool = False) BoundMultiMethodValueCall[source]¶
Same as
aggregate(), but takes an associated amount that will be passed to the corresponding method. The sum of the amounts must be lesser or equal to the amount passed with the invocation of the aggregated call.
- class pons.BoundMultiMethodCall(calls: Iterable[BoundMethodCall], multicall3_address: Address, *, allow_failure: bool = False)[source]¶
Bases:
BaseBoundMethodCall- decode_output(output_bytes: bytes) list[Any][source]¶
Decodes contract output packed into the bytestring.
- property contract_abi: ContractABI¶
The corresponding contract’s ABI.
- class pons.BoundMultiMethodValueCall(calls: Iterable[tuple[BoundMethodCall, Amount]], multicall3_address: Address, *, allow_failure: bool = False)[source]¶
Bases:
BoundMethodCall- decode_output(output_bytes: bytes) list[Any][source]¶
Decodes contract output packed into the bytestring.
- property contract_abi: ContractABI¶
The corresponding contract’s ABI.
Secondary classes¶
The instances of these classes are not created by the user directly, but rather found as return values, or attributes of other objects.
- class pons.BoundConstructor[source]¶
A constructor bound to a specific contract’s bytecode.
- __call__(*args: Any, **kwargs: Any) BoundConstructorCall[source]¶
Returns a constructor call with encoded arguments and bytecode.
- class pons.BoundConstructorCall[source]¶
A constructor call with encoded arguments and bytecode.
- contract_abi: ContractABI¶
The corresponding contract’s ABI
- class pons.BoundMethod[source]¶
A regular method bound to a specific contract’s address.
- __call__(*args: Any, **kwargs: Any) BoundMethodCall[source]¶
Returns a contract call with encoded arguments bound to a specific address.
- class pons.BaseBoundMethodCall[source]¶
The base class for contract method calls bound to a specific contract address.
- abstractmethod decode_output(output_bytes: bytes) Any[source]¶
Decodes contract output packed into the bytestring.
- abstract property contract_abi: ContractABI¶
The corresponding contract’s ABI.
- class pons.BoundMethodCall[source]¶
Bases:
BaseBoundMethodCallA regular method call with encoded arguments bound to a specific contract address.
- property contract_abi: ContractABI¶
The corresponding contract’s ABI.
Utility classes¶
- class pons._contract_abi.Methods[source]¶
Bases:
Generic[MethodType].A holder for named methods which can be accessed as attributes, or iterated over.
- __getattr__(method_name: str) MethodType[source]¶
Returns the method by name.
- __iter__() Iterator[MethodType][source]¶
Returns the iterator over all methods.
- class pons._contract_abi.MethodType¶
Generic method type parameter.
- class pons._contract_abi.Signature[source]¶
Generalized signature of either inputs or outputs of a method.
- class pons._contract_abi.Method(name: str, mutability: Mutability, inputs: Mapping[str, Type] | Sequence[Type], outputs: None | Mapping[str, Type] | Sequence[Type] | Type = None)[source]¶
A contract method.
Note
If the name of a parameter (input or output) given to the constructor matches a Python keyword,
_will be appended to it.- bind(*args: Any, **kwargs: Any) BoundArguments[source]¶
- call_bound(bound_args: BoundArguments) MethodCall[source]¶
- classmethod from_json(method_entry: None | bool | int | float | str | Sequence[ABI_JSON] | Mapping[str, ABI_JSON]) Method[source]¶
Creates this object from a JSON ABI method entry.
- with_method(method: Method) MultiMethod[source]¶
Utility methods¶
- pons.get_create_address(deployer: Address, nonce: int) Address[source]¶
Returns the deterministic deployed contract address as produced by
CREATEopcode. Here deployer is the contract address invokingCREATE(if initiated in a contract), or the transaction initiator, if a contract is created via an RPC transaction.init_codeis the deployment code (seedata_bytes).
- pons.get_create2_address(deployer: Address, init_code: bytes, salt: bytes) Address[source]¶
Returns the deterministic deployed contract address as produced by
CREATE2opcode. Here deployer is the contract address invokingCREATE2(not the transaction initiator),init_codeis the deployment code (seedata_bytes), andsaltis a length 32 bytestring.
Compiled and deployed contracts¶
- class pons.CompiledContract(abi: ContractABI, bytecode: bytes)[source]¶
A compiled contract (ABI and bytecode).
- classmethod from_compiler_output(json_abi: list[dict[str, None | bool | int | float | str | Sequence[ABI_JSON] | Mapping[str, ABI_JSON]]], bytecode: bytes) CompiledContract[source]¶
Creates a compiled contract object from the output of a Solidity compiler.
- abi: ContractABI¶
Contract’s ABI.
- property constructor: BoundConstructor¶
Returns the constructor bound to this contract’s bytecode.
- class pons.DeployedContract(abi: ContractABI, address: Address)[source]¶
A deployed contract (ABI and address).
- abi: ContractABI¶
Contract’s ABI.
- event: Methods[BoundEvent]¶
Contract’s events bound to the address.
- method: Methods[BoundMethod]¶
Contract’s regular methods bound to the address.
Filter objects¶
- class pons.BlockFilter[source]¶
A block filter created on a remote provider.
Expires after some time subject to the provider’s settings.
Solidity types¶
Type aliases are exported from the abi submodule.
Arrays can be obtained from Type objects by indexing them (either with an integer for a fixed-size array, or with ... for a variable-sized array).
Helper aliases are exported from pons.abi submodule:
- pons.abi.bytes(size: int | None = None) Bytes[source]¶
Returns the
bytes<size>type, orbytesifsizeisNone.
- pons.abi.address: AddressType¶
addresstype.
Actual type objects, for reference:
- class pons._abi_types.Bytes(size: int | None = None)[source]¶
Corresponds to the Solidity
bytes<size>type.
- class pons._abi_types.AddressType[source]¶
Corresponds to the Solidity
addresstype. Not to be confused withethereum_rpc.Addresswhich represents an address value.