Bases: ProtocolBase
This is how we make the ProtocolBase inject the ContractFunc via the type signature. After a lot of research, this goes slightly outside of the bounds for type hinting, but I think the improved expressiveness makes it fully worthwhile.
Source code in eth_rpc/contract/base.py
| def __init__(self, **kwargs):
"""
This is how we make the ProtocolBase inject the ContractFunc via the type signature.
After a lot of research, this goes slightly outside of the bounds for type hinting,
but I think the improved expressiveness makes it fully worthwhile.
"""
super().__init__(**kwargs)
for alias, func in self._func_sigs.items():
name = alias
if is_annotation(func):
annotation_args = get_args(func)
args = annotation_args[0]
for annotation in annotation_args:
if isinstance(annotation, Name):
name = annotation.value
else:
args = func
T, U = get_args(args)
setattr(
self,
alias,
ContractFunc[T, U]( # type: ignore
func=FuncSignature[T, U](name=name, alias=alias), # type: ignore
contract=self,
),
)
|
class-attribute instance-attribute
code_override = Field(default=None)
class-attribute instance-attribute
functions = Field(default_factory=list)
class-attribute instance-attribute
model_config = ConfigDict(extra='allow')
class-attribute instance-attribute
class-attribute instance-attribute
block_and_aggregate = METHOD
class-attribute instance-attribute
try_block_and_aggregate = METHOD
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)
|
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
|
Source code in eth_rpc/contract/contract.py
| def add_func(self, func: "FuncSignature"):
if func not in self.functions:
self.functions.append(ContractFunc(func=func, contract=self))
|
get_storage_at(*, slot, block_number='latest', sync=False)
Source code in eth_rpc/contract/contract.py
| def get_storage_at(
self, *, slot: int | HexStr, block_number="latest", sync: bool = False
) -> MaybeAwaitable[HexStr]:
return run(
self._get_storage_at,
slot=slot,
block_number=block_number,
sync=sync,
)
|
get_code(*, block_number=None, block_hash=None, sync=False)
Source code in eth_rpc/contract/contract.py
| def get_code(
self,
*,
block_number: int | BLOCK_STRINGS | None = None,
block_hash: HexStr | None = None,
sync: bool = False,
) -> MaybeAwaitable[HexStr]:
return run(
self._get_code,
block_number=block_number,
block_hash=block_hash,
sync=sync,
)
|
create2(salt, keccak_init_code)
EIP-104 https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1014.md
Source code in eth_rpc/contract/contract.py
| def create2(self, salt: bytes, keccak_init_code: bytes) -> HexAddress:
"""
EIP-104
https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1014.md
"""
pre = "0xff"
b_pre = bytes.fromhex(pre[2:])
b_address = bytes.fromhex(self.address[2:])
b_result = keccak_256(b_pre + b_address + salt + keccak_init_code)
result_address = "0x" + b_result[12:].hex()
return HexAddress(HexStr(result_address))
|
execute(*calls, block_number='latest', sync=False)
Source code in eth_typeshed/multicall.py
| def execute(
self,
*calls: ContractFunc,
block_number: int | BLOCK_STRINGS = "latest",
sync: bool = False,
) -> MaybeAwaitable[list[Any]]:
return run(self._execute, *calls, block_number=block_number, sync=sync)
|
try_execute(
*calls,
require_success=False,
sync=False,
block_number="latest"
)
Source code in eth_typeshed/multicall.py
| def try_execute(
self,
*calls: ContractFunc,
require_success: bool = False,
sync: bool = False,
block_number: int | BLOCK_STRINGS = "latest",
) -> MaybeAwaitable[list[TryResult]]:
return run(
self._try_execute,
*calls,
require_success=require_success,
sync=sync,
block_number=block_number,
)
|