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,
),
)
|
address instance-attribute
code_override class-attribute
instance-attribute
code_override = Field(default=None)
functions class-attribute
instance-attribute
functions = Field(default_factory=list)
model_config class-attribute
instance-attribute
model_config = ConfigDict(extra='allow')
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
|
add_func
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
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
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
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))
|