"""Generated wrapper for AssetProxyOwner Solidity contract."""
# pylint: disable=too-many-arguments
import json
from typing import ( # pylint: disable=unused-import
Any,
List,
Optional,
Tuple,
Union,
)
from eth_utils import to_checksum_address
from mypy_extensions import TypedDict # pylint: disable=unused-import
from hexbytes import HexBytes
from web3 import Web3
from web3.contract import ContractFunction
from web3.datastructures import AttributeDict
from web3.providers.base import BaseProvider
from zero_ex.contract_wrappers.bases import ContractMethod, Validator
from zero_ex.contract_wrappers.tx_params import TxParams
# Try to import a custom validator class definition; if there isn't one,
# declare one that we can instantiate for the default argument to the
# constructor for AssetProxyOwner below.
try:
# both mypy and pylint complain about what we're doing here, but this
# works just fine, so their messages have been disabled here.
from . import ( # type: ignore # pylint: disable=import-self
AssetProxyOwnerValidator,
)
except ImportError:
[docs] class AssetProxyOwnerValidator( # type: ignore
Validator
):
"""No-op input validator."""
try:
from .middleware import MIDDLEWARE # type: ignore
except ImportError:
pass
[docs]class MaxOwnerCountMethod(ContractMethod):
"""Various interfaces to the MAX_OWNER_COUNT method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address)
self._underlying_method = contract_function
[docs] def call(self, tx_params: Optional[TxParams] = None) -> int:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method().call(tx_params.as_dict())
return int(returned)
[docs] def estimate_gas(self, tx_params: Optional[TxParams] = None) -> int:
"""Estimate gas consumption of method call."""
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method().estimateGas(tx_params.as_dict())
[docs]class AddOwnerMethod(ContractMethod):
"""Various interfaces to the addOwner method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(self, owner: str, tx_params: Optional[TxParams] = None) -> None:
"""Execute underlying contract method via eth_call.
Allows to add a new owner. Transaction has to be sent by wallet.
:param owner: Address of new owner.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(owner) = self.validate_and_normalize_inputs(owner)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(owner).call(tx_params.as_dict())
[docs] def send_transaction(
self, owner: str, tx_params: Optional[TxParams] = None
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Allows to add a new owner. Transaction has to be sent by wallet.
:param owner: Address of new owner.
:param tx_params: transaction parameters
"""
(owner) = self.validate_and_normalize_inputs(owner)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner).transact(tx_params.as_dict())
[docs] def build_transaction(
self, owner: str, tx_params: Optional[TxParams] = None
) -> dict:
"""Construct calldata to be used as input to the method."""
(owner) = self.validate_and_normalize_inputs(owner)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner).buildTransaction(
tx_params.as_dict()
)
[docs] def estimate_gas(
self, owner: str, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(owner) = self.validate_and_normalize_inputs(owner)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner).estimateGas(tx_params.as_dict())
[docs]class ChangeRequirementMethod(ContractMethod):
"""Various interfaces to the changeRequirement method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, _required: int, tx_params: Optional[TxParams] = None
) -> None:
"""Execute underlying contract method via eth_call.
Allows to change the number of required confirmations. Transaction has
to be sent by wallet.
:param _required: Number of required confirmations.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(_required) = self.validate_and_normalize_inputs(_required)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(_required).call(tx_params.as_dict())
[docs] def send_transaction(
self, _required: int, tx_params: Optional[TxParams] = None
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Allows to change the number of required confirmations. Transaction has
to be sent by wallet.
:param _required: Number of required confirmations.
:param tx_params: transaction parameters
"""
(_required) = self.validate_and_normalize_inputs(_required)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(_required).transact(tx_params.as_dict())
[docs] def build_transaction(
self, _required: int, tx_params: Optional[TxParams] = None
) -> dict:
"""Construct calldata to be used as input to the method."""
(_required) = self.validate_and_normalize_inputs(_required)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(_required).buildTransaction(
tx_params.as_dict()
)
[docs] def estimate_gas(
self, _required: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(_required) = self.validate_and_normalize_inputs(_required)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(_required).estimateGas(
tx_params.as_dict()
)
[docs]class ChangeTimeLockMethod(ContractMethod):
"""Various interfaces to the changeTimeLock method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, _seconds_time_locked: int, tx_params: Optional[TxParams] = None
) -> None:
"""Execute underlying contract method via eth_call.
Changes the duration of the time lock for transactions.
:param _secondsTimeLocked: Duration needed after a transaction is
confirmed and before it becomes executable, in seconds.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(_seconds_time_locked) = self.validate_and_normalize_inputs(
_seconds_time_locked
)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(_seconds_time_locked).call(tx_params.as_dict())
[docs] def send_transaction(
self, _seconds_time_locked: int, tx_params: Optional[TxParams] = None
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Changes the duration of the time lock for transactions.
:param _secondsTimeLocked: Duration needed after a transaction is
confirmed and before it becomes executable, in seconds.
:param tx_params: transaction parameters
"""
(_seconds_time_locked) = self.validate_and_normalize_inputs(
_seconds_time_locked
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(_seconds_time_locked).transact(
tx_params.as_dict()
)
[docs] def build_transaction(
self, _seconds_time_locked: int, tx_params: Optional[TxParams] = None
) -> dict:
"""Construct calldata to be used as input to the method."""
(_seconds_time_locked) = self.validate_and_normalize_inputs(
_seconds_time_locked
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(_seconds_time_locked).buildTransaction(
tx_params.as_dict()
)
[docs] def estimate_gas(
self, _seconds_time_locked: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(_seconds_time_locked) = self.validate_and_normalize_inputs(
_seconds_time_locked
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(_seconds_time_locked).estimateGas(
tx_params.as_dict()
)
[docs]class ConfirmTransactionMethod(ContractMethod):
"""Various interfaces to the confirmTransaction method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> None:
"""Execute underlying contract method via eth_call.
Allows an owner to confirm a transaction.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(transaction_id).call(tx_params.as_dict())
[docs] def send_transaction(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Allows an owner to confirm a transaction.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).transact(
tx_params.as_dict()
)
[docs] def build_transaction(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> dict:
"""Construct calldata to be used as input to the method."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).buildTransaction(
tx_params.as_dict()
)
[docs] def estimate_gas(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).estimateGas(
tx_params.as_dict()
)
[docs]class ConfirmationTimesMethod(ContractMethod):
"""Various interfaces to the confirmationTimes method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(self, index_0: int, tx_params: Optional[TxParams] = None) -> int:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
(index_0) = self.validate_and_normalize_inputs(index_0)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(index_0).call(tx_params.as_dict())
return int(returned)
[docs] def estimate_gas(
self, index_0: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(index_0) = self.validate_and_normalize_inputs(index_0)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(index_0).estimateGas(
tx_params.as_dict()
)
[docs]class ConfirmationsMethod(ContractMethod):
"""Various interfaces to the confirmations method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, index_0: int, index_1: str, tx_params: Optional[TxParams] = None
) -> bool:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
(index_0, index_1) = self.validate_and_normalize_inputs(
index_0, index_1
)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(index_0, index_1).call(
tx_params.as_dict()
)
return bool(returned)
[docs] def estimate_gas(
self, index_0: int, index_1: str, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(index_0, index_1) = self.validate_and_normalize_inputs(
index_0, index_1
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(index_0, index_1).estimateGas(
tx_params.as_dict()
)
[docs]class ExecuteTransactionMethod(ContractMethod):
"""Various interfaces to the executeTransaction method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> None:
"""Execute underlying contract method via eth_call.
Allows anyone to execute a confirmed transaction. Transactions *must*
encode the values with the signature "bytes[] data, address[]
destinations, uint256[] values" The `destination` and `value` fields of
the transaction in storage are ignored. All function calls must be
successful or the entire call will revert.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(transaction_id).call(tx_params.as_dict())
[docs] def send_transaction(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Allows anyone to execute a confirmed transaction. Transactions *must*
encode the values with the signature "bytes[] data, address[]
destinations, uint256[] values" The `destination` and `value` fields of
the transaction in storage are ignored. All function calls must be
successful or the entire call will revert.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).transact(
tx_params.as_dict()
)
[docs] def build_transaction(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> dict:
"""Construct calldata to be used as input to the method."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).buildTransaction(
tx_params.as_dict()
)
[docs] def estimate_gas(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).estimateGas(
tx_params.as_dict()
)
[docs]class FunctionCallTimeLocksMethod(ContractMethod):
"""Various interfaces to the functionCallTimeLocks method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self,
index_0: Union[bytes, str],
index_1: str,
tx_params: Optional[TxParams] = None,
) -> Tuple[bool, int]:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
(index_0, index_1) = self.validate_and_normalize_inputs(
index_0, index_1
)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(index_0, index_1).call(
tx_params.as_dict()
)
return (
returned[0],
returned[1],
)
[docs] def estimate_gas(
self,
index_0: Union[bytes, str],
index_1: str,
tx_params: Optional[TxParams] = None,
) -> int:
"""Estimate gas consumption of method call."""
(index_0, index_1) = self.validate_and_normalize_inputs(
index_0, index_1
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(index_0, index_1).estimateGas(
tx_params.as_dict()
)
[docs]class GetConfirmationCountMethod(ContractMethod):
"""Various interfaces to the getConfirmationCount method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> int:
"""Execute underlying contract method via eth_call.
Returns number of confirmations of a transaction.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
:returns: Number of confirmations.
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(transaction_id).call(
tx_params.as_dict()
)
return int(returned)
[docs] def estimate_gas(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).estimateGas(
tx_params.as_dict()
)
[docs]class GetConfirmationsMethod(ContractMethod):
"""Various interfaces to the getConfirmations method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> List[str]:
"""Execute underlying contract method via eth_call.
Returns array with owner addresses, which confirmed transaction.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
:returns: Returns array of owner addresses.
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(transaction_id).call(
tx_params.as_dict()
)
return [str(element) for element in returned]
[docs] def estimate_gas(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).estimateGas(
tx_params.as_dict()
)
[docs]class GetOwnersMethod(ContractMethod):
"""Various interfaces to the getOwners method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address)
self._underlying_method = contract_function
[docs] def call(self, tx_params: Optional[TxParams] = None) -> List[str]:
"""Execute underlying contract method via eth_call.
Returns list of owners.
:param tx_params: transaction parameters
:returns: List of owner addresses.
"""
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method().call(tx_params.as_dict())
return [str(element) for element in returned]
[docs] def estimate_gas(self, tx_params: Optional[TxParams] = None) -> int:
"""Estimate gas consumption of method call."""
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method().estimateGas(tx_params.as_dict())
[docs]class GetTransactionCountMethod(ContractMethod):
"""Various interfaces to the getTransactionCount method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self,
pending: bool,
executed: bool,
tx_params: Optional[TxParams] = None,
) -> int:
"""Execute underlying contract method via eth_call.
Returns total number of transactions after filers are applied.
:param executed: Include executed transactions.
:param pending: Include pending transactions.
:param tx_params: transaction parameters
:returns: Total number of transactions after filters are applied.
"""
(pending, executed) = self.validate_and_normalize_inputs(
pending, executed
)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(pending, executed).call(
tx_params.as_dict()
)
return int(returned)
[docs] def estimate_gas(
self,
pending: bool,
executed: bool,
tx_params: Optional[TxParams] = None,
) -> int:
"""Estimate gas consumption of method call."""
(pending, executed) = self.validate_and_normalize_inputs(
pending, executed
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(pending, executed).estimateGas(
tx_params.as_dict()
)
[docs]class GetTransactionIdsMethod(ContractMethod):
"""Various interfaces to the getTransactionIds method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self,
_from: int,
to: int,
pending: bool,
executed: bool,
tx_params: Optional[TxParams] = None,
) -> List[int]:
"""Execute underlying contract method via eth_call.
Returns list of transaction IDs in defined range.
:param executed: Include executed transactions.
:param from: Index start position of transaction array.
:param pending: Include pending transactions.
:param to: Index end position of transaction array.
:param tx_params: transaction parameters
:returns: Returns array of transaction IDs.
"""
(_from, to, pending, executed) = self.validate_and_normalize_inputs(
_from, to, pending, executed
)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(_from, to, pending, executed).call(
tx_params.as_dict()
)
return [int(element) for element in returned]
[docs] def estimate_gas(
self,
_from: int,
to: int,
pending: bool,
executed: bool,
tx_params: Optional[TxParams] = None,
) -> int:
"""Estimate gas consumption of method call."""
(_from, to, pending, executed) = self.validate_and_normalize_inputs(
_from, to, pending, executed
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(
_from, to, pending, executed
).estimateGas(tx_params.as_dict())
[docs]class IsConfirmedMethod(ContractMethod):
"""Various interfaces to the isConfirmed method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> bool:
"""Execute underlying contract method via eth_call.
Returns the confirmation status of a transaction.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
:returns: Confirmation status.
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(transaction_id).call(
tx_params.as_dict()
)
return bool(returned)
[docs] def estimate_gas(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).estimateGas(
tx_params.as_dict()
)
[docs]class IsOwnerMethod(ContractMethod):
"""Various interfaces to the isOwner method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(self, index_0: str, tx_params: Optional[TxParams] = None) -> bool:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
(index_0) = self.validate_and_normalize_inputs(index_0)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(index_0).call(tx_params.as_dict())
return bool(returned)
[docs] def estimate_gas(
self, index_0: str, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(index_0) = self.validate_and_normalize_inputs(index_0)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(index_0).estimateGas(
tx_params.as_dict()
)
[docs]class OwnersMethod(ContractMethod):
"""Various interfaces to the owners method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(self, index_0: int, tx_params: Optional[TxParams] = None) -> str:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
(index_0) = self.validate_and_normalize_inputs(index_0)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(index_0).call(tx_params.as_dict())
return str(returned)
[docs] def estimate_gas(
self, index_0: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(index_0) = self.validate_and_normalize_inputs(index_0)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(index_0).estimateGas(
tx_params.as_dict()
)
[docs]class RegisterFunctionCallMethod(ContractMethod):
"""Various interfaces to the registerFunctionCall method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self,
has_custom_time_lock: bool,
function_selector: Union[bytes, str],
destination: str,
new_seconds_time_locked: int,
tx_params: Optional[TxParams] = None,
) -> None:
"""Execute underlying contract method via eth_call.
Registers a custom timelock to a specific function selector /
destination combo
:param destination: Address of destination where function will be
called.
:param functionSelector: 4 byte selector of registered function.
:param hasCustomTimeLock: True if timelock is custom.
:param newSecondsTimeLocked: Duration in seconds needed after a
transaction is confirmed to become executable.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
) = self.validate_and_normalize_inputs(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
).call(tx_params.as_dict())
[docs] def send_transaction(
self,
has_custom_time_lock: bool,
function_selector: Union[bytes, str],
destination: str,
new_seconds_time_locked: int,
tx_params: Optional[TxParams] = None,
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Registers a custom timelock to a specific function selector /
destination combo
:param destination: Address of destination where function will be
called.
:param functionSelector: 4 byte selector of registered function.
:param hasCustomTimeLock: True if timelock is custom.
:param newSecondsTimeLocked: Duration in seconds needed after a
transaction is confirmed to become executable.
:param tx_params: transaction parameters
"""
(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
) = self.validate_and_normalize_inputs(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
).transact(tx_params.as_dict())
[docs] def build_transaction(
self,
has_custom_time_lock: bool,
function_selector: Union[bytes, str],
destination: str,
new_seconds_time_locked: int,
tx_params: Optional[TxParams] = None,
) -> dict:
"""Construct calldata to be used as input to the method."""
(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
) = self.validate_and_normalize_inputs(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
).buildTransaction(tx_params.as_dict())
[docs] def estimate_gas(
self,
has_custom_time_lock: bool,
function_selector: Union[bytes, str],
destination: str,
new_seconds_time_locked: int,
tx_params: Optional[TxParams] = None,
) -> int:
"""Estimate gas consumption of method call."""
(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
) = self.validate_and_normalize_inputs(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(
has_custom_time_lock,
function_selector,
destination,
new_seconds_time_locked,
).estimateGas(tx_params.as_dict())
[docs]class RemoveOwnerMethod(ContractMethod):
"""Various interfaces to the removeOwner method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(self, owner: str, tx_params: Optional[TxParams] = None) -> None:
"""Execute underlying contract method via eth_call.
Allows to remove an owner. Transaction has to be sent by wallet.
:param owner: Address of owner.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(owner) = self.validate_and_normalize_inputs(owner)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(owner).call(tx_params.as_dict())
[docs] def send_transaction(
self, owner: str, tx_params: Optional[TxParams] = None
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Allows to remove an owner. Transaction has to be sent by wallet.
:param owner: Address of owner.
:param tx_params: transaction parameters
"""
(owner) = self.validate_and_normalize_inputs(owner)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner).transact(tx_params.as_dict())
[docs] def build_transaction(
self, owner: str, tx_params: Optional[TxParams] = None
) -> dict:
"""Construct calldata to be used as input to the method."""
(owner) = self.validate_and_normalize_inputs(owner)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner).buildTransaction(
tx_params.as_dict()
)
[docs] def estimate_gas(
self, owner: str, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(owner) = self.validate_and_normalize_inputs(owner)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner).estimateGas(tx_params.as_dict())
[docs]class ReplaceOwnerMethod(ContractMethod):
"""Various interfaces to the replaceOwner method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, owner: str, new_owner: str, tx_params: Optional[TxParams] = None
) -> None:
"""Execute underlying contract method via eth_call.
Allows to replace an owner with a new owner. Transaction has to be sent
by wallet.
:param newOwner: Address of new owner.
:param owner: Address of owner to be replaced.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(owner, new_owner) = self.validate_and_normalize_inputs(
owner, new_owner
)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(owner, new_owner).call(tx_params.as_dict())
[docs] def send_transaction(
self, owner: str, new_owner: str, tx_params: Optional[TxParams] = None
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Allows to replace an owner with a new owner. Transaction has to be sent
by wallet.
:param newOwner: Address of new owner.
:param owner: Address of owner to be replaced.
:param tx_params: transaction parameters
"""
(owner, new_owner) = self.validate_and_normalize_inputs(
owner, new_owner
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner, new_owner).transact(
tx_params.as_dict()
)
[docs] def build_transaction(
self, owner: str, new_owner: str, tx_params: Optional[TxParams] = None
) -> dict:
"""Construct calldata to be used as input to the method."""
(owner, new_owner) = self.validate_and_normalize_inputs(
owner, new_owner
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner, new_owner).buildTransaction(
tx_params.as_dict()
)
[docs] def estimate_gas(
self, owner: str, new_owner: str, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(owner, new_owner) = self.validate_and_normalize_inputs(
owner, new_owner
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(owner, new_owner).estimateGas(
tx_params.as_dict()
)
[docs]class RequiredMethod(ContractMethod):
"""Various interfaces to the required method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address)
self._underlying_method = contract_function
[docs] def call(self, tx_params: Optional[TxParams] = None) -> int:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method().call(tx_params.as_dict())
return int(returned)
[docs] def estimate_gas(self, tx_params: Optional[TxParams] = None) -> int:
"""Estimate gas consumption of method call."""
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method().estimateGas(tx_params.as_dict())
[docs]class RevokeConfirmationMethod(ContractMethod):
"""Various interfaces to the revokeConfirmation method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> None:
"""Execute underlying contract method via eth_call.
Allows an owner to revoke a confirmation for a transaction.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
self._underlying_method(transaction_id).call(tx_params.as_dict())
[docs] def send_transaction(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Allows an owner to revoke a confirmation for a transaction.
:param transactionId: Transaction ID.
:param tx_params: transaction parameters
"""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).transact(
tx_params.as_dict()
)
[docs] def build_transaction(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> dict:
"""Construct calldata to be used as input to the method."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).buildTransaction(
tx_params.as_dict()
)
[docs] def estimate_gas(
self, transaction_id: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(transaction_id) = self.validate_and_normalize_inputs(transaction_id)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(transaction_id).estimateGas(
tx_params.as_dict()
)
[docs]class SecondsTimeLockedMethod(ContractMethod):
"""Various interfaces to the secondsTimeLocked method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address)
self._underlying_method = contract_function
[docs] def call(self, tx_params: Optional[TxParams] = None) -> int:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method().call(tx_params.as_dict())
return int(returned)
[docs] def estimate_gas(self, tx_params: Optional[TxParams] = None) -> int:
"""Estimate gas consumption of method call."""
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method().estimateGas(tx_params.as_dict())
[docs]class SubmitTransactionMethod(ContractMethod):
"""Various interfaces to the submitTransaction method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self,
destination: str,
value: int,
data: Union[bytes, str],
tx_params: Optional[TxParams] = None,
) -> int:
"""Execute underlying contract method via eth_call.
Allows an owner to submit and confirm a transaction.
:param data: Transaction data payload.
:param destination: Transaction target address.
:param value: Transaction ether value.
:param tx_params: transaction parameters
:returns: the return value of the underlying method.
"""
(destination, value, data) = self.validate_and_normalize_inputs(
destination, value, data
)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(destination, value, data).call(
tx_params.as_dict()
)
return int(returned)
[docs] def send_transaction(
self,
destination: str,
value: int,
data: Union[bytes, str],
tx_params: Optional[TxParams] = None,
) -> Union[HexBytes, bytes]:
"""Execute underlying contract method via eth_sendTransaction.
Allows an owner to submit and confirm a transaction.
:param data: Transaction data payload.
:param destination: Transaction target address.
:param value: Transaction ether value.
:param tx_params: transaction parameters
"""
(destination, value, data) = self.validate_and_normalize_inputs(
destination, value, data
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(destination, value, data).transact(
tx_params.as_dict()
)
[docs] def build_transaction(
self,
destination: str,
value: int,
data: Union[bytes, str],
tx_params: Optional[TxParams] = None,
) -> dict:
"""Construct calldata to be used as input to the method."""
(destination, value, data) = self.validate_and_normalize_inputs(
destination, value, data
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(
destination, value, data
).buildTransaction(tx_params.as_dict())
[docs] def estimate_gas(
self,
destination: str,
value: int,
data: Union[bytes, str],
tx_params: Optional[TxParams] = None,
) -> int:
"""Estimate gas consumption of method call."""
(destination, value, data) = self.validate_and_normalize_inputs(
destination, value, data
)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(destination, value, data).estimateGas(
tx_params.as_dict()
)
[docs]class TransactionCountMethod(ContractMethod):
"""Various interfaces to the transactionCount method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address)
self._underlying_method = contract_function
[docs] def call(self, tx_params: Optional[TxParams] = None) -> int:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method().call(tx_params.as_dict())
return int(returned)
[docs] def estimate_gas(self, tx_params: Optional[TxParams] = None) -> int:
"""Estimate gas consumption of method call."""
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method().estimateGas(tx_params.as_dict())
[docs]class TransactionsMethod(ContractMethod):
"""Various interfaces to the transactions method."""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
contract_function: ContractFunction,
validator: Validator = None,
):
"""Persist instance data."""
super().__init__(web3_or_provider, contract_address, validator)
self._underlying_method = contract_function
[docs] def call(
self, index_0: int, tx_params: Optional[TxParams] = None
) -> Tuple[str, int, Union[bytes, str], bool]:
"""Execute underlying contract method via eth_call.
:param tx_params: transaction parameters
"""
(index_0) = self.validate_and_normalize_inputs(index_0)
tx_params = super().normalize_tx_params(tx_params)
returned = self._underlying_method(index_0).call(tx_params.as_dict())
return (
returned[0],
returned[1],
returned[2],
returned[3],
)
[docs] def estimate_gas(
self, index_0: int, tx_params: Optional[TxParams] = None
) -> int:
"""Estimate gas consumption of method call."""
(index_0) = self.validate_and_normalize_inputs(index_0)
tx_params = super().normalize_tx_params(tx_params)
return self._underlying_method(index_0).estimateGas(
tx_params.as_dict()
)
# pylint: disable=too-many-public-methods,too-many-instance-attributes
[docs]class AssetProxyOwner:
"""Wrapper class for AssetProxyOwner Solidity contract.
All method parameters of type `bytes`:code: should be encoded as UTF-8,
which can be accomplished via `str.encode("utf_8")`:code:.
"""
max_owner_count: MaxOwnerCountMethod
"""Constructor-initialized instance of
:class:`MaxOwnerCountMethod`.
"""
add_owner: AddOwnerMethod
"""Constructor-initialized instance of
:class:`AddOwnerMethod`.
"""
change_requirement: ChangeRequirementMethod
"""Constructor-initialized instance of
:class:`ChangeRequirementMethod`.
"""
change_time_lock: ChangeTimeLockMethod
"""Constructor-initialized instance of
:class:`ChangeTimeLockMethod`.
"""
confirm_transaction: ConfirmTransactionMethod
"""Constructor-initialized instance of
:class:`ConfirmTransactionMethod`.
"""
confirmation_times: ConfirmationTimesMethod
"""Constructor-initialized instance of
:class:`ConfirmationTimesMethod`.
"""
confirmations: ConfirmationsMethod
"""Constructor-initialized instance of
:class:`ConfirmationsMethod`.
"""
execute_transaction: ExecuteTransactionMethod
"""Constructor-initialized instance of
:class:`ExecuteTransactionMethod`.
"""
function_call_time_locks: FunctionCallTimeLocksMethod
"""Constructor-initialized instance of
:class:`FunctionCallTimeLocksMethod`.
"""
get_confirmation_count: GetConfirmationCountMethod
"""Constructor-initialized instance of
:class:`GetConfirmationCountMethod`.
"""
get_confirmations: GetConfirmationsMethod
"""Constructor-initialized instance of
:class:`GetConfirmationsMethod`.
"""
get_owners: GetOwnersMethod
"""Constructor-initialized instance of
:class:`GetOwnersMethod`.
"""
get_transaction_count: GetTransactionCountMethod
"""Constructor-initialized instance of
:class:`GetTransactionCountMethod`.
"""
get_transaction_ids: GetTransactionIdsMethod
"""Constructor-initialized instance of
:class:`GetTransactionIdsMethod`.
"""
is_confirmed: IsConfirmedMethod
"""Constructor-initialized instance of
:class:`IsConfirmedMethod`.
"""
is_owner: IsOwnerMethod
"""Constructor-initialized instance of
:class:`IsOwnerMethod`.
"""
owners: OwnersMethod
"""Constructor-initialized instance of
:class:`OwnersMethod`.
"""
register_function_call: RegisterFunctionCallMethod
"""Constructor-initialized instance of
:class:`RegisterFunctionCallMethod`.
"""
remove_owner: RemoveOwnerMethod
"""Constructor-initialized instance of
:class:`RemoveOwnerMethod`.
"""
replace_owner: ReplaceOwnerMethod
"""Constructor-initialized instance of
:class:`ReplaceOwnerMethod`.
"""
required: RequiredMethod
"""Constructor-initialized instance of
:class:`RequiredMethod`.
"""
revoke_confirmation: RevokeConfirmationMethod
"""Constructor-initialized instance of
:class:`RevokeConfirmationMethod`.
"""
seconds_time_locked: SecondsTimeLockedMethod
"""Constructor-initialized instance of
:class:`SecondsTimeLockedMethod`.
"""
submit_transaction: SubmitTransactionMethod
"""Constructor-initialized instance of
:class:`SubmitTransactionMethod`.
"""
transaction_count: TransactionCountMethod
"""Constructor-initialized instance of
:class:`TransactionCountMethod`.
"""
transactions: TransactionsMethod
"""Constructor-initialized instance of
:class:`TransactionsMethod`.
"""
[docs] def __init__(
self,
web3_or_provider: Union[Web3, BaseProvider],
contract_address: str,
validator: AssetProxyOwnerValidator = None,
):
"""Get an instance of wrapper for smart contract.
:param web3_or_provider: Either an instance of `web3.Web3`:code: or
`web3.providers.base.BaseProvider`:code:
:param contract_address: where the contract has been deployed
:param validator: for validation of method inputs.
"""
# pylint: disable=too-many-statements
self.contract_address = contract_address
if not validator:
validator = AssetProxyOwnerValidator(
web3_or_provider, contract_address
)
web3 = None
if isinstance(web3_or_provider, BaseProvider):
web3 = Web3(web3_or_provider)
elif isinstance(web3_or_provider, Web3):
web3 = web3_or_provider
else:
raise TypeError(
"Expected parameter 'web3_or_provider' to be an instance of either"
+ " Web3 or BaseProvider"
)
# if any middleware was imported, inject it
try:
MIDDLEWARE
except NameError:
pass
else:
try:
for middleware in MIDDLEWARE:
web3.middleware_onion.inject(
middleware["function"], layer=middleware["layer"],
)
except ValueError as value_error:
if value_error.args == (
"You can't add the same un-named instance twice",
):
pass
self._web3_eth = web3.eth
functions = self._web3_eth.contract(
address=to_checksum_address(contract_address),
abi=AssetProxyOwner.abi(),
).functions
self.max_owner_count = MaxOwnerCountMethod(
web3_or_provider, contract_address, functions.MAX_OWNER_COUNT
)
self.add_owner = AddOwnerMethod(
web3_or_provider, contract_address, functions.addOwner, validator
)
self.change_requirement = ChangeRequirementMethod(
web3_or_provider,
contract_address,
functions.changeRequirement,
validator,
)
self.change_time_lock = ChangeTimeLockMethod(
web3_or_provider,
contract_address,
functions.changeTimeLock,
validator,
)
self.confirm_transaction = ConfirmTransactionMethod(
web3_or_provider,
contract_address,
functions.confirmTransaction,
validator,
)
self.confirmation_times = ConfirmationTimesMethod(
web3_or_provider,
contract_address,
functions.confirmationTimes,
validator,
)
self.confirmations = ConfirmationsMethod(
web3_or_provider,
contract_address,
functions.confirmations,
validator,
)
self.execute_transaction = ExecuteTransactionMethod(
web3_or_provider,
contract_address,
functions.executeTransaction,
validator,
)
self.function_call_time_locks = FunctionCallTimeLocksMethod(
web3_or_provider,
contract_address,
functions.functionCallTimeLocks,
validator,
)
self.get_confirmation_count = GetConfirmationCountMethod(
web3_or_provider,
contract_address,
functions.getConfirmationCount,
validator,
)
self.get_confirmations = GetConfirmationsMethod(
web3_or_provider,
contract_address,
functions.getConfirmations,
validator,
)
self.get_owners = GetOwnersMethod(
web3_or_provider, contract_address, functions.getOwners
)
self.get_transaction_count = GetTransactionCountMethod(
web3_or_provider,
contract_address,
functions.getTransactionCount,
validator,
)
self.get_transaction_ids = GetTransactionIdsMethod(
web3_or_provider,
contract_address,
functions.getTransactionIds,
validator,
)
self.is_confirmed = IsConfirmedMethod(
web3_or_provider,
contract_address,
functions.isConfirmed,
validator,
)
self.is_owner = IsOwnerMethod(
web3_or_provider, contract_address, functions.isOwner, validator
)
self.owners = OwnersMethod(
web3_or_provider, contract_address, functions.owners, validator
)
self.register_function_call = RegisterFunctionCallMethod(
web3_or_provider,
contract_address,
functions.registerFunctionCall,
validator,
)
self.remove_owner = RemoveOwnerMethod(
web3_or_provider,
contract_address,
functions.removeOwner,
validator,
)
self.replace_owner = ReplaceOwnerMethod(
web3_or_provider,
contract_address,
functions.replaceOwner,
validator,
)
self.required = RequiredMethod(
web3_or_provider, contract_address, functions.required
)
self.revoke_confirmation = RevokeConfirmationMethod(
web3_or_provider,
contract_address,
functions.revokeConfirmation,
validator,
)
self.seconds_time_locked = SecondsTimeLockedMethod(
web3_or_provider, contract_address, functions.secondsTimeLocked
)
self.submit_transaction = SubmitTransactionMethod(
web3_or_provider,
contract_address,
functions.submitTransaction,
validator,
)
self.transaction_count = TransactionCountMethod(
web3_or_provider, contract_address, functions.transactionCount
)
self.transactions = TransactionsMethod(
web3_or_provider,
contract_address,
functions.transactions,
validator,
)
[docs] def get_confirmation_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for Confirmation event.
:param tx_hash: hash of transaction emitting Confirmation event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.Confirmation()
.processReceipt(tx_receipt)
)
[docs] def get_confirmation_time_set_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for ConfirmationTimeSet event.
:param tx_hash: hash of transaction emitting ConfirmationTimeSet event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.ConfirmationTimeSet()
.processReceipt(tx_receipt)
)
[docs] def get_deposit_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for Deposit event.
:param tx_hash: hash of transaction emitting Deposit event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.Deposit()
.processReceipt(tx_receipt)
)
[docs] def get_execution_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for Execution event.
:param tx_hash: hash of transaction emitting Execution event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.Execution()
.processReceipt(tx_receipt)
)
[docs] def get_execution_failure_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for ExecutionFailure event.
:param tx_hash: hash of transaction emitting ExecutionFailure event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.ExecutionFailure()
.processReceipt(tx_receipt)
)
[docs] def get_function_call_time_lock_registration_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for FunctionCallTimeLockRegistration event.
:param tx_hash: hash of transaction emitting
FunctionCallTimeLockRegistration event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.FunctionCallTimeLockRegistration()
.processReceipt(tx_receipt)
)
[docs] def get_owner_addition_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for OwnerAddition event.
:param tx_hash: hash of transaction emitting OwnerAddition event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.OwnerAddition()
.processReceipt(tx_receipt)
)
[docs] def get_owner_removal_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for OwnerRemoval event.
:param tx_hash: hash of transaction emitting OwnerRemoval event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.OwnerRemoval()
.processReceipt(tx_receipt)
)
[docs] def get_requirement_change_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for RequirementChange event.
:param tx_hash: hash of transaction emitting RequirementChange event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.RequirementChange()
.processReceipt(tx_receipt)
)
[docs] def get_revocation_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for Revocation event.
:param tx_hash: hash of transaction emitting Revocation event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.Revocation()
.processReceipt(tx_receipt)
)
[docs] def get_submission_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for Submission event.
:param tx_hash: hash of transaction emitting Submission event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.Submission()
.processReceipt(tx_receipt)
)
[docs] def get_time_lock_change_event(
self, tx_hash: Union[HexBytes, bytes]
) -> Tuple[AttributeDict]:
"""Get log entry for TimeLockChange event.
:param tx_hash: hash of transaction emitting TimeLockChange event
"""
tx_receipt = self._web3_eth.getTransactionReceipt(tx_hash)
return (
self._web3_eth.contract(
address=to_checksum_address(self.contract_address),
abi=AssetProxyOwner.abi(),
)
.events.TimeLockChange()
.processReceipt(tx_receipt)
)
[docs] @staticmethod
def abi():
"""Return the ABI to the underlying contract."""
return json.loads(
'[{"inputs":[{"internalType":"bytes4[]","name":"_functionSelectors","type":"bytes4[]"},{"internalType":"address[]","name":"_destinations","type":"address[]"},{"internalType":"uint128[]","name":"_functionCallTimeLockSeconds","type":"uint128[]"},{"internalType":"address[]","name":"_owners","type":"address[]"},{"internalType":"uint256","name":"_required","type":"uint256"},{"internalType":"uint256","name":"_defaultSecondsTimeLocked","type":"uint256"}],"payable":false,"stateMutability":"nonpayable","type":"constructor"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"sender","type":"address"},{"indexed":true,"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"Confirmation","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"uint256","name":"transactionId","type":"uint256"},{"indexed":false,"internalType":"uint256","name":"confirmationTime","type":"uint256"}],"name":"ConfirmationTimeSet","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"sender","type":"address"},{"indexed":false,"internalType":"uint256","name":"value","type":"uint256"}],"name":"Deposit","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"Execution","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"ExecutionFailure","type":"event"},{"anonymous":false,"inputs":[{"indexed":false,"internalType":"bytes4","name":"functionSelector","type":"bytes4"},{"indexed":false,"internalType":"address","name":"destination","type":"address"},{"indexed":false,"internalType":"bool","name":"hasCustomTimeLock","type":"bool"},{"indexed":false,"internalType":"uint128","name":"newSecondsTimeLocked","type":"uint128"}],"name":"FunctionCallTimeLockRegistration","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"owner","type":"address"}],"name":"OwnerAddition","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"owner","type":"address"}],"name":"OwnerRemoval","type":"event"},{"anonymous":false,"inputs":[{"indexed":false,"internalType":"uint256","name":"required","type":"uint256"}],"name":"RequirementChange","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"sender","type":"address"},{"indexed":true,"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"Revocation","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"Submission","type":"event"},{"anonymous":false,"inputs":[{"indexed":false,"internalType":"uint256","name":"secondsTimeLocked","type":"uint256"}],"name":"TimeLockChange","type":"event"},{"payable":true,"stateMutability":"payable","type":"fallback"},{"constant":true,"inputs":[],"name":"MAX_OWNER_COUNT","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"internalType":"address","name":"owner","type":"address"}],"name":"addOwner","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":false,"inputs":[{"internalType":"uint256","name":"_required","type":"uint256"}],"name":"changeRequirement","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":false,"inputs":[{"internalType":"uint256","name":"_secondsTimeLocked","type":"uint256"}],"name":"changeTimeLock","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":false,"inputs":[{"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"confirmTransaction","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[{"internalType":"uint256","name":"index_0","type":"uint256"}],"name":"confirmationTimes","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"uint256","name":"index_0","type":"uint256"},{"internalType":"address","name":"index_1","type":"address"}],"name":"confirmations","outputs":[{"internalType":"bool","name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"executeTransaction","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[{"internalType":"bytes4","name":"index_0","type":"bytes4"},{"internalType":"address","name":"index_1","type":"address"}],"name":"functionCallTimeLocks","outputs":[{"internalType":"bool","name":"hasCustomTimeLock","type":"bool"},{"internalType":"uint128","name":"secondsTimeLocked","type":"uint128"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"getConfirmationCount","outputs":[{"internalType":"uint256","name":"count","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"getConfirmations","outputs":[{"internalType":"address[]","name":"_confirmations","type":"address[]"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"getOwners","outputs":[{"internalType":"address[]","name":"","type":"address[]"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"bool","name":"pending","type":"bool"},{"internalType":"bool","name":"executed","type":"bool"}],"name":"getTransactionCount","outputs":[{"internalType":"uint256","name":"count","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"uint256","name":"from","type":"uint256"},{"internalType":"uint256","name":"to","type":"uint256"},{"internalType":"bool","name":"pending","type":"bool"},{"internalType":"bool","name":"executed","type":"bool"}],"name":"getTransactionIds","outputs":[{"internalType":"uint256[]","name":"_transactionIds","type":"uint256[]"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"isConfirmed","outputs":[{"internalType":"bool","name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"address","name":"index_0","type":"address"}],"name":"isOwner","outputs":[{"internalType":"bool","name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"uint256","name":"index_0","type":"uint256"}],"name":"owners","outputs":[{"internalType":"address","name":"","type":"address"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"internalType":"bool","name":"hasCustomTimeLock","type":"bool"},{"internalType":"bytes4","name":"functionSelector","type":"bytes4"},{"internalType":"address","name":"destination","type":"address"},{"internalType":"uint128","name":"newSecondsTimeLocked","type":"uint128"}],"name":"registerFunctionCall","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":false,"inputs":[{"internalType":"address","name":"owner","type":"address"}],"name":"removeOwner","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":false,"inputs":[{"internalType":"address","name":"owner","type":"address"},{"internalType":"address","name":"newOwner","type":"address"}],"name":"replaceOwner","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"required","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"internalType":"uint256","name":"transactionId","type":"uint256"}],"name":"revokeConfirmation","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"secondsTimeLocked","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"internalType":"address","name":"destination","type":"address"},{"internalType":"uint256","name":"value","type":"uint256"},{"internalType":"bytes","name":"data","type":"bytes"}],"name":"submitTransaction","outputs":[{"internalType":"uint256","name":"transactionId","type":"uint256"}],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"transactionCount","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"uint256","name":"index_0","type":"uint256"}],"name":"transactions","outputs":[{"internalType":"address","name":"destination","type":"address"},{"internalType":"uint256","name":"value","type":"uint256"},{"internalType":"bytes","name":"data","type":"bytes"},{"internalType":"bool","name":"executed","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"}]' # noqa: E501 (line-too-long)
)
# pylint: disable=too-many-lines