Bases: Request
, Generic[T, U]
alias class-attribute
instance-attribute
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
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
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]}"
|
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
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
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
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
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
|