Skip to content

FuncSignature

eth_rpc.FuncSignature #

Bases: Request, Generic[T, U]

alias class-attribute instance-attribute #

alias = None

name instance-attribute #

name

model_post_init #

model_post_init(__context)
Source code in eth_rpc/_request.py
def model_post_init(self, __context):
    network = self.__class__._network
    object.__setattr__(self, "_network", network)
    # overwrite the .rpc() classmethod
    object.__setattr__(self, "rpc", self._rpc)

rpc classmethod #

rpc()

This uses the default network, unless a network has been provided

Source code in eth_rpc/_request.py
@classmethod
def rpc(cls) -> "RPC":
    """
    This uses the default network, unless a network has been provided
    """
    from ._transport import _force_get_global_rpc

    if cls._network is None:
        return _force_get_global_rpc()
    response = _force_get_global_rpc(cls._network)
    return response

get_identifier #

get_identifier()

This works most of the time

Source code in eth_rpc/contract/func_signature.py
def get_identifier(self):
    """This works most of the time"""
    signature = f'{self.name}({",".join(self.get_inputs())})'
    return f"0x{keccak_256(signature.encode('utf-8')).hex()[:8]}"

get_inputs #

get_inputs()
Source code in eth_rpc/contract/func_signature.py
def get_inputs(self):
    from ..types import Struct

    inputs, _ = self.__pydantic_generic_metadata__["args"]
    if inputs is NoArgs:
        return []
    if (
        type(inputs) is not GenericAlias
        and isclass(inputs)
        and issubclass(inputs, BaseModel)
    ):
        converted_inputs = transform_primitive(inputs)
        if issubclass(inputs, Struct):
            return [converted_inputs]
    else:
        if get_origin(inputs) == list:
            return [transform_primitive(inputs)]
        elif get_origin(inputs) == tuple:
            return tuple(transform_primitive(input) for input in get_args(inputs))
        else:
            converted_inputs = transform_primitive(inputs)
    if not isinstance(converted_inputs, list):
        return [converted_inputs]
    return converted_inputs

get_output #

get_output()
Source code in eth_rpc/contract/func_signature.py
def get_output(self):
    outputs = self._output
    if outputs is type(None):
        return None

    if is_annotation(outputs):
        outputs = get_args(outputs)[0]

    if (
        isclass(outputs)
        and not isinstance(outputs, GenericAlias)
        and issubclass(outputs, BaseModel)
    ):
        converted_outputs = transform_primitive(outputs)
    else:
        converted_outputs = transform_primitive(outputs)
    return converted_outputs

get_output_name #

get_output_name()
Source code in eth_rpc/contract/func_signature.py
def get_output_name(self):
    return [self._get_name(output) for output in get_args(self._output)]

encode_call #

encode_call(*, inputs)
Source code in eth_rpc/contract/func_signature.py
def encode_call(self, *, inputs: T) -> HexStr:
    from ..types import Struct

    identifier = self.get_identifier()
    # TODO: this is hard
    if inputs == ():
        return identifier

    if isinstance(inputs, BaseModel):
        if isinstance(inputs, Struct):
            input_data = inputs.to_bytes().hex()
        else:
            input_data = encode(
                self.get_inputs(),
                list(inputs.model_dump().values()),
            ).hex()
    elif isinstance(inputs, tuple):
        input_data = encode(
            self.get_inputs(), [self._encode(val) for val in inputs]
        ).hex()
    elif isinstance(inputs, list):
        input_data = encode(
            self.get_inputs(), [[self._encode(val) for val in inputs]]
        ).hex()
    else:
        input_data = encode(self.get_inputs(), [inputs]).hex()
    return HexStr(f"{identifier}{input_data}")

decode_result #

decode_result(result)
Source code in eth_rpc/contract/func_signature.py
def decode_result(self, result: HexStr) -> U:
    if isclass(self._output) and issubclass(self._output, Struct):
        return self._output.from_bytes(result)

    output = self.get_output()
    if output is None:
        # return None if the expected return type is None
        return output  # type: ignore

    if not isinstance(output, list):
        output = [output]
        decoded_output = decode(output, bytes.fromhex(result.removeprefix("0x")))[0]

        if get_origin(self._output) == list:
            output_list_type = get_args(self._output)[0]
            if isclass(output_list_type) and issubclass(output_list_type, Struct):
                decoded_arr: U = cast(
                    U,
                    [output_list_type.from_tuple(item) for item in decoded_output],
                )
                return decoded_arr
    else:
        if isclass(self._output) and issubclass(self._output, Struct):
            return self._output.from_bytes(bytes.fromhex(result.removeprefix("0x")))
        else:
            decoded_output = decode(
                output, bytes.fromhex(result.removeprefix("0x"))
            )

    # NOTE: https://github.com/pydantic/pydantic/discussions/5970
    # TODO: this is discussed to see if its a bug or not.  Annotations are a class but can't be checked as a subclass
    if (
        isclass(self._output)
        and not isinstance(self._output, GenericAlias)
        and issubclass(self._output, BaseModel)
    ):
        decoded = decoded_output
        output = self._output
        response = {}
        for (name, field), value in zip(output.model_fields.items(), decoded):
            response[name] = Struct.cast(field.annotation, value)
        return output(**response)
    return decoded_output