diff --git a/web3/_utils/contracts.py b/web3/_utils/contracts.py index 1d80d47d49..3fef59b9a3 100644 --- a/web3/_utils/contracts.py +++ b/web3/_utils/contracts.py @@ -1,4 +1,3 @@ -import functools from typing import ( TYPE_CHECKING, Any, @@ -20,7 +19,6 @@ ) from eth_typing import ( ABI, - ABIEvent, ABIFunction, ChecksumAddress, HexStr, @@ -35,24 +33,16 @@ is_list_like, is_text, ) -from eth_utils.toolz import ( - pipe, -) from hexbytes import ( HexBytes, ) from web3.utils.abi import ( get_abi_input_types, + get_function_abi, ) from web3._utils.abi import ( - abi_to_signature, check_if_arguments_can_be_encoded, - filter_by_argument_count, - filter_by_argument_name, - filter_by_encodability, - filter_by_name, - filter_by_type, get_aligned_abi_inputs, get_fallback_func_abi, get_receive_func_abi, @@ -120,98 +110,6 @@ def extract_argument_types(*args: Sequence[Any]) -> str: return ",".join(collapsed_args) -def find_matching_event_abi( - abi: ABI, - event_name: Optional[str] = None, - argument_names: Optional[Sequence[str]] = None, -) -> ABIEvent: - filters = [ - functools.partial(filter_by_type, "event"), - ] - - if event_name is not None: - filters.append(functools.partial(filter_by_name, event_name)) - - if argument_names is not None: - filters.append(functools.partial(filter_by_argument_name, argument_names)) - - event_abi_candidates = pipe(abi, *filters) - - if len(event_abi_candidates) == 1: - return event_abi_candidates[0] - elif not event_abi_candidates: - raise Web3ValueError("No matching events found") - else: - raise Web3ValueError("Multiple events found") - - -def find_matching_fn_abi( - abi: ABI, - abi_codec: ABICodec, - fn_identifier: Optional[Union[str, Type[FallbackFn], Type[ReceiveFn]]] = None, - args: Optional[Sequence[Any]] = None, - kwargs: Optional[Any] = None, -) -> ABIFunction: - args = args or tuple() - kwargs = kwargs or dict() - num_arguments = len(args) + len(kwargs) - - if fn_identifier is FallbackFn: - return get_fallback_func_abi(abi) - - if fn_identifier is ReceiveFn: - return get_receive_func_abi(abi) - - if not is_text(fn_identifier): - raise Web3TypeError("Unsupported function identifier") - - name_filter = functools.partial(filter_by_name, fn_identifier) - arg_count_filter = functools.partial(filter_by_argument_count, num_arguments) - encoding_filter = functools.partial(filter_by_encodability, abi_codec, args, kwargs) - - function_candidates = pipe(abi, name_filter, arg_count_filter, encoding_filter) - - if len(function_candidates) == 1: - return function_candidates[0] - else: - matching_identifiers = name_filter(abi) - matching_function_signatures = [ - abi_to_signature(func) for func in matching_identifiers - ] - - arg_count_matches = len(arg_count_filter(matching_identifiers)) - encoding_matches = len(encoding_filter(matching_identifiers)) - - if arg_count_matches == 0: - diagnosis = ( - "\nFunction invocation failed due to improper number of arguments." - ) - elif encoding_matches == 0: - diagnosis = ( - "\nFunction invocation failed due to no matching argument types." - ) - elif encoding_matches > 1: - diagnosis = ( - "\nAmbiguous argument encoding. " - "Provided arguments can be encoded to multiple functions " - "matching this call." - ) - - collapsed_args = extract_argument_types(args) - collapsed_kwargs = dict( - {(k, extract_argument_types([v])) for k, v in kwargs.items()} - ) - message = ( - f"\nCould not identify the intended function with name `{fn_identifier}`, " - f"positional arguments with type(s) `{collapsed_args}` and " - f"keyword arguments with type(s) `{collapsed_kwargs}`." - f"\nFound {len(matching_identifiers)} function(s) with " - f"the name `{fn_identifier}`: {matching_function_signatures}{diagnosis}" - ) - - raise Web3ValidationError(message) - - def encode_abi( w3: Union["AsyncWeb3", "Web3"], abi: ABIFunction, @@ -271,7 +169,7 @@ def prepare_transaction( TODO: add new prepare_deploy_transaction API """ if fn_abi is None: - fn_abi = find_matching_fn_abi( + fn_abi = get_function_abi( contract_abi, w3.codec, fn_identifier, fn_args, fn_kwargs ) @@ -387,7 +285,7 @@ def get_function_info( kwargs = {} if fn_abi is None: - fn_abi = find_matching_fn_abi(contract_abi, abi_codec, fn_name, args, kwargs) + fn_abi = get_function_abi(contract_abi, abi_codec, fn_name, args, kwargs) fn_selector = encode_hex(function_abi_to_4byte_selector(fn_abi)) diff --git a/web3/contract/base_contract.py b/web3/contract/base_contract.py index 87bc9511b3..45859656dd 100644 --- a/web3/contract/base_contract.py +++ b/web3/contract/base_contract.py @@ -23,6 +23,7 @@ Address, ChecksumAddress, HexStr, + ValidationError, ) from eth_utils import ( add_0x_prefix, @@ -50,9 +51,6 @@ from web3._utils.contracts import ( decode_transaction_data, encode_abi, - find_matching_event_abi, - find_matching_fn_abi, - get_function_info, prepare_transaction, ) from web3._utils.datatypes import ( @@ -116,6 +114,11 @@ TxParams, TxReceipt, ) +from web3.utils.abi import ( + get_event_abi, + get_function_abi, + get_function_info, +) if TYPE_CHECKING: from web3 import ( # noqa: F401 @@ -152,7 +155,7 @@ def __init__(self, *argument_names: Tuple[str]) -> None: @classmethod def _get_event_abi(cls) -> ABIEvent: - return find_matching_event_abi(cls.contract_abi, event_name=cls.event_name) + return get_event_abi(cls.contract_abi, event_name=cls.event_name) @combomethod def process_receipt( @@ -478,12 +481,12 @@ def __init__(self, abi: Optional[ABIFunction] = None) -> None: def _set_function_info(self) -> None: if not self.abi: - self.abi = find_matching_fn_abi( + self.abi = get_function_abi( self.contract_abi, - self.w3.codec, self.function_identifier, self.args, self.kwargs, + self.w3.codec, ) if self.function_identifier in [FallbackFn, ReceiveFn]: self.selector = encode_hex(b"") @@ -740,10 +743,10 @@ def encode_abi( """ fn_abi, fn_selector, fn_arguments = get_function_info( fn_name, - cls.w3.codec, contract_abi=cls.abi, args=args, kwargs=kwargs, + codec=cls.w3.codec, ) if data is None: @@ -860,9 +863,16 @@ def _find_matching_fn_abi( args: Optional[Any] = None, kwargs: Optional[Any] = None, ) -> ABIFunction: - return find_matching_fn_abi( - cls.abi, cls.w3.codec, fn_identifier=fn_identifier, args=args, kwargs=kwargs - ) + try: + return get_function_abi( + cls.abi, + fn_identifier=fn_identifier, + args=args, + kwargs=kwargs, + codec=cls.w3.codec, + ) + except MismatchedABI as message: + raise Web3ValidationError(message) @classmethod def _find_matching_event_abi( @@ -870,9 +880,12 @@ def _find_matching_event_abi( event_name: Optional[str] = None, argument_names: Optional[Sequence[str]] = None, ) -> ABIEvent: - return find_matching_event_abi( - abi=cls.abi, event_name=event_name, argument_names=argument_names - ) + try: + return get_event_abi( + abi=cls.abi, event_name=event_name, argument_names=argument_names + ) + except (ValidationError, ValueError) as message: + raise Web3ValidationError(message) @combomethod def _encode_constructor_data( diff --git a/web3/contract/utils.py b/web3/contract/utils.py index beab47076f..171db6db20 100644 --- a/web3/contract/utils.py +++ b/web3/contract/utils.py @@ -36,7 +36,6 @@ async_fill_transaction_defaults, ) from web3._utils.contracts import ( - find_matching_fn_abi, prepare_transaction, ) from web3._utils.normalizers import ( @@ -56,6 +55,9 @@ TContractFn, TxParams, ) +from web3.utils.abi import ( + get_function_abi, +) if TYPE_CHECKING: from web3 import ( # noqa: F401 @@ -104,8 +106,8 @@ def call_contract_function( ) if fn_abi is None: - fn_abi = find_matching_fn_abi( - contract_abi, w3.codec, function_identifier, args, kwargs + fn_abi = get_function_abi( + contract_abi, function_identifier, args, kwargs, abi_codec=w3.codec ) try: @@ -321,8 +323,8 @@ async def async_call_contract_function( ) if fn_abi is None: - fn_abi = find_matching_fn_abi( - contract_abi, async_w3.codec, function_identifier, args, kwargs + fn_abi = get_function_abi( + contract_abi, function_identifier, args, kwargs, abi_codec=async_w3.codec ) try: