Replace private utils with public get_event_abi and `get_function_a…

…bi` utils.

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))
 

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,19 +863,29 @@ 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(
  cls,
  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(

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: