Skip to content

Wallet

eth_rpc.PrivateKeyWallet #

Bases: BaseWallet

private_key instance-attribute #

private_key

model_config class-attribute instance-attribute #

model_config = ConfigDict(arbitrary_types_allowed=True)

address property #

address

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_nonce #

get_nonce(block_number='latest')
Source code in eth_rpc/wallet.py
def get_nonce(self, block_number: int | BLOCK_STRINGS = "latest"):
    return RPCResponseModel(
        self.rpc().get_tx_count,
        GetAccountArgs(
            address=self.address,
            block_number=(
                HexInteger(block_number)
                if isinstance(block_number, int)
                else block_number
            ),
        ),
    )

model_post_init #

model_post_init(__context)
Source code in eth_rpc/wallet.py
def model_post_init(self, __context: Any) -> None:
    self._account = EthAccount.from_key(self.private_key)
    return super().model_post_init(__context)

get_pvt_key staticmethod #

get_pvt_key()
Source code in eth_rpc/wallet.py
@staticmethod
def get_pvt_key() -> HexStr:
    priv = secrets.token_hex(32)
    return HexStr(f"0x{priv}")

create_new classmethod #

create_new()
Source code in eth_rpc/wallet.py
@classmethod
def create_new(cls):
    return cls(private_key=cls.get_pvt_key())

sign_transaction #

sign_transaction(tx)
Source code in eth_rpc/wallet.py
def sign_transaction(self, tx: PreparedTransaction) -> SignedTransaction:
    signed_tx = self._account.sign_transaction(tx.model_dump())
    return SignedTransaction(
        raw_transaction=signed_tx[0].hex(),
        hash=signed_tx.hash.hex(),
        r=signed_tx.r,
        s=signed_tx.s,
        v=signed_tx.v,
    )

send_raw_transaction #

send_raw_transaction(tx)
Source code in eth_rpc/wallet.py
def send_raw_transaction(
    self, tx: HexStr
) -> RPCResponseModel[RawTransaction, HexStr]:
    return RPCResponseModel(
        self.rpc().send_raw_tx,
        RawTransaction(
            signed_tx=tx,
        ),
    )

prepare_and_sign #

prepare_and_sign(
    *,
    to,
    value=0,
    max_fee_per_gas=None,
    max_priority_fee_per_gas=None,
    data=HexStr("0x"),
    nonce=None
)
Source code in eth_rpc/wallet.py
def prepare_and_sign(
    self,
    *,
    to: HexAddress,
    value: int = 0,
    max_fee_per_gas: Optional[int] = None,
    max_priority_fee_per_gas: Optional[int] = None,
    data: HexStr = HexStr("0x"),
    nonce: Optional[int] = None,
):
    prepared = self.prepare(
        to=to,
        value=value,
        max_fee_per_gas=max_fee_per_gas,
        max_priority_fee_per_gas=max_priority_fee_per_gas,
        data=data,
        nonce=nonce,
    )
    return self.sign_transaction(prepared)

prepare #

prepare(
    *,
    to,
    value=0,
    max_fee_per_gas=None,
    max_priority_fee_per_gas=None,
    data=HexStr("0x"),
    nonce=None
)
Source code in eth_rpc/wallet.py
def prepare(
    self,
    *,
    to: HexAddress,
    value: int = 0,
    max_fee_per_gas: Optional[int] = None,
    max_priority_fee_per_gas: Optional[int] = None,
    data: HexStr = HexStr("0x"),
    nonce: Optional[int] = None,
):
    # TODO: this assumes sync
    gas = self.estimate_gas(to=to, data=data).sync
    access_list = None
    rpc = _force_get_global_rpc()
    chain_id = rpc.chain_id.sync()

    max_priority_fee_per_gas = max_priority_fee_per_gas or Block.priority_fee().sync
    base_fee_per_gas = Block.pending().sync.base_fee_per_gas
    assert base_fee_per_gas, "block is earlier than London Hard Fork"
    max_fee_per_gas = max_fee_per_gas or (
        2 * base_fee_per_gas + max_priority_fee_per_gas
    )

    return PreparedTransaction(
        data=data,
        to=to,
        gas=HexInteger(gas),
        max_fee_per_gas=max_fee_per_gas,
        max_priority_fee_per_gas=max_priority_fee_per_gas,
        nonce=nonce or self.get_nonce().sync,
        value=value,
        access_list=access_list,
        chain_id=chain_id,
    )

estimate_gas #

estimate_gas(to, block_number='latest', data=HexStr('0x'))
Source code in eth_rpc/wallet.py
def estimate_gas(
    self,
    to: HexAddress,
    block_number: HexInteger | Literal["latest", "pending"] = "latest",
    data: HexStr = HexStr("0x"),
) -> RPCResponseModel[CallWithBlockArgs, HexInteger]:
    return RPCResponseModel(
        self.rpc().estimate_gas,
        CallWithBlockArgs(
            params=EthCallParams(
                from_=self.address,
                to=to,
                data=data,
            ),
            block_number=block_number,
        ),
    )

transfer #

transfer(to, value)
Source code in eth_rpc/wallet.py
def transfer(
    self, to: HexAddress, value: int
) -> RPCResponseModel[RawTransaction, HexStr]:
    prepared_tx = self.prepare(to=to, value=value)
    signed_tx = self.sign_transaction(prepared_tx)
    return self.send_raw_transaction(HexStr("0x" + signed_tx.raw_transaction))

sign_hash #

sign_hash(hashed)
Source code in eth_rpc/wallet.py
def sign_hash(self, hashed: bytes) -> SignedMessage:
    return EthAccount._sign_hash(hashed, self._account.key)

balance async #

balance(block_number='latest')
Source code in eth_rpc/wallet.py
async def balance(self, block_number: int | BLOCK_STRINGS = "latest") -> int:
    return await Account.get_balance(self.address, block_number=block_number)

rsv_to_signature staticmethod #

rsv_to_signature(r, s, v)
Source code in eth_rpc/wallet.py
@staticmethod
def rsv_to_signature(r: int, s: int, v: int) -> HexStr:
    rr = hex(r)[2:].zfill(64)
    ss = hex(s)[2:].zfill(64)
    vv = hex(v)[2:]
    return HexStr("0x" + rr + ss + vv)

signature_to_rsv staticmethod #

signature_to_rsv(signature)
Source code in eth_rpc/wallet.py
@staticmethod
def signature_to_rsv(signature: HexStr) -> tuple[int, int, int]:
    v = signature[-2:]
    s = signature[-66:-2]
    r = signature[:-66].lstrip("0x")

    return int(r, 16), int(s, 16), int(v, 16)