← Files AgentBrokoARCHIVED FILE
.venv/Lib/site-packages/eth_utils/abi.py
26 KB · Sep 30, 2026 · 23:15 UTC
from collections import (
abc,
)
from collections.abc import (
Iterable,
Mapping,
Sequence,
)
import copy
import itertools
import re
from typing import (
Any,
Literal,
cast,
overload,
)
from eth_typing import (
ABI,
ABIComponent,
ABIConstructor,
ABIElement,
ABIError,
ABIEvent,
ABIFallback,
ABIFunction,
ABIReceive,
)
from eth_utils.types import (
is_list_like,
)
from .crypto import (
keccak,
)
def _align_abi_input(
arg_abi: ABIComponent, normalized_arg: Any
) -> Any | tuple[Any, ...]:
"""
Aligns the values of any mapping at any level of nesting in ``normalized_arg``
according to the layout of the corresponding abi spec.
"""
tuple_parts = _get_tuple_type_str_and_dims(arg_abi.get("type", ""))
if tuple_parts is None:
# normalized_arg is non-tuple. Just return value.
return normalized_arg
tuple_prefix, tuple_dims = tuple_parts
if tuple_dims is None:
# normalized_arg is non-list tuple. Each sub arg in `normalized_arg` will be
# aligned according to its corresponding abi.
sub_abis = cast(Iterable[ABIComponent], arg_abi.get("components", []))
else:
num_dims = tuple_dims.count("[")
# normalized_arg is list tuple. A non-list version of its abi will be used to
# align each element in `normalized_arg`.
new_abi = copy.copy(arg_abi)
new_abi["type"] = tuple_prefix + "[]" * (num_dims - 1)
sub_abis = itertools.repeat(new_abi)
if isinstance(normalized_arg, abc.Mapping):
# normalized_arg is mapping. Align values according to abi order.
aligned_arg = tuple(normalized_arg[abi["name"]] for abi in sub_abis)
else:
aligned_arg = normalized_arg
if not is_list_like(aligned_arg):
raise TypeError(
f'Expected non-string sequence for "{arg_abi.get("type")}" '
f"component type: got {aligned_arg}"
)
# convert NamedTuple to regular tuple
typing = tuple if isinstance(aligned_arg, tuple) else type(aligned_arg)
return typing(
_align_abi_input(sub_abi, sub_arg)
for sub_abi, sub_arg in zip(sub_abis, aligned_arg)
)
def _get_tuple_type_str_and_dims(s: str) -> tuple[str, str | None] | None:
"""
Takes a JSON ABI type string. For tuple type strings, returns the separated
prefix and array dimension parts. For all other strings, returns ``None``.
"""
tuple_type_str_re = "^(tuple)((\\[([1-9]\\d*\b)?])*)??$"
match = re.compile(tuple_type_str_re).match(s)
if match is not None:
tuple_prefix = match.group(1)
tuple_dims = match.group(2)
return tuple_prefix, tuple_dims
return None
def _raise_if_not_function_abi(abi_element: ABIElement) -> None:
if abi_element["type"] != "function":
raise ValueError(
f"Outputs only supported for ABI type `function`. Provided"
f" ABI type was `{abi_element.get('type')}` and outputs were "
f"`{abi_element.get('outputs')}`."
)
def _raise_if_fallback_or_receive_abi(abi_element: ABIElement) -> None:
if abi_element["type"] == "fallback" or abi_element["type"] == "receive":
raise ValueError(
f"Inputs not supported for function types `fallback` or `receive`. Provided"
f" ABI type was `{abi_element.get('type')}` with inputs "
f"`{abi_element.get('inputs')}`."
)
def collapse_if_tuple(abi: ABIComponent | dict[str, Any] | str) -> str:
"""
Extract argument types from a function or event ABI parameter.
With tuple argument types, return a Tuple of each type.
Returns the param if `abi` is an instance of str or another non-tuple
type.
:param abi: A Function or Event ABI component or a string with type info.
:type abi: `Union[ABIComponent, Dict[str, Any], str]`
:return: Type(s) for the function or event ABI param.
:rtype: `str`
.. doctest::
>>> from eth_utils.abi import collapse_if_tuple
>>> abi = {
... 'components': [
... {'name': 'anAddress', 'type': 'address'},
... {'name': 'anInt', 'type': 'uint256'},
... {'name': 'someBytes', 'type': 'bytes'},
... ],
... 'type': 'tuple',
... }
>>> collapse_if_tuple(abi)
'(address,uint256,bytes)'
"""
if isinstance(abi, str):
return abi
element_type = abi.get("type")
if not isinstance(element_type, str):
raise TypeError(
f"The 'type' must be a string, but got {repr(element_type)} of type "
f"{type(element_type)}"
)
elif not element_type.startswith("tuple"):
return element_type
delimited = ",".join(collapse_if_tuple(c) for c in abi["components"])
# Whatever comes after "tuple" is the array dims. The ABI spec states that
# this will have the form "", "[]", or "[k]".
array_dim = element_type[5:]
collapsed = f"({delimited}){array_dim}"
return collapsed
def abi_to_signature(abi_element: ABIElement) -> str:
"""
Returns a string signature representation of the function or event ABI
and arguments.
Signatures consist of the name followed by a list of arguments.
:param abi_element: ABI element.
:type abi_element: `ABIElement`
:return: Stringified ABI signature
:rtype: `str`
.. doctest::
>>> from eth_utils import abi_to_signature
>>> abi_element = {
... 'constant': False,
... 'inputs': [
... {
... 'name': 's',
... 'type': 'uint256'
... }
... ],
... 'name': 'f',
... 'outputs': [],
... 'payable': False,
... 'stateMutability': 'nonpayable',
... 'type': 'function'
... }
>>> abi_to_signature(abi_element)
'f(uint256)'
"""
signature = "{name}({input_types})"
abi_type = str(abi_element.get("type", ""))
if abi_type == "fallback" or abi_type == "receive":
return signature.format(name=abi_type, input_types="")
if abi_type == "constructor":
fn_name = abi_type
else:
fn_name = str(abi_element.get("name", abi_type))
return signature.format(
name=fn_name, input_types=",".join(get_abi_input_types(abi_element))
)
def filter_abi_by_name(abi_name: str, contract_abi: ABI) -> Sequence[ABIElement]:
"""
Get one or more function and event ABIs by name.
:param abi_name: Name of the function, event or error.
:type abi_name: `str`
:param contract_abi: Contract ABI.
:type contract_abi: `ABI`
:return: Function or event ABIs with matching name.
:rtype: `Sequence[ABIElement]`
.. doctest::
>>> from eth_utils.abi import filter_abi_by_name
>>> abi = [
... {
... "constant": False,
... "inputs": [],
... "name": "func_1",
... "outputs": [],
... "type": "function",
... },
... {
... "constant": False,
... "inputs": [
... {"name": "a", "type": "uint256"},
... ],
... "name": "func_2",
... "outputs": [],
... "type": "function",
... },
... {
... "constant": False,
... "inputs": [
... {"name": "a", "type": "uint256"},
... {"name": "b", "type": "uint256"},
... ],
... "name": "func_3",
... "outputs": [],
... "type": "function",
... },
... {
... "constant": False,
... "inputs": [
... {"name": "a", "type": "uint256"},
... {"name": "b", "type": "uint256"},
... {"name": "c", "type": "uint256"},
... ],
... "name": "func_4",
... "outputs": [],
... "type": "function",
... },
... ]
>>> filter_abi_by_name("func_1", abi)
[{'constant': False, 'inputs': [], 'name': 'func_1', 'outputs': [], \
'type': 'function'}]
"""
return [
abi
for abi in contract_abi
if (
(
abi["type"] == "function"
or abi["type"] == "event"
or abi["type"] == "error"
)
and abi["name"] == abi_name
)
]
@overload
def filter_abi_by_type(
abi_type: Literal["function"],
contract_abi: ABI,
) -> Sequence[ABIFunction]:
pass
@overload
def filter_abi_by_type(
abi_type: Literal["constructor"],
contract_abi: ABI,
) -> Sequence[ABIConstructor]:
pass
@overload
def filter_abi_by_type(
abi_type: Literal["fallback"],
contract_abi: ABI,
) -> Sequence[ABIFallback]:
pass
@overload
def filter_abi_by_type(
abi_type: Literal["receive"],
contract_abi: ABI,
) -> Sequence[ABIReceive]:
pass
@overload
def filter_abi_by_type(
abi_type: Literal["event"],
contract_abi: ABI,
) -> Sequence[ABIEvent]:
pass
@overload
def filter_abi_by_type(
abi_type: Literal["error"],
contract_abi: ABI,
) -> Sequence[ABIError]:
pass
def filter_abi_by_type(
abi_type: Literal[
"function", "constructor", "fallback", "receive", "event", "error"
],
contract_abi: ABI,
) -> Sequence[
ABIFunction | ABIConstructor | ABIFallback | ABIReceive | ABIEvent | ABIError
]:
"""
Return a list of each ``ABIElement`` that is of type ``abi_type``.
For mypy, function overloads ensures the correct type is returned based on the
``abi_type``. For example, if ``abi_type`` is "function", the return type will be
``Sequence[ABIFunction]``.
:param abi_type: Type of ABI element to filter by.
:type abi_type: `str`
:param contract_abi: Contract ABI.
:type contract_abi: `ABI`
:return: List of ABI elements of the specified type.
:rtype: `Sequence[Union[ABIFunction, ABIConstructor, ABIFallback, ABIReceive, \
ABIEvent, ABIError]]`
.. doctest::
>>> from eth_utils import filter_abi_by_type
>>> abi = [
... {"type": "function", "name": "myFunction", "inputs": [], "outputs": []},
... {"type": "function", "name": "myFunction2", "inputs": [], "outputs": []},
... {"type": "event", "name": "MyEvent", "inputs": []}
... ]
>>> filter_abi_by_type("function", abi)
[{'type': 'function', 'name': 'myFunction', 'inputs': [], 'outputs': []}, \
{'type': 'function', 'name': 'myFunction2', 'inputs': [], 'outputs': []}]
"""
if abi_type == "function":
return [abi for abi in contract_abi if abi["type"] == "function"]
elif abi_type == "constructor":
return [abi for abi in contract_abi if abi["type"] == "constructor"]
elif abi_type == "fallback":
return [abi for abi in contract_abi if abi["type"] == "fallback"]
elif abi_type == "receive":
return [abi for abi in contract_abi if abi["type"] == "receive"]
elif abi_type == "event":
return [abi for abi in contract_abi if abi["type"] == "event"]
elif abi_type == "error":
return [abi for abi in contract_abi if abi["type"] == "error"]
else:
raise ValueError(f"Unsupported ABI type: {abi_type}")
def get_all_function_abis(contract_abi: ABI) -> Sequence[ABIFunction]:
"""
Return interfaces for each function in the contract ABI.
:param contract_abi: Contract ABI.
:type contract_abi: `ABI`
:return: List of ABIs for each function interface.
:rtype: `Sequence[ABIFunction]`
.. doctest::
>>> from eth_utils import get_all_function_abis
>>> contract_abi = [
... {"type": "function", "name": "myFunction", "inputs": [], "outputs": []},
... {"type": "function", "name": "myFunction2", "inputs": [], "outputs": []},
... {"type": "event", "name": "MyEvent", "inputs": []}
... ]
>>> get_all_function_abis(contract_abi)
[{'type': 'function', 'name': 'myFunction', 'inputs': [], 'outputs': []}, \
{'type': 'function', 'name': 'myFunction2', 'inputs': [], 'outputs': []}]
"""
return filter_abi_by_type("function", contract_abi)
def get_all_event_abis(contract_abi: ABI) -> Sequence[ABIEvent]:
"""
Return interfaces for each event in the contract ABI.
:param contract_abi: Contract ABI.
:type contract_abi: `ABI`
:return: List of ABIs for each event interface.
:rtype: `Sequence[ABIEvent]`
.. doctest::
>>> from eth_utils import get_all_event_abis
>>> contract_abi = [
... {"type": "function", "name": "myFunction", "inputs": [], "outputs": []},
... {"type": "function", "name": "myFunction2", "inputs": [], "outputs": []},
... {"type": "event", "name": "MyEvent", "inputs": []}
... ]
>>> get_all_event_abis(contract_abi)
[{'type': 'event', 'name': 'MyEvent', 'inputs': []}]
"""
return filter_abi_by_type("event", contract_abi)
def get_normalized_abi_inputs(
abi_element: ABIElement,
*args: Sequence[Any] | None,
**kwargs: dict[str, Any] | None,
) -> tuple[Any, ...]:
r"""
Flattens positional args (``args``) and keyword args (``kwargs``) into a Tuple and
uses the ``abi_element`` for validation.
Checks to ensure that the correct number of args were given, no duplicate args were
given, and no unknown args were given. Returns a list of argument values aligned
to the order of inputs defined in ``abi_element``.
:param abi_element: ABI element.
:type abi_element: `ABIElement`
:param args: Positional arguments for the function.
:type args: `Optional[Sequence[Any]]`
:param kwargs: Keyword arguments for the function.
:type kwargs: `Optional[Dict[str, Any]]`
:return: Arguments list.
:rtype: `Tuple[Any, ...]`
.. doctest::
>>> from eth_utils import get_normalized_abi_inputs
>>> abi = {
... 'constant': False,
... 'inputs': [
... {
... 'name': 'name',
... 'type': 'string'
... },
... {
... 'name': 's',
... 'type': 'uint256'
... },
... {
... 'name': 't',
... 'components': [
... {'name': 'anAddress', 'type': 'address'},
... {'name': 'anInt', 'type': 'uint256'},
... {'name': 'someBytes', 'type': 'bytes'},
... ],
... 'type': 'tuple'
... }
... ],
... 'name': 'f',
... 'outputs': [],
... 'payable': False,
... 'stateMutability': 'nonpayable',
... 'type': 'function'
... }
>>> get_normalized_abi_inputs(
... abi, *('myName', 123), **{'t': ('0x1', 1, b'\x01')}
... )
('myName', 123, ('0x1', 1, b'\x01'))
"""
_raise_if_fallback_or_receive_abi(abi_element)
function_inputs = cast(Sequence[ABIComponent], abi_element.get("inputs", []))
if len(args) + len(kwargs) != len(function_inputs):
raise TypeError(
f"Incorrect argument count. Expected '{len(function_inputs)}'"
f", got '{len(args) + len(kwargs)}'."
)
# If no keyword args were given, we don't need to align them
if not kwargs:
return cast(tuple[Any, ...], args)
kwarg_names = set(kwargs.keys())
sorted_arg_names = tuple(arg_abi["name"] for arg_abi in function_inputs)
args_as_kwargs = dict(zip(sorted_arg_names, args))
# Check for duplicate args
duplicate_args = kwarg_names.intersection(args_as_kwargs.keys())
if duplicate_args:
raise TypeError(
f"{abi_element.get('name')}() got multiple values for argument(s) "
f"'{', '.join(duplicate_args)}'."
)
# Check for unknown args
# Arg names sorted to raise consistent error messages
unknown_args = tuple(sorted(kwarg_names.difference(sorted_arg_names)))
if unknown_args:
message = "{} got unexpected keyword argument(s) '{}'."
if abi_element.get("name"):
raise TypeError(
message.format(f"{abi_element.get('name')}()", ", ".join(unknown_args))
)
raise TypeError(
message.format(
f"Type: '{abi_element.get('type')}'", ", ".join(unknown_args)
)
)
# Sort args according to their position in the ABI and unzip them from their
# names
sorted_args = tuple(
zip(
*sorted(
itertools.chain(kwargs.items(), args_as_kwargs.items()),
key=lambda kv: sorted_arg_names.index(kv[0]),
)
)
)
if len(sorted_args) > 0:
return tuple(sorted_args[1])
else:
return tuple()
def get_aligned_abi_inputs(
abi_element: ABIElement,
normalized_args: tuple[Any, ...] | Mapping[Any, Any],
) -> tuple[tuple[str, ...], tuple[Any, ...]]:
"""
Returns a pair of nested Tuples containing a list of types and a list of input
values sorted by the order specified by the ``abi``.
``normalized_args`` can be obtained by using
:py:meth:`eth_utils.abi.get_normalized_abi_inputs`, which returns nested mappings
or sequences corresponding to tuple-encoded values in ``abi``.
:param abi_element: ABI element.
:type abi_element: `ABIElement`
:param normalized_args: Normalized arguments for the function.
:type normalized_args: `Union[Tuple[Any, ...], Mapping[Any, Any]]`
:return: Tuple of types and aligned arguments.
:rtype: `Tuple[Tuple[str, ...], Tuple[Any, ...]]`
.. doctest::
>>> from eth_utils import get_aligned_abi_inputs
>>> abi = {
... 'constant': False,
... 'inputs': [
... {
... 'name': 'name',
... 'type': 'string'
... },
... {
... 'name': 's',
... 'type': 'uint256'
... }
... ],
... 'name': 'f',
... 'outputs': [],
... 'payable': False,
... 'stateMutability': 'nonpayable',
... 'type': 'function'
... }
>>> get_aligned_abi_inputs(abi, ('myName', 123))
(('string', 'uint256'), ('myName', 123))
"""
_raise_if_fallback_or_receive_abi(abi_element)
abi_element_inputs = cast(Sequence[ABIComponent], abi_element.get("inputs", []))
if isinstance(normalized_args, abc.Mapping):
# `args` is mapping. Align values according to abi order.
normalized_args = tuple(
normalized_args[abi["name"]] for abi in abi_element_inputs
)
return (
tuple(collapse_if_tuple(abi) for abi in abi_element_inputs),
type(normalized_args)(
_align_abi_input(abi, arg)
for abi, arg in zip(abi_element_inputs, normalized_args)
),
)
def get_abi_input_names(abi_element: ABIElement) -> list[str | None]:
"""
Return names for each input from the function or event ABI.
:param abi_element: ABI element.
:type abi_element: `ABIElement`
:return: Names for each input in the function or event ABI.
:rtype: `List[Optional[str]]`
.. doctest::
>>> from eth_utils import get_abi_input_names
>>> abi = {
... 'constant': False,
... 'inputs': [
... {
... 'name': 's',
... 'type': 'uint256'
... }
... ],
... 'name': 'f',
... 'outputs': [],
... 'payable': False,
... 'stateMutability': 'nonpayable',
... 'type': 'function'
... }
>>> get_abi_input_names(abi)
['s']
"""
_raise_if_fallback_or_receive_abi(abi_element)
return [
arg.get("name", None)
for arg in cast(Sequence[ABIComponent], abi_element.get("inputs", []))
]
def get_abi_input_types(abi_element: ABIElement) -> list[str]:
"""
Return types for each input from the function or event ABI.
:param abi_element: ABI element.
:type abi_element: `ABIElement`
:return: Types for each input in the function or event ABI.
:rtype: `List[str]`
.. doctest::
>>> from eth_utils import get_abi_input_types
>>> abi = {
... 'constant': False,
... 'inputs': [
... {
... 'name': 's',
... 'type': 'uint256'
... }
... ],
... 'name': 'f',
... 'outputs': [],
... 'payable': False,
... 'stateMutability': 'nonpayable',
... 'type': 'function'
... }
>>> get_abi_input_types(abi)
['uint256']
"""
_raise_if_fallback_or_receive_abi(abi_element)
return [
collapse_if_tuple(arg)
for arg in cast(Sequence[ABIComponent], abi_element.get("inputs", []))
]
def get_abi_output_names(abi_element: ABIElement) -> list[str | None]:
"""
Return names for each output from the ABI element.
:param abi_element: ABI element.
:type abi_element: `ABIElement`
:return: Names for each function output in the function ABI.
:rtype: `List[Optional[str]]`
.. doctest::
>>> from eth_utils import get_abi_output_names
>>> abi = {
... 'constant': False,
... 'inputs': [
... {
... 'name': 's',
... 'type': 'uint256'
... }
... ],
... 'name': 'f',
... 'outputs': [
... {
... 'name': 'name',
... 'type': 'string'
... },
... {
... 'name': 's',
... 'type': 'uint256'
... }
... ],
... 'payable': False,
... 'stateMutability': 'nonpayable',
... 'type': 'function'
... }
>>> get_abi_output_names(abi)
['name', 's']
"""
_raise_if_not_function_abi(abi_element)
return [
arg.get("name", None)
for arg in cast(Sequence[ABIComponent], abi_element.get("outputs", []))
]
def get_abi_output_types(abi_element: ABIElement) -> list[str]:
"""
Return types for each output from the function ABI.
:param abi_element: ABI element.
:type abi_element: `ABIElement`
:return: Types for each function output in the function ABI.
:rtype: `List[str]`
.. doctest::
>>> from eth_utils import get_abi_output_types
>>> abi = {
... 'constant': False,
... 'inputs': [
... {
... 'name': 's',
... 'type': 'uint256'
... }
... ],
... 'name': 'f',
... 'outputs': [
... {
... 'name': 'name',
... 'type': 'string'
... },
... {
... 'name': 's',
... 'type': 'uint256'
... }
... ],
... 'payable': False,
... 'stateMutability': 'nonpayable',
... 'type': 'function'
... }
>>> get_abi_output_types(abi)
['string', 'uint256']
"""
_raise_if_not_function_abi(abi_element)
return [
collapse_if_tuple(arg)
for arg in cast(Sequence[ABIComponent], abi_element.get("outputs", []))
]
def function_signature_to_4byte_selector(function_signature: str) -> bytes:
r"""
Return the 4-byte function selector from a function signature string.
:param function_signature: String representation of the function name and arguments.
:type function_signature: `str`
:return: 4-byte function selector.
:rtype: `bytes`
.. doctest::
>>> from eth_utils import function_signature_to_4byte_selector
>>> function_signature_to_4byte_selector('myFunction()')
b'\xc3x\n:'
"""
return keccak(text=function_signature.replace(" ", ""))[:4]
def function_abi_to_4byte_selector(abi_element: ABIElement) -> bytes:
r"""
Return the 4-byte function signature of the provided function ABI.
:param abi_element: ABI element.
:type abi_element: `ABIElement`
:return: 4-byte function signature.
:rtype: `bytes`
.. doctest::
>>> from eth_utils import function_abi_to_4byte_selector
>>> abi_element = {
... 'type': 'function',
... 'name': 'myFunction',
... 'inputs': [],
... 'outputs': []
... }
>>> function_abi_to_4byte_selector(abi_element)
b'\xc3x\n:'
"""
function_signature = abi_to_signature(abi_element)
return function_signature_to_4byte_selector(function_signature)
def event_signature_to_log_topic(event_signature: str) -> bytes:
r"""
Return the 32-byte keccak signature of the log topic for an event signature.
:param event_signature: String representation of the event name and arguments.
:type event_signature: `str`
:return: Log topic bytes.
:rtype: `bytes`
.. doctest::
>>> from eth_utils import event_signature_to_log_topic
>>> event_signature_to_log_topic('MyEvent()')
b'M\xbf\xb6\x8bC\xdd\xdf\xa1+Q\xeb\xe9\x9a\xb8\xfd\xedb\x0f\x9a\n\xc21B\x87\x9aO\x19*\x1byR\xd2'
"""
return keccak(text=event_signature.replace(" ", ""))
def event_abi_to_log_topic(event_abi: ABIEvent) -> bytes:
r"""
Return the 32-byte keccak signature of the log topic from an event ABI.
:param event_abi: Event ABI.
:type event_abi: `ABIEvent`
:return: Log topic bytes.
:rtype: `bytes`
.. doctest::
>>> from eth_utils import event_abi_to_log_topic
>>> abi = {
... 'type': 'event',
... 'anonymous': False,
... 'name': 'MyEvent',
... 'inputs': []
... }
>>> event_abi_to_log_topic(abi)
b'M\xbf\xb6\x8bC\xdd\xdf\xa1+Q\xeb\xe9\x9a\xb8\xfd\xedb\x0f\x9a\n\xc21B\x87\x9aO\x19*\x1byR\xd2'
"""
event_signature = abi_to_signature(event_abi)
return event_signature_to_log_topic(event_signature)
SHA-256: 1665b6edb5aa56ae4982a2327ad684d9818f35a3e10782bdf2f013ef9173d812